Fabric User data functions programming model overview

The Fabric User data functions programming model is an SDK that provides the necessary functionality to author and publish runnable functions in Fabric. The SDK also allows you to seamlessly integrate with other items in the Fabric ecosystem, such as Fabric data sources. This library is publicly available in PyPI and is pre-installed in your user data functions items.

User data functions SDK

A user data functions item contains one or many functions you can invoke from the Fabric portal, from another Fabric item, or from an external application using the provided REST endpoint. Each function is a method in your Python script that allows passing parameters and returning an output to the invoker. The User data functions programming model contains the following components:

• The fabric.functions library provides the code you need to create user data functions in Python. You can see this library being imported in your first function template when you create a new user data functions item.

• The method fn.UserDataFunctions() provides the execution context found at the beginning of the code file in all new user data functions items, before any function definitions.

  Example:

  import datetime
  import fabric.functions as fn
  import logging
  
  udf = fn.UserDataFunctions()
  

• Every function is identified with a @udf.function() decorator. This decorator defines if your function can be invoked individually from the portal or an external invoker. Using this decorator will also require the function to have a return value. Functions with this decorator can access the connection objects denoted by the @udf.connection decorator.

  Invokable function example

  # 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()}!"
  

• Any Python methods without the @udf.function() decorator can't be invoked directly. They can only be invoked from functions that contain the decorator, and can be used as helper functions.

  Helper function example

  # 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()
  

Supported input types

You can define input parameters for the function such as primitive data types like str, int, float, etc. The supported input data types are:

Example of request body for input types supported:

{
  "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}
  ]
}

Supported output types

The supported output data types are:

How to write an async function

Add async decorator with your function definition in your code. With an async function you can improve responsiveness and efficiency of your application by handling multiple tasks at once. They are ideal for managing high volumes of I/O-bound operations. This example function reads a CSV file from a lakehouse using pandas. Function takes file name as an input parameter.

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}"

Data connections to Fabric data sources

This module allows you to reference the data connections without the need for writing connection strings in your code. The fabric.functions library provides two ways to handle data connections:

• fabric.functions.FabricSqlConnection: Allows you to work with SQL databases in Fabric, including SQL Analytics endpoints and Fabric warehouses.
• fabric.functions.FabricLakehouseClient: Lets you work with Lakehouses, with a way to connect to both Lakehouse tables and Lakehouse files.

To reference a connection to a data source, you need to use the @udf.connection decorator. You can apply it in any of the following formats:

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

The arguments for @udf.connection are:

• argName, the name of the variable the connection uses in your function.
• alias, the alias of the connection you added with the Manage connections menu.
• If argName and alias have the same value, you can use @udf.connection("<alias and argName for the data connection>").

Example

# 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

Generic connections for Fabric items or Azure resources

Generic connections allow you to create connections to Fabric items or Azure resources using your User Data Functions item owner identity. This feature generates an Entra ID token with the item owner's identity and a provided audience type. This token is used to authenticate with Fabric items or Azure resources that support that audience type. This process will give you a similar programming experience to using managed connections objects from the Manage Connections feature but only for the provided audience type in the connection.

This feature uses the @udf.generic_connection() decorator with the following parameters:

Connect to Fabric Cosmos DB container using a generic connection

Generic connections support native Fabric Cosmos DB items by using the CosmosDB audience type. The included User Data Functions SDK provides a helper method called get_cosmos_client that fetches a singleton Cosmos DB client for every invocation.

You can connect to a Fabric Cosmos DB item using a generic connection by following these steps:

1. In your Fabric User Data Functions item, install the azure-cosmos library using the Library Management experience.

2. Go to your Fabric Cosmos DB item settings.

   

3. Retrieve your Fabric Cosmos DB endpoint URL.

   

4. Go to your User Data Functions item. Use the following sample code to connect to your Fabric Cosmos DB container and run a read query using the Cosmos DB sample dataset. Replace the values of the following variables:

   • COSMOS_DB_URI with your Fabric Cosmos DB endpoint.
   • DB_NAME with the name of your Fabric Cosmos DB item.

   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)
   

5. Test or run this function by providing a category name, such as Accessory in the invocation parameters.

Connect to Azure Key Vault using a generic connection

Generic connections support connecting to an Azure Key Vault by using the KeyVault audience type. This type of connection requires that the Fabric User Data Functions owner has permissions to connect to the Azure Key Vault. You can use this connection to retrieve keys, secrets, or certificates by name.

You can connect to Azure Key Vault to retrieve a client secret to call an API using a generic connection by following these steps:

1. In your Fabric User Data Functions item, install the requests and the azure-keyvault-secrets libraries using the Library Management experience.

2. Go to your Azure Key Vault resource and retrieve the Vault URI and the name of your key, secret or certificate.

   

3. Go back to your Fabric User Data Functions item and use this sample. In this sample, we will retrieve a secret from Azure Key Vault to connect to a public API. Replace the value of the following variables:

   • KEY_VAULT_URL with the Vault URI you retrieved in the previous step.
   • KEY_VAULT_SECRET_NAME with the name of your secret.
   • API_URL variable with the URL of the API you'd like to connect to. This sample assumes that you are connecting to a public API that accepts GET requests and takes the following parameters api-key and request-body.

   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}"
   

4. Test or run this function by providing a request body in your code.

Get invocation properties using UserDataFunctionContext

The programming model also includes the UserDataFunctionContext object. This object contains the function invocation metadata and can be used to create specific app logic for certain invocation mechanisms.

The following table shows the properties for the UserDataFunctionContext object:

The ExecutingUser object contains the following information:

To access the UserDataFunctionContext parameter, you must use the following decorator at the top of the function definition: @udf.context(argName="<parameter name>")

Example

@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}"

Throw a handled error with UserThrownError

When developing your function, you can throw an expected error response by using the UserThrownError method available in the Python programming model. One use of this method is managing cases where the user-provided inputs fail to pass business validation rules.

Example

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()}!"

This UserThrownError method takes two parameters:

• Message: This string is returned as the error message to the application that is invoking this function.
• A dictionary of properties is returned to the application that is invoking this function.

Next steps

• Reference API documentation
• Create a Fabric User data functions item
• User data functions samples