Fabric ユーザーデータ関数プログラミングモデルの概要

Fabric ユーザーデータ関数プログラミングモデルは、Fabric で実行可能な関数を作成して発行するために必要な機能を提供する SDK です。 SDK を使用すると、Fabric データソースなど、Fabric エコシステム内の他の項目とシームレスに統合することもできます。このライブラリは PyPI で一般に公開されており、ユーザーデータ関数項目にプレインストールされています。

ユーザーデータ機能 SDK

ユーザーデータ関数項目には、Fabric ポータル、別の Fabric 項目、または指定された REST エンドポイントを使用して外部アプリケーションから呼び出すことができる 1 つまたは複数の関数が含まれています。各関数は Python スクリプト内のメソッドであり、パラメーターを渡して呼び出し側に出力を返すことができます。ユーザーデータ関数プログラミングモデルには、次のコンポーネントが含まれています。

fabric.functions ライブラリには、Python でユーザーデータ関数を作成するために必要なコードが用意されています。新しいユーザーデータ関数項目を作成すると、このライブラリが最初の関数テンプレートにインポートされていることがわかります。
fn.UserDataFunctions()メソッドは、すべての新しいユーザーデータ関数項目のコードファイルの先頭にある実行コンテキストを、関数定義の前に提供します。

例:
```
import datetime
import fabric.functions as fn
import logging

udf = fn.UserDataFunctions()
```
すべての関数は、@udf.function() デコレーターで識別されます。このデコレーターは、ポータルまたは外部呼び出し側から関数を個別に呼び出すことができるかどうかを定義します。このデコレーターを使用するためには、関数が戻り値を持っている必要があります。このデコレーターを持つ関数は、 @udf.connection デコレーターによって示される接続オブジェクトにアクセスできます。

呼び出し可能な関数の例
```
# This is a hello fabric function sample that can be invoked from the Fabric portal, another Fabric item, or an external application.

@udf.function()
def hello_fabric(name: str) -> str:
    logging.info('Python UDF trigger function processed a request.')
    logging.info('Executing hello fabric function.')

    return f"Welcome to Fabric Functions, {name}, at {datetime.datetime.now()}!"
```
@udf.function() デコレーターのない Python メソッドは、直接呼び出すことはできません。これらは、デコレーターを含む関数からのみ呼び出すことができ、ヘルパー関数として使用できます。

ヘルパー関数の例
```
# This is a helper function that can be invoked from other functions, but can't be invoked or run directly because it doesn't have the @udf.function() decorator

def uppercase_name(name: str) -> str:
    return name.upper()
```

サポートされている入力の種類

str、int、float のようなプリミティブデータ型などの関数の入力パラメーターを定義できます。サポートされている入力データ型は次のとおりです。

JSON 型	Python データ型
ストリング	str
日付時刻文字列	datetime
ブーリアン	ブール
番号	int、float
配列	list[] (例: list[int])
オブジェクト	辞書
オブジェクト	pandasデータフレーム
オブジェクトまたはオブジェクトの配列	pandas シリーズ

注

pandas DataFrame 型と Series 型を使用するには、ユーザーデータ関数に対して Fabric ポータルの ライブラリ管理 を選択し、 fabric-user-data-function バージョンを 1.0.0 に更新します。

サポートされている入力の種類の要求本文の例:

{
  "name": "Alice",                          // String (str)
  "signup_date": "2025-07-08T13:44:40Z",    // Datetime string (datetime)
  "is_active": true,                        // Boolean (bool)
  "age": 30,                                // Number (int)
  "height": 5.6,                            // Number (float)
  "favorite_numbers": [3, 7, 42],           // Array (list[int])
  "profile": {                              // Object (dict)
    "email": "alice@example.com",
    "location": "Sammamish"
  },
  "sales_data": {                           // Object (pandas DataFrame)
    "2025-07-01": {"product": "A", "units": 10},
    "2025-07-02": {"product": "B", "units": 15}
  },
  "weekly_scores": [                        // Object or Array of Objects (pandas Series)
    {"week": 1, "score": 88},
    {"week": 2, "score": 92},
    {"week": 3, "score": 85}
  ]
}

サポートされている出力の種類

サポートされている出力データ型は次のとおりです。

Python データ型
str
datetime
ブール
int、float
list[data-type] (例: list[int])
辞書
なし
pandas シリーズ
pandasデータフレーム

非同期関数を記述する方法

コードに関数定義を使用して非同期デコレーターを追加します。 async機能を使用すると、一度に複数のタスクを処理することで、アプリケーションの応答性と効率を向上させることができます。大量の I/O バインド操作を管理するのに最適です。この関数の例では、pandas を使用してレイクハウスから CSV ファイルを読み取ります。関数は、入力パラメーターとしてファイル名を受け取ります。

import pandas as pd 

# Replace the alias "<My Lakehouse alias>" with your connection alias.
@udf.connection(argName="myLakehouse", alias="<My Lakehouse alias>")
@udf.function()
async def read_csv_from_lakehouse(myLakehouse: fn.FabricLakehouseClient, csvFileName: str) -> str:

    # Connect to the Lakehouse
    connection = myLakehouse.connectToFilesAsync()   

    # Download the CSV file from the Lakehouse
    csvFile = connection.get_file_client(csvFileName)

    downloadFile = await csvFile.download_file()
    csvData = await downloadFile.readall()
    
    # Read the CSV data into a pandas DataFrame
    from io import StringIO
    df = pd.read_csv(StringIO(csvData.decode('utf-8')))

    # Display the DataFrame    
    result="" 
    for index, row in df.iterrows():
        result=result + "["+ (",".join([str(item) for item in row]))+"]"
    
    # Close the connection
    csvFile.close()
    connection.close()

    return f"CSV file read successfully.{result}"

Fabric データソースへのデータ接続

このモジュールを使用すると、コードに接続文字列を記述することなくデータ接続を参照できます。 fabric.functions ライブラリには、データ接続を処理する 2 つの方法が用意されています。

fabric.functions.FabricSqlConnection: SQL 分析エンドポイントや Fabric ウェアハウスなど、Fabric の SQL データベースを操作できます。
fabric.functions.FabricLakehouseClient: レイクハウスを使用して、レイクハウステーブルとレイクハウスファイルの両方に接続できます。

データソースへの接続を参照するには、@udf.connection デコレーターを使用する必要があります。次のいずれかの形式で適用できます。

@udf.connection(alias="<alias for data connection>", argName="sqlDB")
@udf.connection("<alias for data connection>", "<argName>")
@udf.connection("<alias for data connection>")

@udf.connection の引数は次のとおりです。

argName: 関数で接続が使用する変数の名前。
alias: [接続の管理] メニューで追加した接続のエイリアス。
argName と alias の値が同じである場合は、@udf.connection("<alias and argName for the data connection>") を使用できます。

例

# Where demosqldatabase is the argument name and the alias for my data connection used for this function
@udf.connection("demosqldatabase")
@udf.function()
def read_from_sql_db(demosqldatabase: fn.FabricSqlConnection)-> list:
  # Replace with the query you want to run
  query = "SELECT * FROM (VALUES ('John Smith', 31), ('Kayla Jones', 33)) AS Employee(EmpName, DepID);"

  # [...] Here is where the rest of your SqlConnection code would be.

  return results

Fabric 項目または Azure リソースの汎用接続

汎用接続を使用すると、ユーザーデータ関数項目の所有者 ID を使用して、Fabric 項目または Azure リソースへの接続を作成できます。この機能により、アイテム所有者の ID と提供された対象ユーザーの種類を含む Microsoft Entra ID トークンが生成されます。このトークンは、対象ユーザーの種類をサポートする Fabric アイテムまたは Azure リソースで認証するために使用されます。このプロセスにより、接続の管理機能からマネージド接続オブジェクトを使用するのと同様のプログラミングエクスペリエンスが得られますが、接続で指定された対象ユーザーの種類に対してのみ使用できます。

この機能では、 @udf.generic_connection() デコレーターと次のパラメーターを使用します。

パラメーター	説明	価値
`argName`	関数に渡される変数の名前。ユーザーは、関数の引数でこの変数を指定し、それに `fn.FabricItem` の型を使用する必要があります	たとえば、 `argName=CosmosDb`の場合、関数にはこの引数を含める必要があります。 `cosmosDb: fn.FabricItem`
`audienceType`	接続が作成される対象ユーザーの種類。このパラメーターは、Fabric 項目または Azure サービスの種類に関連付けされ、接続に使用されるクライアントを決定します。	このパラメーターに使用できる値は、 `CosmosDb` または `KeyVault`。

汎用接続を使用して Fabric Cosmos DB コンテナーに接続する

汎用接続では、 CosmosDB 対象ユーザーの種類を使用して、ネイティブの Fabric Cosmos DB 項目がサポートされます。付属の User Data Functions SDK には、呼び出しごとにシングルトン Cosmos DB クライアントをフェッチする get_cosmos_client というヘルパーメソッドが用意されています。

次の手順に従って、汎用接続を使用して Fabric Cosmos DB 項目に接続できます。

Fabric ユーザーデータ関数項目で、azure-cosmosを使用してライブラリをインストールします。
Fabric Cosmos DB 項目の設定に移動します。
Fabric Cosmos DB エンドポイント URL を取得します。

[ユーザーデータ関数] 項目に移動します。次のサンプルコードを使用して、Fabric Cosmos DB コンテナーに接続し、Cosmos DB サンプルデータセットを使用して読み取りクエリを実行します。次の変数の値を置き換えます。

COSMOS_DB_URI を Fabric Cosmos DB エンドポイントと共に使用できます。
DB_NAME に Fabric Cosmos DB アイテムの名前を設定します。

from fabric.functions.cosmosdb import get_cosmos_client
import json

@udf.generic_connection(argName="cosmosDb", audienceType="CosmosDB")
@udf.function()
def get_product_by_category(cosmosDb: fn.FabricItem, category: str) -> list:

    COSMOS_DB_URI = "YOUR_COSMOS_DB_URL"
    DB_NAME = "YOUR_COSMOS_DB_NAME" # Note: This is the Fabric item name
    CONTAINER_NAME = "SampleData" # Note: This is your container name. In this example, we are using the SampleData container.

    cosmosClient = get_cosmos_client(cosmosDb, COSMOS_DB_URI)

    # Get the database and container
    database = cosmosClient.get_database_client(DB_NAME)
    container = database.get_container_client(CONTAINER_NAME)

    query = 'select * from c WHERE c.category=@category' #"select * from c where c.category=@category"
    parameters = [
        {
            "name": "@category", "value": category
        }
    ]
    results = container.query_items(query=query, parameters=parameters)
    items = [item for item in results]

    logging.info(f"Found {len(items)} products in {category}")

    return json.dumps(items)

呼び出しパラメーターになどのカテゴリ名を指定して、Accessoryします。

注

これらの手順を使用して、アカウントの URL とデータベース名を使用して Azure Cosmos DB データベースに接続することもできます。ユーザーデータ関数の所有者アカウントには、その Azure Cosmos DB アカウントへのアクセス許可が必要です。

汎用接続を使用して Azure Key Vault に接続する

汎用接続では、 KeyVault 対象ユーザーの種類を使用した Azure Key Vault への接続がサポートされます。この種類の接続では、Fabric ユーザーデータ関数の所有者が Azure Key Vault に接続するためのアクセス許可を持っている必要があります。この接続を使用して、名前でキー、シークレット、または証明書を取得できます。

次の手順に従って、 Azure Key Vault に接続して、汎用接続を使用して API を呼び出すクライアントシークレットを取得できます。

Fabric ユーザーデータ関数項目で、requestsを使用して、azure-keyvault-secretsライブラリとライブラリをインストールします。
Azure Key Vault リソースに移動し、Vault URIとキー、シークレット、または証明書の名前を取得します。

Fabric ユーザーデータ関数項目に戻り、このサンプルを使用します。このサンプルでは、パブリック API に接続するために Azure Key Vault からシークレットを取得します。次の変数の値を置き換えます。

KEY_VAULT_URL 前のステップで取得した Vault URI を使用します。
KEY_VAULT_SECRET_NAME シークレットの名前を指定します。
API_URL は、接続先の API の URL を含む変数です。このサンプルでは、GET 要求を受け入れ、 api-key および request-body次のパラメーターを受け取るパブリック API に接続していることを前提としています。

from azure.keyvault.secrets import SecretClient
from azure.identity import DefaultAzureCredential
import requests

@udf.generic_connection(argName="keyVaultClient", audienceType="KeyVault")
@udf.function()
def retrieveNews(keyVaultClient: fn.FabricItem, requestBody:str) -> str:
    KEY_VAULT_URL = 'YOUR_KEY_VAULT_URL'
    KEY_VAULT_SECRET_NAME= 'YOUR_SECRET'
    API_URL = 'YOUR_API_URL'

    credential = keyVaultClient.get_access_token()

    client = SecretClient(vault_url=KEY_VAULT_URL, credential=credential)

    api_key = client.get_secret(KEY_VAULT_SECRET_NAME).value

    api_url = API_URL
    params = {
        "api-key": api_key,
        "request-body": requestBody
    }

    response = requests.get(api_url, params=params)

    data = "" 

    if response.status_code == 200:
        data = response.json()
    else:
        print(f"Error {response.status_code}: {response.text}")

    return f"Response: {data}"

コードに要求本文を指定して、この関数をテストまたは実行します。

`UserDataFunctionContext` を使用して呼び出しプロパティを取得する

プログラミングモデルには、UserDataFunctionContext オブジェクトも含まれています。このオブジェクトには関数呼び出しメタデータが含まれており、特定の呼び出しメカニズム用の特定のアプリロジックを作成するために使用できます。

次の表に、UserDataFunctionContext オブジェクトのプロパティを示します。

プロパティ名	データ型	説明
呼び出し ID	文字列	ユーザーデータ関数項目の呼び出しに関連付けられた一意の GUID。
実行ユーザー	オブジェクト	呼び出しを承認するために使用されるユーザーの情報のメタデータ。

ExecutingUser オブジェクトには次の情報が含まれています。

プロパティ名	データ型	説明
オイド	文字列 (GUID)	ユーザーのオブジェクト ID。要求元の不変識別子です。これは、アプリケーション間でこの関数を呼び出すために使用されるユーザーまたはサービスプリンシパルの検証済み ID です。
テナントID (TenantId)	文字列 (GUID)	ユーザーがサインインしているテナントの ID。
優先ユーザー名	文字列	ユーザーが設定した呼び出し元ユーザーの優先ユーザー名。この値は変更可能です。

UserDataFunctionContext パラメーターにアクセスするには、関数定義の先頭で次のデコレーターを使用する必要があります。@udf.context(argName="<parameter name>")

例

@udf.context(argName="myContext")
@udf.function()
def getContext(myContext: fabric.functions.UserDataFunctionContext)-> str:
    logging.info('Python UDF trigger function processed a request.')
    return f"Hello oid = {context.executing_user['Oid']}, TenantId = {context.executing_user['TenantId']}, PreferredUsername = {context.executing_user['PreferredUsername']}, InvocationId = {context.invocation_id}"

`UserThrownError` で処理されたエラーをスローする

関数を開発するときに、Python プログラミングモデルで使用できる UserThrownError メソッドを使って、予期されるエラー応答をスローできます。このメソッドの 1 つの用途は、ユーザーが指定した入力がビジネス検証規則に合格しないケースを管理することです。

例

import datetime

@udf.function()
def raise_userthrownerror(age: int)-> str:
    if age < 18:
        raise fn.UserThrownError("You must be 18 years or older to use this service.", {"age": age})

    return f"Welcome to Fabric Functions at {datetime.datetime.now()}!"

この UserThrownError メソッドは次の 2 つのパラメーターを受け取ります。

Message: この文字列は、この関数を呼び出しているアプリケーションにエラーメッセージとして返されます。
プロパティのディクショナリは、この関数を呼び出しているアプリケーションに返されます。

Fabric 変数ライブラリから変数を取得する

Microsoft Fabric の Fabric 変数ライブラリは、ワークスペース内のさまざまな項目で使用できる変数を管理するための一元化されたリポジトリです。これにより、開発者は項目の構成を効率的にカスタマイズして共有できます。

[接続の管理] を使用して変数ライブラリへの接続を追加し、変数ライブラリ項目の エイリアス を取得します。
変数ライブラリ項目のエイリアスを参照する @udf.connection(argName="varLib", alias="<My Variable Library Alias>") 、接続デコレーターを追加します。
関数定義に、 fn.FabricVariablesClient型の引数を含めます。このクライアントには、変数ライブラリ項目を操作するために必要なメソッドが用意されています。たとえば、def standardize_date(rawDate: str, varLib: fn.FabricVariablesClient) -> str: のように指定します。
getVariables()メソッドを使用して、変数ライブラリからすべての変数を取得します。

    # Get all variables from the variable library item
    variables = varLib.getVariables()

1 使用する変数の値を読み取るために、 ["variable-name"] または .get("variable-name")。

  # Get desired format from environment or use default
  date_format = variables["DATE_FORMAT"]
  # Another way to get the variable
  # date_format= variables.get("DATE_FORMAT")

フィードバック

このページはお役に立ちましたか?

Last updated on 2025-11-10

次の方法で共有

Fabric ユーザー データ関数プログラミング モデルの概要