Översikt över programmeringsmodell för användardatafunktioner inom nätverksinfrastruktur

Programmeringsmodellen för Fabric-användardatafunktioner definierar mönster och begrepp för att skapa funktioner inom Fabric.

fabric-user-data-functions SDK implementerar den här programmeringsmodellen och tillhandahåller nödvändiga funktioner för att skapa och publicera körbara funktioner. Med SDK kan du också integrera sömlöst med andra objekt i Infrastruktur-ekosystemet, till exempel Infrastrukturdatakällor. Det här biblioteket är offentligt tillgängligt i PyPI och är förinstallerat i dina användardatafunktioner.

Den här artikeln beskriver hur du använder SDK för att skapa funktioner som kan anropas från Fabric-portalen, andra infrastrukturobjekt eller externa program med hjälp av REST-API:et. Du lär dig programmeringsmodellen och viktiga begrepp med praktiska exempel.

Kom igång med SDK

Det här avsnittet beskriver kärnkomponenterna i SDK:et för användardatafunktioner och förklarar hur du strukturerar dina funktioner. Du lär dig mer om nödvändiga importer, dekoratörer och vilka typer av indata och utdata som dina funktioner kan hantera.

SDK för användardatafunktioner

fabric-user-data-functions SDK:et innehåller de kärnkomponenter som du behöver för att skapa användardatafunktioner i Python.

Nödvändig import och initiering

Varje användardatafunktionsfil måste importera modulen fabric.functions och initiera körningskontexten:

import datetime
import fabric.functions as fn
import logging

udf = fn.UserDataFunctions()

Dekoratören @udf.function()

Funktioner som är markerade med dekoratören @udf.function() kan anropas från Fabric-portalen, ett annat Fabric-objekt eller ett externt program. Funktioner med den här dekoratören måste ange en returtyp.

Exempel:

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

Hjälpfunktioner

Python-metoder utan dekoratören @udf.function() kan inte anropas direkt. De kan bara anropas från dekorerade funktioner och fungera som hjälpfunktioner.

Exempel:

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

Indatatyper som stöds

Du kan definiera indataparametrar för funktionen, till exempel primitiva datatyper som str, int, float osv. De indatatyper som stöds är:

Exempel på begärandetext för indatatyper som stöds:

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

Utdatatyper som stöds

De utdatatyper som stöds är:

Skriva funktioner

Syntaxkrav och begränsningar

När du skriver Användardatafunktioner måste du följa specifika syntaxregler för att säkerställa att dina funktioner fungerar korrekt.

Namngivning av parametrar

• Använd camelCase: Parameternamn måste använda camelCase-namngivningskonvention och får inte innehålla understreck. Använd till exempel productName i stället för product_name.
• Reserverade nyckelord: Du kan inte använda reserverade Python-nyckelord eller följande infrastrukturspecifika nyckelord som parameternamn eller funktionsnamn: req, contextoch reqInvocationId.

Parameterkrav

• Typanteckningar krävs: Alla parametrar måste innehålla typanteckningar (till exempel name: str).
• Standardvärden: Standardparametervärden stöds. Du kan definiera standardargument i Infrastruktur-användardatafunktioner för att göra koden enklare att anropa och underhålla. Standardvärden stöder vanliga JSON-serialiserbara typer, inklusive strängar, booleskt värde, tal (int, flyttal), matriser (listor) och objekt (ordlistor). Syntax

      @udf.function()
      def function_name(param1: type = value1, param2: type = value2, listparam: list | None = None, ...) -> output_type:
          # function body
  

  Observera att standardvärdet måste vara JSON-serialiserbart. Kapslade listor som [1, 2, [3]] är till exempel tillåtna, medan kapslade uppsättningar eller tupplar inte stöds. För list- eller ordlistestandarder föredrar du att använda Ingen i signaturen och tilldela den verkliga standardinställningen i funktionen.

Funktionskrav

• Returtyp krävs: Funktioner med dekoratören @udf.function() måste ange en returtypskommentar (till exempel -> str).
• Nödvändig import: Instruktionen import fabric.functions as fn och udf = fn.UserDataFunctions() initieringen krävs för att funktionerna ska fungera.

Exempel på korrekt syntax

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

Så här skriver du en asynkron funktion

Lägg till async-dekoratör i din funktionsdefinition i koden. Med en async funktion kan du förbättra svarstiden och effektiviteten i ditt program genom att hantera flera uppgifter samtidigt. De är idealiska för att hantera stora volymer av I/O-bundna åtgärder. Den här exempelfunktionen läser en CSV-fil från ett lakehouse-system med pandas. Funktionen tar filnamnet som en indataparameter.

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

Arbeta med data

Dataanslutningar till Fabric-datakällor

Med SDK kan du referera till dataanslutningar utan att behöva skriva anslutningssträngar i koden. I fabric.functions-biblioteket finns två sätt att hantera dataanslutningar:

• fabric.functions.FabricSqlConnection: Gör att du kan arbeta med SQL-databaser i Fabric, inklusive SQL Analytics-slutpunkter och Fabric-lager.
• fabric.functions.FabricLakehouseClient: Låter dig arbeta med Lakehouses, med ett sätt att ansluta till både Lakehouse-tabeller och Lakehouse-filer.

Om du vill referera till en anslutning till en datakälla måste du använda @udf.connection dekorator. Du kan använda den i något av följande format:

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

Argumenten för @udf.connection är:

• argName, namnet på variabeln som anslutningen använder i din funktion.
• alias, aliaset för anslutningen som du lade till med menyn Hantera anslutningar.
• Om argName och alias har samma värde kan du använda @udf.connection("<alias and argName for the data connection>").

Exempel

# 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

Allmänna anslutningar för nätverkskomponenter eller Azure-resurser

SDK stöder allmänna anslutningar som gör att du kan skapa anslutningar till infrastrukturobjekt eller Azure resurser med hjälp av användardatafunktionernas ägaridentitet. Den här funktionen genererar en Microsoft Entra ID token med objektägarens identitet och en angivet målgruppstyp. Den här token används för att autentisera med Infrastrukturobjekt eller Azure resurser som stöder den målgruppstypen. Den här metoden ger en liknande programmeringsupplevelse som att använda hanterade anslutningsobjekt från funktionen Hantera anslutningar , men bara för den angivna målgruppstypen i anslutningen.

Den här funktionen använder dekoratören @udf.generic_connection() med följande parametrar:

Ansluta till Cosmos DB-container med en generisk anslutning

Allmänna anslutningar stöder interna Fabric Cosmos DB-objekt med hjälp av målgruppstypen CosmosDB . Den inkluderade SDK:t för användardatafunktioner innehåller en hjälpmetod som heter get_cosmos_client som hämtar en singleton Cosmos DB-klient för varje anrop.

Du kan ansluta till ett Fabric Cosmos DB-objekt med hjälp av en allmän anslutning genom att följa dessa steg:

1. Gå till Fabric-portalen, leta upp din arbetsyta och öppna objektet användardatafunktioner. Välj Library management, sök efter biblioteket azure-cosmos och installera det. Mer information finns i Hantera bibliotek.

2. Gå till inställningarna för Fabric Cosmos DB-objekt .

   

3. Hämta Fabric Cosmos DB-slutpunktens URL.

   

4. Gå till objektet Användardatafunktioner. Använd följande exempelkod för att ansluta till din Fabric Cosmos DB-container och köra en läsfråga med hjälp av Cosmos DB-exempeldatauppsättningen. Ersätt värdena för följande variabler:

   • COSMOS_DB_URI med din Fabric Cosmos DB-slutpunkt.
   • DB_NAME med namnet på ditt Fabric Cosmos DB-objekt.

   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. Testa eller kör den här funktionen genom att ange ett kategorinamn, till exempel Accessory i anropsparametrarna.

Ansluta till Azure Key Vault med en allmän anslutning

Allmänna anslutningar stöder anslutning till en Azure Key Vault med hjälp av målgruppstypen KeyVault. Den här typen av anslutning kräver att ägaren av Användardatafunktioner för infrastrukturresurser har behörighet att ansluta till Azure Key Vault. Du kan använda den här anslutningen för att hämta nycklar, hemligheter eller certifikat efter namn.

Du kan ansluta till Azure Key Vault för att hämta en klienthemlighet för att anropa ett API med hjälp av en allmän anslutning genom att följa dessa steg:

1. Gå till Fabric-portalen, leta upp din arbetsyta och öppna objektet användardatafunktioner. Välj Library management och sök sedan efter och installera biblioteken requests och azure-keyvault-secrets. Mer information finns i Hantera bibliotek.

2. Gå till resursen Azure Key Vault i Azure portal och hämta Vault URI och namnet på din nyckel, hemlighet eller certifikat.

   

3. Gå tillbaka till ditt Fabric User Data Functions-objekt och använd det här exemplet. I det här exemplet hämtar vi en hemlighet från Azure Key Vault för att ansluta till ett offentligt API. Ersätt värdet för följande variabler:

   • KEY_VAULT_URL med det Vault URI hämtade i föregående steg.
   • KEY_VAULT_SECRET_NAME med namnet på din hemlighet.
   • API_URL variabel med URL:en för det API som du vill ansluta till. Det här exemplet förutsätter att du ansluter till ett offentligt API som accepterar GET-begäranden och tar följande parametrar api-key och 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. Testa eller kör den här funktionen genom att ange en begärandetext i koden.

Avancerade funktioner

Programmeringsmodellen definierar avancerade mönster som ger dig större kontroll över dina funktioner. SDK implementerar dessa mönster via klasser och metoder som gör att du kan:

• Få tillgång till metadata för anrop om vem som har anropat din funktion och på vilket sätt
• Hantera anpassade felscenarier med strukturerade felsvar
• Integrera med infrastrukturvariabelbibliotek för centraliserad konfigurationshantering

Hämta anropsegenskaper med hjälp av UserDataFunctionContext

SDK:et innehåller objektet UserDataFunctionContext . Det här objektet innehåller funktionsanropsmetadata och kan användas för att skapa specifik applogik för olika anropsmekanismer (till exempel portalanrop jämfört med REST API-anrop).

I följande tabell visas egenskaperna för UserDataFunctionContext-objektet:

Objektet executing_user innehåller följande information:

Om du vill access parametern UserDataFunctionContext måste du använda följande dekoratör överst i funktionsdefinitionen: @udf.context(argName="<parameter name>")

Exempel

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

Kastar ett hanterat fel med UserThrownError

När du utvecklar din funktion kan du utlösa ett förväntat felsvar med hjälp av klassen UserThrownError som är tillgänglig i SDK:et. En användning av den här klassen är att hantera fall där de användarindata som tillhandahålls misslyckas med att godkänna affärsverifieringsregler.

Exempel

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

Klasskonstruktorn UserThrownError tar två parametrar:

• Message: Den här strängen returneras som felmeddelande till programmet som anropar den här funktionen.
• En ordlista med egenskaper returneras till programmet som anropar den här funktionen.

Hämta variabler från Fabric-variabelbibliotek

Ett Fabric-variabelbibliotek i Microsoft Fabric är en centraliserad lagringsplats för att hantera variabler som kan användas i olika objekt på en arbetsyta. Det gör att utvecklare kan anpassa och dela objektkonfigurationer effektivt. Om du inte har något variabelbibliotek än kan du läsa Skapa och hantera variabelbibliotek.

Om du vill använda ett variabelbibliotek i dina funktioner lägger du till en anslutning till det från användardatafunktionsobjektet. Variabelbibliotek visas i OneLake-katalogen tillsammans med datakällor som SQL-databaser och lakehouses.

Följ dessa steg för att använda variabelbibliotek i dina funktioner:

1. Lägg till en anslutning till variabelbiblioteket i ditt användardatafunktionsobjekt. I OneLake-katalogen letar du upp och väljer ditt variabelbibliotek och väljer sedan Anslut. Observera aliaset som Fabric genererar för anslutningen.
2. Lägg till en anslutningsdekoratör för variabelbiblioteksobjektet. Till exempel @udf.connection(argName="varLib", alias="<My Variable Library Alias>") och ersätt alias till den nyligen tillagda anslutningen för variabelbiblioteksobjektet.
3. I funktionsdefinitionen inkluderar du ett argument med typen fn.FabricVariablesClient. Den här klienten innehåller metoder som du behöver för att arbeta med biblioteksobjektet variabler.
4. Använd getVariables() metoden för att hämta alla variabler från variabelbiblioteket.
5. Om du vill läsa värdena för variablerna använder du antingen ["variable-name"] eller .get("variable-name").

Exempel

I det här exemplet simulerar vi ett konfigurationsscenario för en produktions- och utvecklingsmiljö. Den här funktionen anger en storage sökväg beroende på den valda miljön med hjälp av ett värde som hämtats från variabelbiblioteket. Variabelbiblioteket innehåller en variabel med namnet ENV där användare kan ange värdet dev eller 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"

Relaterat innehåll

• Referens-API-dokumentation
• Skapa ett objekt för Fabric-användardatafunktioner
• Exempel på användardatafunktioner