หมายเหตุ
การเข้าถึงหน้านี้ต้องได้รับการอนุญาต คุณสามารถลอง ลงชื่อเข้าใช้หรือเปลี่ยนไดเรกทอรีได้
การเข้าถึงหน้านี้ต้องได้รับการอนุญาต คุณสามารถลองเปลี่ยนไดเรกทอรีได้
โมเดลการเขียนโปรแกรมฟังก์ชันข้อมูลผู้ใช้ Fabric กําหนดรูปแบบและแนวคิดสําหรับการสร้างฟังก์ชันใน Fabric
fabric-user-data-functions SDK ใช้โมเดลการเขียนโปรแกรมนี้ โดยให้ฟังก์ชันที่จําเป็นในการเขียนและเผยแพร่ฟังก์ชันที่เรียกใช้ได้ SDK ยังช่วยให้คุณสามารถรวมกับรายการอื่น ๆ ในระบบนิเวศของ Fabric ได้อย่างราบรื่น เช่น แหล่งข้อมูล Fabric
ไลบรารีนี้พร้อมใช้งานแบบสาธารณะใน PyPI และติดตั้งไว้ล่วงหน้าในรายการฟังก์ชันข้อมูลผู้ใช้ของคุณ
บทความนี้อธิบายวิธีการใช้ SDK เพื่อสร้างฟังก์ชันที่สามารถเรียกใช้ได้จากพอร์ทัล Fabric รายการ Fabric อื่นๆ หรือแอปพลิเคชันภายนอกโดยใช้ REST API คุณจะได้เรียนรู้รูปแบบการเขียนโปรแกรมและแนวคิดหลักพร้อมตัวอย่างที่ใช้งานได้จริง
เคล็ดลับ
สําหรับรายละเอียดทั้งหมดเกี่ยวกับคลาส เมธอด และพารามิเตอร์ทั้งหมด โปรดดูเอกสารอ้างอิง SDK
Getting started กับ SDK
ส่วนนี้จะแนะนําส่วนประกอบหลักของ 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()}!"
ฟังก์ชั่นตัวช่วย
ไม่สามารถเรียกใช้เมธอด Python ที่ไม่มี @udf.function() ตัวตกแต่งได้โดยตรง สามารถเรียกได้จากฟังก์ชันที่ตกแต่งและทําหน้าที่เป็นฟังก์ชันตัวช่วยเท่านั้น
ตัวอย่าง:
def uppercase_name(name: str) -> str:
return name.upper()
ชนิดการป้อนข้อมูลที่ได้รับการสนับสนุน
คุณสามารถกําหนดพารามิเตอร์การป้อนข้อมูลสําหรับฟังก์ชัน เช่น ชนิดข้อมูลแรกเริ่ม เช่น str, int, float และอื่น ๆ ชนิดข้อมูลป้อนเข้าที่ได้รับการสนับสนุนคือ:
| ประเภท JSON ของ | ชนิดข้อมูล Python |
|---|---|
| เชือก | str |
| สตริงวันที่เวลา | วันที่เวลา |
| บูลีน | บูล |
| ตัวเลข | int, ลอย |
| อาร์เรย์ | list[], ตัวอย่าง list[int] |
| วัตถุของ | เขียนตามคําบอก |
| วัตถุของ | DataFrame ของ pandas |
| วัตถุ หรือ อาร์เรย์ของวัตถุ | ซีรีส์ pandas |
หมายเหตุ
หากต้องการใช้ประเภท DataFrame และ Series ของ pandas ให้ไปที่พอร์ทัล 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 |
| วันที่เวลา |
| บูล |
| int, ลอย |
| list[data-type] ตัวอย่างเช่น list[int] |
| เขียนตามคําบอก |
| ไม่มีใคร |
| ซีรีส์ pandas |
| DataFrame ของ pandas |
ฟังก์ชันการเขียน
ข้อกําหนดและข้อจํากัดทางไวยากรณ์
เมื่อเขียนฟังก์ชันข้อมูลผู้ใช้ ให้ปฏิบัติตามกฎไวยากรณ์เฉพาะเพื่อให้แน่ใจว่าฟังก์ชันของคุณทํางานได้ถูกต้อง
การตั้งชื่อพารามิเตอร์
-
ใช้ camelCase: ชื่อพารามิเตอร์ต้องใช้แบบแผนการตั้งชื่อ camelCase และต้องไม่มีขีดล่าง ตัวอย่างเช่น ใช้
productNameแทนproduct_name -
คีย์เวิร์ดที่สงวนไว้: คุณไม่สามารถใช้คีย์เวิร์ด Python ที่สงวนไว้หรือคีย์เวิร์ดเฉพาะ Fabric ต่อไปนี้เป็นชื่อพารามิเตอร์หรือชื่อฟังก์ชัน:
req,contextและreqInvocationId
ข้อกําหนดพารามิเตอร์
ต้องมีคําอธิบายประกอบชนิด: พารามิเตอร์ทั้งหมดต้องมีคําอธิบายประกอบชนิด (เช่น
name: str)ค่าเริ่มต้น: ค่าพารามิเตอร์ดีฟอลต์ได้รับการสนับสนุน คุณสามารถกําหนดอาร์กิวเมนต์เริ่มต้นในฟังก์ชันข้อมูลผู้ใช้ Fabric เพื่อทําให้โค้ดของคุณเรียกและบํารุงรักษาได้ง่ายขึ้น พารามิเตอร์ที่มีค่าเริ่มต้นเป็นตัวเลือกในเวลาเรียกใช้จําเป็นต้องใช้พารามิเตอร์ที่ไม่มีค่าเริ่มต้น ชนิดต่อไปนี้ได้รับการสนับสนุนเป็นค่าเริ่มต้น:
ชนิดเริ่มต้น Notes String สตริงที่อนุกรมได้ JSON สตริงวันที่และเวลา ระบุเป็นสตริงในลายเซ็นฟังก์ชัน รันไทม์จะแยกวิเคราะห์สตริงในเวลา datetimeเรียกใช้ ใช้รูปแบบ ISO 8601 (เช่น2025-12-31T23:59:59Z) สําหรับการแยกวิเคราะห์ที่สอดคล้องกันและไม่คลุมเครือแบบบูลีน TrueหรือFalseเลขจำนวนเต็ม ค่าจํานวนเต็มใดๆ Float ค่าทศนิยมใด ๆ รายการ ต้องเป็นแบบอนุกรมของ JSON ชอบ Noneในลายเซ็นและกําหนดค่าเริ่มต้นที่แท้จริงภายในฟังก์ชันเพื่อหลีกเลี่ยงค่าเริ่มต้นที่เปลี่ยนแปลงได้ที่ใช้ร่วมกันพจนานุกรม ต้องเป็นแบบอนุกรมของ JSON ชอบ Noneในลายเซ็นและกําหนดค่าเริ่มต้นที่แท้จริงภายในฟังก์ชันDataFrame ของ pandas มีให้เป็นออบเจ็กต์ JSON ที่ SDK แปลงเป็นประเภทแพนด้า ต้องใช้ fabric-user-data-functionsเวอร์ชัน 1.0.0 หรือใหม่กว่าซีรีส์ pandas มีให้เป็นอาร์เรย์ JSON ของออบเจ็กต์ที่ SDK แปลงเป็นประเภทแพนด้า ต้องใช้ fabric-user-data-functionsเวอร์ชัน 1.0.0 หรือใหม่กว่าไวยากรณ์
@udf.function() def function_name( requiredParam: str, optionalStr: str = "hello", optionalDate: datetime.datetime = "2025-01-01T00:00:00Z", # specify as a string; the runtime parses it to datetime at invocation time optionalBool: bool = True, optionalInt: int = 10, optionalFloat: float = 1.5, optionalList: list | None = None, # assign real default inside the function optionalDict: dict | None = None, # assign real default inside the function ) -> dict: optionalList = optionalList or [1, 2, 3] optionalDict = optionalDict or {"key": "value"} return {"param": requiredParam}ค่าเริ่มต้นต้องเป็นแบบอนุกรมสอบ JSON ได้ (ไม่รองรับชุดและทูเปิล) สําหรับค่าเริ่มต้นของรายการหรือพจนานุกรม ให้ใช้
Noneในลายเซ็นและกําหนดค่าเริ่มต้นที่แท้จริงภายในฟังก์ชันเพื่อหลีกเลี่ยงค่าเริ่มต้นที่เปลี่ยนแปลงได้ที่ใช้ร่วมกัน ใช้รูปแบบ ISO 8601 (ตัวอย่างเช่น2025-12-31T23:59:59Z) สําหรับค่าเริ่มต้นของวันที่และเวลา การใช้แพนด้า DataFrame หรือ Series เป็นค่าเริ่มต้นต้องใช้fabric-user-data-functionsเวอร์ชัน 1.0.0 หรือใหม่กว่า
ข้อกําหนดของฟังก์ชัน
-
ต้องใช้ชนิดการส่งคืน: ฟังก์ชันที่มี
@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
เพิ่มตกแต่งแบบ async ด้วยข้อกําหนดฟังก์ชันของคุณในโค้ดของคุณ
asyncด้วยฟังก์ชันคุณสามารถปรับปรุงการตอบสนองและประสิทธิภาพของแอปพลิเคชันของคุณโดยการจัดการงานหลายอย่างพร้อมกัน เหมาะอย่างยิ่งสําหรับการจัดการการดําเนินการที่ผูกกับ I/O ในปริมาณมาก ฟังก์ชันตัวอย่างนี้อ่านไฟล์ CSV จาก lakehouse โดยใช้ pandas ฟังก์ชันจะใช้ชื่อไฟล์เป็นพารามิเตอร์การป้อนข้อมูล
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 มีสองวิธีในการจัดการการเชื่อมต่อข้อมูล:
- fabric.functions.FabricSqlConnection: ช่วยให้คุณสามารถทํางานกับฐานข้อมูล SQL ใน Fabric รวมถึงจุดสิ้นสุด SQL Analytics และคลังสินค้า Fabric
- fabric.functions.FabricLakehouseClient: ช่วยให้คุณทํางานกับ Lakehouses พร้อมวิธีการเชื่อมต่อทั้งตารางของเลคเฮ้าส์และไฟล์ของเลคเฮ้าส์
เมื่อต้องการอ้างอิงการเชื่อมต่อไปยังแหล่งข้อมูล คุณจําเป็นต้องใช้ตกแต่ง @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 รองรับการเชื่อมต่อทั่วไปที่คุณสามารถใช้สร้างการเชื่อมต่อกับรายการ Fabric หรือทรัพยากร Azure โดยใช้ข้อมูลผู้ใช้ ฟังก์ชัน ผู้ใช้ ชื่อเจ้าของรายการ คุณลักษณะนี้สร้างโทเค็น Microsoft Entra ID ที่มีข้อมูลประจําตัวของเจ้าของรายการและชนิดผู้ชมที่ระบุ ใช้โทเค็นนี้เพื่อยืนยันตัวตนกับรายการ Fabric หรือทรัพยากร Azure ที่รองรับกลุ่มเป้าหมายนั้น วิธีการนี้ให้ประสบการณ์การเขียนโปรแกรมที่คล้ายคลึงกับการใช้ออบเจ็กต์การเชื่อมต่อที่มีการจัดการจาก คุณลักษณะ จัดการการเชื่อมต่อ แต่สําหรับชนิดผู้ชมที่ระบุในการเชื่อมต่อเท่านั้น
คุณลักษณะนี้ใช้ตัว @udf.generic_connection() ตกแต่งด้วยพารามิเตอร์ต่อไปนี้:
| พารามิเตอร์ | คำอธิบาย | ค่า |
|---|---|---|
argName |
ชื่อของตัวแปรที่ส่งผ่านไปยังฟังก์ชัน ผู้ใช้จําเป็นต้องระบุตัวแปรนี้ในอาร์กิวเมนต์ของฟังก์ชันและใช้ประเภทของfn.FabricItem |
ตัวอย่างเช่น ถ้า argName=CosmosDbฟังก์ชัน ควรมีอาร์กิวเมนต์นี้ cosmosDb: fn.FabricItem |
audienceType |
ประเภทของกลุ่มเป้าหมายที่สร้างการเชื่อมต่อ พารามิเตอร์นี้เชื่อมโยงกับชนิดของรายการ Fabric หรือบริการ Azure และกําหนดไคลเอ็นต์ที่ใช้สําหรับการเชื่อมต่อ | ค่าที่อนุญาตสําหรับพารามิเตอร์CosmosDbนี้คือ หรือKeyVault |
เชื่อมต่อกับ Cosmos DB ในคอนเทนเนอร์ Fabric โดยใช้การเชื่อมต่อทั่วไป
การเชื่อมต่อทั่วไปรองรับฐานข้อมูล Cosmos DB แบบเนทีฟใน Fabric โดยใช้CosmosDBประเภทกลุ่มเป้าหมาย ฟังก์ชันข้อมูลผู้ใช้ SDK มีเมธอดช่วยเหลือที่เรียก get_cosmos_client ว่าดึงไคลเอนต์ Cosmos DB เดี่ยวสําหรับทุกการเรียกใช้
คุณสามารถเชื่อมต่อกับฐานข้อมูลฐานข้อมูล Cosmos ใน Fabric โดยใช้การเชื่อมต่อทั่วไปโดยทําตามขั้นตอนดังนี้:
ไปที่พอร์ทัล Fabric ค้นหาพื้นที่ทํางานของคุณ และเปิดรายการฟังก์ชันข้อมูลผู้ใช้ของคุณ เลือก การจัดการไลบรารี ค้นหาไลบรารี
azure-cosmosและติดตั้ง สําหรับข้อมูลเพิ่มเติม ให้ดูที่ จัดการไลบรารีไปที่ฐานข้อมูล Cosmos ของคุณในเมนู Fabric item settings
ดึง URL endpoint ของ Cosmos DB in Fabric ของคุณ
ไปที่ รายการฟังก์ชันข้อมูลผู้ใช้ของคุณ ใช้โค้ดตัวอย่างต่อไปนี้เพื่อเชื่อมต่อกับฐานข้อมูล Cosmos ของคุณในคอนเทนเนอร์ Fabric และรันคําสั่งอ่านโดยใช้ชุดข้อมูลตัวอย่างฐานข้อมูล Cosmos แทนที่ค่าของตัวแปรต่อไปนี้:
-
COSMOS_DB_URIกับ Cosmos DB ของคุณในปลายทาง Fabric -
DB_NAMEพร้อมชื่อของรายการฐานข้อมูล Cosmos ของคุณ
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ในพารามิเตอร์การเรียกใช้
หมายเหตุ
คุณยังสามารถใช้ขั้นตอนเหล่านี้เพื่อเชื่อมต่อกับฐานข้อมูล Azure Cosmos DB โดยใช้ URL บัญชีและชื่อฐานข้อมูล บัญชีเจ้าของฟังก์ชันข้อมูลผู้ใช้ต้องมีสิทธิ์เข้าถึงบัญชี Azure Cosmos DB นั้น
เชื่อมต่อกับ Azure Key Vault โดยใช้การเชื่อมต่อทั่วไป
การเชื่อมต่อทั่วไปรองรับการเชื่อมต่อกับ Azure Key Vault โดยใช้ประเภทกลุ่มเป้าหมาย KeyVault การเชื่อมต่อประเภทนี้ต้องการให้เจ้าของฟังก์ชันข้อมูลผู้ใช้ Fabric มีสิทธิ์เชื่อมต่อกับ Azure Key Vault คุณสามารถใช้การเชื่อมต่อนี้เพื่อดึงคีย์ ข้อมูลลับ หรือใบรับรองตามชื่อ
คุณสามารถเชื่อมต่อกับ Azure Key Vault เพื่อดึงข้อมูลลับของไคลเอ็นต์เพื่อเรียก API โดยใช้การเชื่อมต่อทั่วไปโดยทําตามขั้นตอนเหล่านี้:
ไปที่พอร์ทัล Fabric ค้นหาพื้นที่ทํางานของคุณ และเปิดรายการฟังก์ชันข้อมูลผู้ใช้ของคุณ เลือก Library management จากนั้นค้นหาและติดตั้งไลบรารี
requestsและazure-keyvault-secretsสําหรับข้อมูลเพิ่มเติม ให้ดูที่ จัดการไลบรารีไปที่ทรัพยากร Azure Key Vault ใน Azure portal และดึงข้อมูล
Vault URIและชื่อของคีย์ ข้อมูลลับ หรือใบรับรองของคุณ
กลับไปที่รายการฟังก์ชันข้อมูลผู้ใช้ Fabric ของคุณและใช้ตัวอย่างนี้ ในตัวอย่างนี้ คุณจะดึงความลับจาก Azure Key Vault เพื่อเชื่อมต่อกับ API สาธารณะ แทนที่ค่าของตัวแปรต่อไปนี้:
-
KEY_VAULT_URLด้วยข้อมูลที่คุณVault URIดึงมาในขั้นตอนก่อนหน้า -
KEY_VAULT_SECRET_NAMEด้วยชื่อความลับของคุณ -
API_URLที่มี URL ของ API ที่คุณต้องการเชื่อมต่อ ตัวอย่างนี้ถือว่าคุณกําลังเชื่อมต่อกับ API สาธารณะที่ยอมรับคําขอ GET และใช้พารามิเตอร์api-keyต่อไปนี้และ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}"-
ทดสอบหรือเรียกใช้ฟังก์ชันนี้ โดยระบุเนื้อหาคําขอในโค้ดของคุณ
คุณลักษณะขั้นสูง
โมเดลการเขียนโปรแกรมกําหนดรูปแบบขั้นสูงที่ให้คุณควบคุมฟังก์ชันของคุณได้ดียิ่งขึ้น SDK ใช้รูปแบบเหล่านี้ผ่านคลาสและวิธีการที่ช่วยให้คุณ:
- Access ข้อมูลเมตาของการเรียกใช้ว่าใครเรียกฟังก์ชันของคุณและวิธี
- จัดการกับสถานการณ์ข้อผิดพลาดที่กําหนดเองด้วยการตอบกลับข้อผิดพลาดที่มีโครงสร้าง
- ผสานรวมกับไลบรารีตัวแปร Fabric สําหรับการจัดการการกําหนดค่าแบบรวมศูนย์
หมายเหตุ
ฟังก์ชันข้อมูลผู้ใช้มีขีดจํากัดของขนาดคําขอ เวลาหมดอายุการทํางาน และขนาดการตอบสนอง โปรดดูรายละเอียดเกี่ยวกับขีดจํากัดเหล่านี้และวิธีการบังคับใช้ที่หัวข้อรายละเอียดและข้อจํากัดของบริการ
รับคุณสมบัติการเรียกโดยใช้ UserDataFunctionContext
SDK มีออบเจ็กต์UserDataFunctionContext ออบเจ็กต์นี้ประกอบด้วยเมตาดาต้าการเรียกใช้ฟังก์ชัน และสามารถใช้เพื่อสร้างตรรกะของแอปเฉพาะสําหรับกลไกการเรียกใช้ที่แตกต่างกัน (เช่น การเรียกใช้พอร์ทัลเทียบกับการเรียกใช้ REST API)
ตารางต่อไปนี้แสดงคุณสมบัติสําหรับวัตถุ UserDataFunctionContext:
| ชื่อคุณสมบัติ | ชนิดข้อมูล | คำอธิบาย |
|---|---|---|
| invocation_id | สตริง | GUID ที่ไม่ซ้ํากันที่เชื่อมโยงกับการเรียกใช้รายการฟังก์ชันข้อมูลผู้ใช้ |
| executing_user | ออบเจ็กต์ | เมตาดาต้าของข้อมูลของผู้ใช้ที่ใช้ในการอนุญาตการเรียก |
วัตถุ executing_user ประกอบด้วยข้อมูลต่อไปนี้:
| ชื่อคุณสมบัติ | ชนิดข้อมูล | คำอธิบาย |
|---|---|---|
| Oid | สตริง (GUID) | รหัสออบเจ็กต์ของผู้ใช้ ซึ่งเป็นตัวระบุที่ไม่สามารถเปลี่ยนแปลงได้สําหรับผู้ร้องขอ นี่คือข้อมูลประจําตัวที่ผ่านการตรวจสอบของผู้ใช้หรือบริการหลักที่ใช้ในการเรียกใช้ฟังก์ชันนี้ในแอปพลิเคชันทั้งหมด |
| รหัสผู้เช่า | สตริง (GUID) | ID ของผู้เช่าที่ผู้ใช้ลงชื่อเข้าใช้ |
| PreferredUsername | สตริง | ชื่อผู้ใช้ที่ต้องการของผู้ใช้ที่เรียก ตามที่กําหนดโดยผู้ใช้ ค่านี้มีความเสถียร |
เมื่อต้องการ access พารามิเตอร์ 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
เมื่อพัฒนาฟังก์ชัน คุณสามารถส่งการตอบสนองข้อผิดพลาดที่คาดไว้ได้โดยใช้คลาสที่มีอยู่ UserThrownError ใน SDK การใช้คลาสนี้อย่างหนึ่งคือการจัดการกรณีที่อินพุตที่ผู้ใช้ให้มาล้มเหลวในการผ่านกฎการตรวจสอบความถูกต้องทางธุรกิจ
ตัวอย่าง
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ตัวสร้างคลาสใช้พารามิเตอร์สองตัว:
-
Message: สตริงนี้จะถูกส่งกลับเป็นข้อความแสดงข้อผิดพลาดไปยังแอปพลิเคชันที่เรียกใช้ฟังก์ชันนี้ - พจนานุกรมของคุณสมบัติจะถูกส่งกลับไปยังโปรแกรมประยุกต์ที่เรียกใช้ฟังก์ชันนี้
รับตัวแปรจากไลบรารีตัวแปร Fabric
ไลบรารีตัวแปรใน Fabric เป็นคลังข้อมูลศูนย์กลางสําหรับจัดการตัวแปรที่คุณสามารถนําไปใช้กับรายการต่าง ๆ ภายในพื้นที่ทํางาน ช่วยให้นักพัฒนาสามารถปรับแต่งและแชร์การกําหนดค่ารายการได้อย่างมีประสิทธิภาพ ถ้าคุณยังไม่มีไลบรารีตัวแปร ให้ดูที่ สร้างและจัดการไลบรารีตัวแปร
เมื่อต้องการใช้ไลบรารีตัวแปรในฟังก์ชันของคุณ ให้เพิ่มการเชื่อมต่อกับไลบรารีตัวแปรจากรายการฟังก์ชันข้อมูลผู้ใช้ของคุณ ไลบรารีตัวแปรจะปรากฏในแค็ตตาล็อก OneLake ควบคู่ไปกับแหล่งข้อมูล เช่น ฐานข้อมูล SQL และเลคเฮาส์
ทําตามขั้นตอนเหล่านี้เพื่อใช้ไลบรารีตัวแปรในฟังก์ชันของคุณ:
- ในรายการฟังก์ชันข้อมูลผู้ใช้ ให้ เพิ่มการเชื่อมต่อกับ ไลบรารีตัวแปรของคุณ ในแค็ตตาล็อก OneLake ให้ค้นหาและเลือกไลบรารีตัวแปรของคุณ จากนั้นเลือก เชื่อมต่อ สังเกต นามแฝง ที่ Fabric สร้างขึ้นสําหรับการเชื่อมต่อ
- เพิ่มตัวตกแต่งการเชื่อมต่อสําหรับรายการไลบรารีตัวแปร ตัวอย่างเช่น
@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"