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

Fabric ユーザーデータ関数プログラミングモデルでは、Fabric で関数を作成するためのパターンと概念が定義されています。

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

この記事では、SDK を使用して、REST API を使用して、Fabric ポータル、他の Fabric 項目、または外部アプリケーションから呼び出すことができる関数を構築する方法について説明します。実際の例を使用して、プログラミングモデルと主要な概念について学習します。

ヒント

すべてのクラス、メソッド、パラメーターの詳細については、 SDK リファレンスドキュメントを参照してください。

SDK の使い始め

このセクションでは、User Data Functions SDK のコアコンポーネントについて説明し、関数を構築する方法について説明します。必要なインポート、デコレーター、関数で処理できる入力データと出力データの種類について説明します。

ユーザーデータ機能 SDK

fabric-user-data-functions SDK には、Python でユーザーデータ関数を作成するために必要なコアコンポーネントが用意されています。

必要なインポートと初期化

すべてのユーザーデータ関数ファイルは、 fabric.functions モジュールをインポートし、実行コンテキストを初期化する必要があります。

import datetime
import fabric.functions as fn
import logging

udf = fn.UserDataFunctions()

@udf.function() デコレーター

@udf.function()デコレーターでマークされた関数は、Fabric ポータル、別の Fabric アイテム、または外部アプリケーションから呼び出すことができます。このデコレーターを持つ関数は、戻り値の型を指定する必要があります。

例:

@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 メソッドは、直接呼び出すことはできません。これらは、装飾された関数からのみ呼び出され、ヘルパー関数として機能します。

例:

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-functions パッケージを検索し、バージョン 1.0.0 以降に更新します。

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

{
  "name": "Alice",                          // String (str)
  "signup_date": "2025-11-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-11-01": {"product": "A", "units": 10},
    "2025-11-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データフレーム

関数の記述

構文の要件と制限事項

ユーザーデータ関数を記述するときは、関数が正しく動作するように、特定の構文規則に従う必要があります。

パラメーターの名前付け

camelCase を使用する: パラメーター名は camelCase の名前付け規則を使用する必要があり、アンダースコアを含めることはできません。たとえば、productName の代わりに product_name を使用します。
予約済みキーワード: 予約済み Python キーワードまたは次の Fabric 固有のキーワードをパラメーター名または関数名として使用することはできません: req、 context、および reqInvocationId。

パラメーターの要件

既定値なし: 既定のパラメーター値はサポートされていません。関数を呼び出すときは、すべてのパラメーターが必要です。たとえば、次の関数は構文エラーをスローします。
```
# The default value for the argument called 'name' is not supported and treated like a syntax error.
@udf.function()
def goodbye_fabric(name: str = "N/A") -> str:
    return f"Goodbye, {name}."
```
型注釈が必要: すべてのパラメーターに型注釈 ( name: str など) を含める必要があります。

機能要件

戻り値の型が必須: @udf.function() デコレーターを持つ関数は、戻り値の型注釈 (たとえば、 -> str) を指定する必要があります。
必須のインポート: 関数を機能させるには、 import fabric.functions as fn ステートメントと udf = fn.UserDataFunctions() 初期化が必要です。

正しい構文の例

@udf.function()
def process_order(orderNumber: int, customerName: str, orderDate: str) -> dict:
    return {
        "order_id": orderNumber,
        "customer": customerName,
        "date": orderDate,
        "status": "processed"
    }

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

コードに関数定義を使用して非同期デコレーターを追加します。 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 データソースへのデータ接続

SDK を使用すると、コードに接続文字列を書き込む必要なく、データ接続を参照できます。 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:
  # Connect to the SQL database
  connection = demosqldatabase.connect()
  cursor = connection.cursor()
  
  # Replace with the query you want to run
  query = "SELECT * FROM (VALUES ('John Smith', 31), ('Kayla Jones', 33)) AS Employee(EmpName, DepID);"
  
  # Execute the query
  cursor.execute(query)
  
  # Fetch all results
  results = cursor.fetchall()
  
  # Close the cursor and connection
  cursor.close()
  connection.close()
  
  return results

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

SDK では、ユーザーデータ関数項目の所有者 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 portal で 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}"

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

高度な機能

プログラミングモデルでは、関数をより詳細に制御できる高度なパターンを定義します。 SDK では、次のことが可能なクラスとメソッドを使用してこれらのパターンを実装します。

関数を呼び出したユーザーとその方法に関する呼び出しメタデータにアクセスする
構造化エラー応答を使用してカスタムエラーシナリオを処理する
一元化された構成管理のために Fabric 変数ライブラリと統合する

注

ユーザーデータ関数には、要求サイズ、実行タイムアウト、応答サイズに関するサービス制限があります。これらの制限とその適用方法の詳細については、「サービスの詳細と制限事項」を参照してください。

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

SDK には、 UserDataFunctionContext オブジェクトが含まれています。このオブジェクトには関数呼び出しメタデータが含まれており、さまざまな呼び出しメカニズム (ポータル呼び出しと REST API 呼び出しなど) の特定のアプリロジックを作成するために使用できます。

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

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

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

プロパティ名	データ型	説明
オイド	文字列 (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 = {myContext.executing_user['Oid']}, TenantId = {myContext.executing_user['TenantId']}, PreferredUsername = {myContext.executing_user['PreferredUsername']}, InvocationId = {myContext.invocation_id}"

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

関数を開発するときに、SDK で使用可能な 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 変数ライブラリは、ワークスペース内のさまざまな項目で使用できる変数を管理するための一元化されたリポジトリです。これにより、開発者は項目の構成を効率的にカスタマイズして共有できます。変数ライブラリがまだない場合は、「変数ライブラリの作成と管理」を参照してください。

関数で変数ライブラリを使用するには、ユーザーデータ関数項目から変数ライブラリへの接続を追加します。変数ライブラリは、SQL データベースや Lakehouse などのデータソースと共に OneLake カタログに表示されます。

関数で変数ライブラリを使用するには、次の手順に従います。

ユーザーデータ関数項目で、変数ライブラリへの接続を追加します。 OneLake カタログで、変数ライブラリを見つけて選択し、[ 接続] を選択します。ファブリックが接続用に生成する エイリアス に注意してください。
変数ライブラリ項目の接続デコレーターを追加します。たとえば、変数ライブラリ項目の新しく追加された接続にエイリアスを @udf.connection(argName="varLib", alias="<My Variable Library Alias>") して置き換えます。
関数定義に、 fn.FabricVariablesClient型の引数を含めます。このクライアントには、変数ライブラリ項目を操作するために必要なメソッドが用意されています。
getVariables()メソッドを使用して、変数ライブラリからすべての変数を取得します。
変数の値を読み取る場合は、 ["variable-name"] または .get("variable-name")を使用します。

例

この例では、運用環境と開発環境の構成シナリオをシミュレートします。この関数は、変数ライブラリから取得した値を使用して、選択した環境に応じてストレージパスを設定します。変数ライブラリには、ユーザーがENVまたはdevの値を設定できるprodと呼ばれる変数が含まれています。

@udf.connection(argName="varLib", alias="<My Variable Library Alias>")
@udf.function()
def get_storage_path(dataset: str, varLib: fn.FabricVariablesClient) -> str:
    """
    Description: Determine storage path for a dataset based on environment configuration from Variable Library.
    
    Args:
        dataset_name (str): Name of the dataset to store.
        varLib (fn.FabricVariablesClient): Fabric Variable Library connection.
    
    Returns:
        str: Full storage path for the dataset.
    """
    # Retrieve variables from Variable Library
    variables = varLib.getVariables()
    
    # Get environment and base paths
    env = variables.get("ENV")    
    dev_path = variables.get("DEV_FILE_PATH")
    prod_path = variables.get("PROD_FILE_PATH")
    
    # Apply environment-specific logic
    if env.lower() == "dev":
        return f"{dev_path}{dataset}/"
    elif env.lower() == "prod":
        return f"{prod_path}{dataset}/"
    else:
        return f"incorrect settings define for ENV variable"

フィードバック

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

Last updated on 2026-01-22