دليل مطور وظائف Azure Python
هذا الدليل هو مقدمة لتطوير Azure Functions باستخدام Python. تفترض المقالة أنك قرأت بالفعل دليل مطوري Azure Functions.
هام
تدعم هذه المقالة كلا من نموذج البرمجة v1 وv2 ل Python في Azure Functions. يستخدم نموذج Python v1 ملف functions.json لتعريف الوظائف، ويتيح لك نموذج v2 الجديد بدلا من ذلك استخدام نهج يستند إلى مصمم الديكور. ينتج عن هذا الأسلوب الجديد بنية ملف أبسط، وهو أكثر تركزا على التعليمات البرمجية. اختر محدد v2 في أعلى المقالة للتعرف على نموذج البرمجة الجديد هذا.
بصفتك مطور Python، قد تكون مهتما أيضا بهذه المواضيع:
- Visual Studio Code: إنشاء أول تطبيق Python باستخدام Visual Studio Code.
- الوحدة الطرفية أو موجه الأوامر: أنشئ أول تطبيق Python من موجه الأوامر باستخدام Azure Functions Core Tools.
- العينات: راجع بعض تطبيقات Python الموجودة في متصفح نماذج Learn.
- Visual Studio Code: إنشاء أول تطبيق Python باستخدام Visual Studio Code.
- الوحدة الطرفية أو موجه الأوامر: أنشئ أول تطبيق Python من موجه الأوامر باستخدام Azure Functions Core Tools.
- العينات: راجع بعض تطبيقات Python الموجودة في متصفح نماذج Learn.
خيارات التطوير
يدعم نموذجا برمجة وظائف Python التطوير المحلي في إحدى البيئات التالية:
نموذج برمجة Python v2:
نموذج برمجة Python v1:
يمكنك أيضا إنشاء وظائف Python v1 في مدخل Microsoft Azure.
تلميح
على الرغم من أنه يمكنك تطوير وظائف Azure المستندة إلى Python محليا على Windows، إلا أن Python مدعومة فقط على خطة استضافة مستندة إلى Linux عند تشغيلها في Azure. لمزيد من المعلومات، راجع قائمة مجموعات نظام التشغيل/وقت التشغيل المدعومة.
نموذج البرمجة
تتوقع وظائف Azure أن تكون الوظيفة طريقة عديمة الحالة في برنامج Python النصي الذي يعالج الإدخال وينتج الإخراج. بشكل افتراضي، يتوقع وقت التشغيل تنفيذ الأسلوب كأسلوب عمومي يسمى main()
في ملف __init__.py . يمكنك أيضا تحديد نقطة إدخال بديلة.
يمكنك ربط البيانات بالدالة من المشغلات والروابط عبر سمات الأسلوب التي تستخدم الخاصية name
المحددة في ملف function.json . على سبيل المثال، يصف ملف function.json التالي دالة بسيطة يتم تشغيلها بواسطة طلب HTTP يسمى req
:
{
"scriptFile": "__init__.py",
"bindings": [
{
"authLevel": "function",
"type": "httpTrigger",
"direction": "in",
"name": "req",
"methods": [
"get",
"post"
]
},
{
"type": "http",
"direction": "out",
"name": "$return"
}
]
}
استنادا إلى هذا التعريف، قد يبدو ملف __init__.py الذي يحتوي على التعليمات البرمجية للدالة مثل المثال التالي:
def main(req):
user = req.params.get('user')
return f'Hello, {user}!'
يمكنك أيضًا الإعلان صراحةً عن أنواع السمات ونوع الإرجاع في الدالة عن طريق استخدام التعليقات التوضيحية من نوع Python. يساعدك القيام بذلك على استخدام ميزات IntelliSense والإكمال التلقائي التي يوفرها العديد من محرري التعليمات البرمجية ل Python.
import azure.functions
def main(req: azure.functions.HttpRequest) -> str:
user = req.params.get('user')
return f'Hello, {user}!'
استخدم التعليقات التوضيحية Python المضمنة في حزمة azure.functions.* لربط الإدخال والمخرجات بأساليبك.
تتوقع وظائف Azure أن تكون الوظيفة طريقة عديمة الحالة في برنامج Python النصي الذي يعالج الإدخال وينتج الإخراج. بشكل افتراضي، يتوقع وقت التشغيل تنفيذ الأسلوب كأسلوب عمومي في ملف function_app.py .
يمكن الإعلان عن المشغلات والروابط واستخدامها في دالة في نهج يستند إلى مصمم الديكور. يتم تعريفها في نفس الملف، function_app.py، كوظائف. على سبيل المثال، يمثل ملف function_app.py التالي مشغل دالة بواسطة طلب HTTP.
@app.function_name(name="HttpTrigger1")
@app.route(route="req")
def main(req):
user = req.params.get("user")
return f"Hello, {user}!"
يمكنك أيضًا الإعلان صراحةً عن أنواع السمات ونوع الإرجاع في الدالة عن طريق استخدام التعليقات التوضيحية من نوع Python. يساعدك القيام بذلك على استخدام ميزات IntelliSense والإكمال التلقائي التي يوفرها العديد من محرري التعليمات البرمجية ل Python.
import azure.functions as func
app = func.FunctionApp()
@app.function_name(name="HttpTrigger1")
@app.route(route="req")
def main(req: func.HttpRequest) -> str:
user = req.params.get("user")
return f"Hello, {user}!"
للتعرف على القيود المعروفة مع نموذج v2 والحلول البديلة الخاصة بها، راجع استكشاف أخطاء Python وإصلاحها في Azure Functions.
نقطة إدخال بديلة
يمكنك تغيير السلوك الافتراضي لدالة عن طريق تحديد scriptFile
خصائص و entryPoint
اختياريا في ملف function.json . على سبيل المثال، تخبر function.json التالية وقت التشغيل باستخدام customentry()
الأسلوب في ملف main.py كنقطة إدخال لدالة Azure.
{
"scriptFile": "main.py",
"entryPoint": "customentry",
"bindings": [
...
]
}
نقطة الإدخال موجودة فقط في ملف function_app.py . ومع ذلك، يمكنك الرجوع إلى الدالات داخل المشروع في function_app.py باستخدام المخططات أو عن طريق الاستيراد.
بنية المجلد
تبدو بنية المجلد الموصى بها لمشروع وظائف Python مثل المثال التالي:
<project_root>/
| - .venv/
| - .vscode/
| - my_first_function/
| | - __init__.py
| | - function.json
| | - example.py
| - my_second_function/
| | - __init__.py
| | - function.json
| - shared_code/
| | - __init__.py
| | - my_first_helper_function.py
| | - my_second_helper_function.py
| - tests/
| | - test_my_second_function.py
| - .funcignore
| - host.json
| - local.settings.json
| - requirements.txt
| - Dockerfile
يمكن أن يحتوي مجلد المشروع الرئيسي، <project_root>، على الملفات التالية:
- local.settings.json: يستخدم لتخزين إعدادات التطبيق سلسلة الاتصال عند التشغيل محليا. لا يتم نشر هذا الملف في Azure. لمعرفة المزيد، راجع local.settings.file.
- requirements.txt: يحتوي على قائمة حزم Python التي يثبتها النظام عند النشر إلى Azure.
- host.json: يحتوي على خيارات التكوين التي تؤثر على جميع الوظائف في مثيل تطبيق الوظائف. لا يتم نشر هذا الملف في Azure. لا يتم دعم جميع الخيارات عند التشغيل محليًا. لمعرفة المزيد، راجع host.json.
- .vscode/: (اختياري) يحتوي على تكوين Visual Studio Code المخزن. لمعرفة المزيد، راجع إعدادات Visual Studio Code.
- .venv/: (اختياري) يحتوي على بيئة Python ظاهرية يستخدمها التطوير المحلي.
- Dockerfile: (اختياري) يستخدم عند نشر مشروعك في حاوية مخصصة.
- الاختبارات/: (اختياري) يحتوي على حالات الاختبار لتطبيق الوظائف.
- .funcignore: (اختياري) يعلن عن الملفات التي لا ينبغي نشرها إلى Azure. عادة ما يحتوي هذا الملف على .vscode/ لتجاهل إعداد المحرر و.venv/ لتجاهل بيئة Python الظاهرية المحلية والاختبارات/ لتجاهل حالات الاختبار local.settings.json لمنع نشر إعدادات التطبيق المحلي.
كل وظيفة لها ملف التعليمات البرمجية الخاصة بها وملف تكوين الربط، function.json.
تبدو بنية المجلد الموصى بها لمشروع وظائف Python مثل المثال التالي:
<project_root>/
| - .venv/
| - .vscode/
| - function_app.py
| - additional_functions.py
| - tests/
| | - test_my_function.py
| - .funcignore
| - host.json
| - local.settings.json
| - requirements.txt
| - Dockerfile
يمكن أن يحتوي مجلد المشروع الرئيسي، <project_root>، على الملفات التالية:
- .venv/: (اختياري) يحتوي على بيئة Python الظاهرية التي يستخدمها التطوير المحلي.
- .vscode/: (اختياري) يحتوي على تكوين Visual Studio Code المخزن. لمعرفة المزيد، راجع إعدادات Visual Studio Code.
- function_app.py: الموقع الافتراضي لكافة الدالات والمشغلات والروابط ذات الصلة.
- additional_functions.py: (اختياري) أي ملفات Python أخرى تحتوي على وظائف (عادة للتجميع المنطقي) المشار إليها في function_app.py من خلال المخططات.
- الاختبارات/: (اختياري) يحتوي على حالات الاختبار لتطبيق الوظائف.
- .funcignore: (اختياري) يعلن عن الملفات التي لا ينبغي نشرها إلى Azure. عادة ما يحتوي هذا الملف على .vscode/ لتجاهل إعداد المحرر الخاص بك، .venv/ لتجاهل بيئة Python الظاهرية المحلية، والاختبارات/ لتجاهل حالات الاختبار، local.settings.json لمنع نشر إعدادات التطبيق المحلي.
- host.json: يحتوي على خيارات التكوين التي تؤثر على جميع الوظائف في مثيل تطبيق الوظائف. لا يتم نشر هذا الملف في Azure. لا يتم دعم جميع الخيارات عند التشغيل محليًا. لمعرفة المزيد، راجع host.json.
- local.settings.json: يستخدم لتخزين إعدادات التطبيق سلسلة الاتصال عند تشغيله محليا. لا يتم نشر هذا الملف في Azure. لمعرفة المزيد، راجع local.settings.file.
- requirements.txt: يحتوي على قائمة حزم Python التي يقوم النظام بتثبيتها عند نشرها إلى Azure.
- Dockerfile: (اختياري) يستخدم عند نشر مشروعك في حاوية مخصصة.
عند نشر مشروعك إلى تطبيق دالة في Azure، يجب تضمين محتويات مجلد المشروع الرئيسي بالكامل، <project_root>، في الحزمة، ولكن ليس المجلد نفسه، ما يعني أن host.json يجب أن تكون في جذر الحزمة. نوصي بالاحتفاظ باختباراتك في مجلد مع دالات أخرى (في هذا المثال، الاختبارات/). لمزيد من المعلومات، راجع اختبار الوحدة.
اتصال بقاعدة بيانات
تتكامل Azure Functions بشكل جيد مع Azure Cosmos DB للعديد من حالات الاستخدام، بما في ذلك IoT والتجارة الإلكترونية والألعاب وما إلى ذلك.
على سبيل المثال، بالنسبة لموارد الأحداث، يتم دمج الخدمتين لتشغيل البنيات المستندة إلى الحدث باستخدام وظيفة موجز التغيير في Azure Cosmos DB. يوفر موجز التغيير الخدمات المصغرة المتلقية للمعلومات القدرة على قراءة الإدراج والتحديثات بشكل موثوق وتدريجي (على سبيل المثال، أحداث الطلب). يمكن استخدام هذه الوظيفة لتوفير مخزن أحداث مستمر كوسيط رسائل للأحداث المتغيرة للحالة وسير عمل معالجة أمر القيادة بين العديد من الخدمات المصغرة (والتي يمكن تنفيذها كوظائف Azure بلا خادم).
للاتصال ب Azure Cosmos DB، قم أولا بإنشاء حساب وقاعدة بيانات وحاوية. ثم يمكنك توصيل التعليمات البرمجية للدالة الخاصة بك ب Azure Cosmos DB باستخدام المشغل والروابط، مثل هذا المثال.
لتنفيذ منطق تطبيق أكثر تعقيدا، يمكنك أيضا استخدام مكتبة Python ل Cosmos DB. يبدو تنفيذ الإدخال/إخراج غير المتزامن كما يلي:
pip install azure-cosmos
pip install aiohttp
from azure.cosmos.aio import CosmosClient
from azure.cosmos import exceptions
from azure.cosmos.partition_key import PartitionKey
import asyncio
# Replace these values with your Cosmos DB connection information
endpoint = "https://azure-cosmos-nosql.documents.azure.com:443/"
key = "master_key"
database_id = "cosmicwerx"
container_id = "cosmicontainer"
partition_key = "/partition_key"
# Set the total throughput (RU/s) for the database and container
database_throughput = 1000
# Singleton CosmosClient instance
client = CosmosClient(endpoint, credential=key)
# Helper function to get or create database and container
async def get_or_create_container(client, database_id, container_id, partition_key):
database = await client.create_database_if_not_exists(id=database_id)
print(f'Database "{database_id}" created or retrieved successfully.')
container = await database.create_container_if_not_exists(id=container_id, partition_key=PartitionKey(path=partition_key))
print(f'Container with id "{container_id}" created')
return container
async def create_products():
container = await get_or_create_container(client, database_id, container_id, partition_key)
for i in range(10):
await container.upsert_item({
'id': f'item{i}',
'productName': 'Widget',
'productModel': f'Model {i}'
})
async def get_products():
items = []
container = await get_or_create_container(client, database_id, container_id, partition_key)
async for item in container.read_all_items():
items.append(item)
return items
async def query_products(product_name):
container = await get_or_create_container(client, database_id, container_id, partition_key)
query = f"SELECT * FROM c WHERE c.productName = '{product_name}'"
items = []
async for item in container.query_items(query=query, enable_cross_partition_query=True):
items.append(item)
return items
async def main():
await create_products()
all_products = await get_products()
print('All Products:', all_products)
queried_products = await query_products('Widget')
print('Queried Products:', queried_products)
if __name__ == "__main__":
asyncio.run(main())
Blueprints
يقدم نموذج برمجة Python v2 مفهوم المخططات. المخطط هو فئة جديدة يتم إنشاء مثيل لها لتسجيل الوظائف خارج تطبيق الدالة الأساسية. لا تتم فهرسة الوظائف المسجلة في مثيلات المخطط مباشرة بواسطة وقت تشغيل الدالة. للحصول على وظائف المخطط هذه المفهرسة، يحتاج تطبيق الدالة إلى تسجيل الوظائف من مثيلات المخطط.
يوفر استخدام المخططات المزايا التالية:
- يتيح لك تقسيم تطبيق الوظائف إلى مكونات نمطية، والتي تمكنك من تحديد الوظائف في ملفات Python متعددة وتقسيمها إلى مكونات مختلفة لكل ملف.
- يوفر واجهات تطبيق الوظائف العامة القابلة للتوسيع لإنشاء واجهات برمجة التطبيقات الخاصة بك وإعادة استخدامها.
يوضح المثال التالي كيفية استخدام المخططات:
أولا، في ملف http_blueprint.py ، يتم أولا تعريف الدالة التي يتم تشغيلها بواسطة HTTP وإضافتها إلى كائن مخطط.
import logging
import azure.functions as func
bp = func.Blueprint()
@bp.route(route="default_template")
def default_template(req: func.HttpRequest) -> func.HttpResponse:
logging.info('Python HTTP trigger function processed a request.')
name = req.params.get('name')
if not name:
try:
req_body = req.get_json()
except ValueError:
pass
else:
name = req_body.get('name')
if name:
return func.HttpResponse(
f"Hello, {name}. This HTTP-triggered function "
f"executed successfully.")
else:
return func.HttpResponse(
"This HTTP-triggered function executed successfully. "
"Pass a name in the query string or in the request body for a"
" personalized response.",
status_code=200
)
بعد ذلك، في ملف function_app.py ، يتم استيراد كائن المخطط ويتم تسجيل وظائفه في تطبيق الوظائف.
import azure.functions as func
from http_blueprint import bp
app = func.FunctionApp()
app.register_functions(bp)
إشعار
تدعم Durable Functions أيضا المخططات. لإنشاء مخططات لتطبيقات Durable Functions، قم بتسجيل مشغلات التنسيق والنشاط والكيان وروابط العميل باستخدام azure-functions-durable
Blueprint
الفئة ، كما هو موضح هنا. يمكن بعد ذلك تسجيل المخطط الناتج كالمعتاد. راجع نموذجنا للحصول على مثال.
سلوك الاستيراد
يمكنك استيراد الوحدات النمطية في تعليمة برمجية الدالة الخاص بك باستخدام كل من المراجع المطلقة والنسبية. استنادا إلى بنية المجلد الموضحة مسبقا، تعمل عمليات الاستيراد التالية من داخل ملف <الدالة project_root>\my_first_function\__init__.py:
from shared_code import my_first_helper_function #(absolute)
import shared_code.my_second_helper_function #(absolute)
from . import example #(relative)
إشعار
عند استخدام بناء جملة استيراد مطلق، يجب أن يحتوي المجلد shared_code/على ملف __init__.py لوضع علامة عليه كحزمة Python.
يتم إهمال الاستيراد __app__ التالي وما بعد الاستيراد النسبي ذي المستوى الأعلى، لأنها غير مدعومة من قبل مدقق النوع الثابت ولا تدعمها أطر عمل اختبار Python:
from __app__.shared_code import my_first_helper_function #(deprecated __app__ import)
from ..shared_code import my_first_helper_function #(deprecated beyond top-level relative import)
المشغلات والمدخلات
تنقسم المدخلات إلى فئتين في Azure Functions: تشغيل الإدخال والإدخالات الأخرى. على الرغم من اختلافها في ملف function.json ، فإن استخدامها متطابق في التعليمات البرمجية ل Python. يتم تعيين سلاسل الاتصال أو البيانات السرية لمصادر المشغل والإدخل إلى القيم في ملف local.settings.json عند تشغيلها محليا، ويتم تعيينها إلى إعدادات التطبيق عند تشغيلها في Azure.
على سبيل المثال، توضح التعليمات البرمجية التالية الفرق بين الإدخالين:
// function.json
{
"scriptFile": "__init__.py",
"bindings": [
{
"name": "req",
"direction": "in",
"type": "httpTrigger",
"authLevel": "anonymous",
"route": "items/{id}"
},
{
"name": "obj",
"direction": "in",
"type": "blob",
"path": "samples/{id}",
"connection": "STORAGE_CONNECTION_STRING"
}
]
}
// local.settings.json
{
"IsEncrypted": false,
"Values": {
"FUNCTIONS_WORKER_RUNTIME": "python",
"STORAGE_CONNECTION_STRING": "<AZURE_STORAGE_CONNECTION_STRING>",
"AzureWebJobsStorage": "<azure-storage-connection-string>"
}
}
# __init__.py
import azure.functions as func
import logging
def main(req: func.HttpRequest, obj: func.InputStream):
logging.info(f'Python HTTP-triggered function processed: {obj.read()}')
عند استدعاء الدالة، يتم تمرير طلب HTTP إلى الدالة ك req
. يتم استرداد إدخال من حساب Azure Blob Storage استنادا إلى المعرف في عنوان URL للمسار وإتاحته كما هو الحال obj
في نص الدالة. هنا، حساب التخزين المحدد هو سلسلة الاتصال الذي تم العثور عليه في CONNECTION_STRING
إعداد التطبيق.
تنقسم المدخلات إلى فئتين في Azure Functions: تشغيل الإدخال والإدخالات الأخرى. على الرغم من تعريفها باستخدام مصممات مختلفة، فإن استخدامها مشابه في تعليمة Python البرمجية. يتم تعيين سلاسل الاتصال أو البيانات السرية لمصادر المشغل والإدخل إلى القيم في ملف local.settings.json عند تشغيلها محليا، ويتم تعيينها إلى إعدادات التطبيق عند تشغيلها في Azure.
على سبيل المثال، توضح التعليمات البرمجية التالية كيفية تعريف ربط إدخال Blob Storage:
// local.settings.json
{
"IsEncrypted": false,
"Values": {
"FUNCTIONS_WORKER_RUNTIME": "python",
"STORAGE_CONNECTION_STRING": "<AZURE_STORAGE_CONNECTION_STRING>",
"AzureWebJobsStorage": "<azure-storage-connection-string>"
}
}
# function_app.py
import azure.functions as func
import logging
app = func.FunctionApp()
@app.route(route="req")
@app.read_blob(arg_name="obj", path="samples/{id}",
connection="STORAGE_CONNECTION_STRING")
def main(req: func.HttpRequest, obj: func.InputStream):
logging.info(f'Python HTTP-triggered function processed: {obj.read()}')
عند استدعاء الدالة، يتم تمرير طلب HTTP إلى الدالة ك req
. يتم استرداد إدخال من حساب Azure Blob Storage استنادا إلى المعرف في عنوان URL للمسار وإتاحته كما هو الحال obj
في نص الدالة. هنا، حساب التخزين المحدد هو سلسلة الاتصال الذي تم العثور عليه في STORAGE_CONNECTION_STRING
إعداد التطبيق.
بالنسبة لعمليات ربط البيانات المكثفة، قد تحتاج إلى استخدام حساب تخزين منفصل. لمزيد من المعلومات، راجع إرشادات حساب التخزين.
روابط نوع SDK (معاينة)
بالنسبة إلى المشغلات والروابط المحددة، يمكنك العمل مع أنواع البيانات التي تنفذها Azure SDKs وأطر العمل الأساسية. تتيح لك روابط نوع SDK هذه التفاعل مع بيانات الربط كما لو كنت تستخدم SDK للخدمة الأساسية.
هام
يتطلب دعم روابط نوع SDK نموذج برمجة Python v2.
تدعم الوظائف روابط نوع Python SDK لتخزين Azure Blob، والذي يتيح لك العمل مع بيانات blob باستخدام النوع الأساسي BlobClient
.
هام
دعم روابط نوع SDK ل Python قيد المعاينة حاليا:
- يجب استخدام نموذج برمجة Python v2.
- حاليا، يتم دعم أنواع SDK المتزامنة فقط.
المتطلبات الأساسية
- إصدار وقت تشغيل Azure Functions 4.34 أو إصدار أحدث.
- Python الإصدار 3.9، أو إصدار مدعوم لاحقا.
تمكين روابط نوع SDK لملحق تخزين Blob
أضف حزمة الملحق
azurefunctions-extensions-bindings-blob
إلىrequirements.txt
الملف في المشروع، والذي يجب أن يتضمن هذه الحزم على الأقل:azure-functions azurefunctions-extensions-bindings-blob
أضف هذه التعليمة البرمجية
function_app.py
إلى الملف في المشروع، الذي يستورد روابط نوع SDK:import azurefunctions.extensions.bindings.blob as blob
أمثلة على روابط نوع SDK
يوضح هذا المثال كيفية الحصول على BlobClient
من كل من مشغل تخزين Blob (blob_trigger
) ومن ربط الإدخال على مشغل HTTP (blob_input
):
import logging
import azure.functions as func
import azurefunctions.extensions.bindings.blob as blob
app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)
@app.blob_trigger(
arg_name="client", path="PATH/TO/BLOB", connection="AzureWebJobsStorage"
)
def blob_trigger(client: blob.BlobClient):
logging.info(
f"Python blob trigger function processed blob \n"
f"Properties: {client.get_blob_properties()}\n"
f"Blob content head: {client.download_blob().read(size=1)}"
)
@app.route(route="file")
@app.blob_input(
arg_name="client", path="PATH/TO/BLOB", connection="AzureWebJobsStorage"
)
def blob_input(req: func.HttpRequest, client: blob.BlobClient):
logging.info(
f"Python blob input function processed blob \n"
f"Properties: {client.get_blob_properties()}\n"
f"Blob content head: {client.download_blob().read(size=1)}"
)
return "ok"
يمكنك عرض نماذج روابط نوع SDK الأخرى لتخزين Blob في مستودع ملحقات Python:
تدفقات HTTP (معاينة)
تتيح لك تدفقات HTTP قبول البيانات وإرجاعها من نقاط نهاية HTTP باستخدام واجهات برمجة التطبيقات لطلب FastAPI والاستجابة الممكنة في وظائفك. تتيح واجهات برمجة التطبيقات هذه للمضيف معالجة البيانات الكبيرة في رسائل HTTP كأجزاء بدلا من قراءة رسالة بأكملها في الذاكرة.
تجعل هذه الميزة من الممكن التعامل مع دفق البيانات الكبير، وتكامل OpenAI، وتقديم محتوى ديناميكي، ودعم سيناريوهات HTTP الأساسية الأخرى التي تتطلب تفاعلات في الوقت الحقيقي عبر HTTP. يمكنك أيضا استخدام أنواع استجابة FastAPI مع تدفقات HTTP. بدون تدفقات HTTP، يكون حجم طلبات واستجابات HTTP مقيدا بقيود الذاكرة التي يمكن مواجهتها عند معالجة حمولات الرسائل بأكملها في الذاكرة.
هام
يتطلب دعم تدفقات HTTP نموذج برمجة Python v2.
هام
دعم تدفقات HTTP ل Python قيد المعاينة حاليا ويتطلب منك استخدام نموذج برمجة Python v2.
المتطلبات الأساسية
- إصدار وقت تشغيل Azure Functions 4.34.1 أو إصدار أحدث.
- Python الإصدار 3.8، أو إصدار مدعوم لاحقا.
تمكين تدفقات HTTP
يتم تعطيل تدفقات HTTP بشكل افتراضي. تحتاج إلى تمكين هذه الميزة في إعدادات التطبيق الخاص بك وتحديث التعليمات البرمجية أيضا لاستخدام حزمة FastAPI. لاحظ أنه عند تمكين تدفقات HTTP، سيستخدم تطبيق الدالة بشكل افتراضي تدفق HTTP، ولن تعمل وظيفة HTTP الأصلية.
أضف حزمة الملحق
azurefunctions-extensions-http-fastapi
إلىrequirements.txt
الملف في المشروع، والذي يجب أن يتضمن هذه الحزم على الأقل:azure-functions azurefunctions-extensions-http-fastapi
أضف هذه التعليمة البرمجية
function_app.py
إلى الملف في المشروع، الذي يستورد ملحق FastAPI:from azurefunctions.extensions.http.fastapi import Request, StreamingResponse
عند النشر إلى Azure، أضف إعداد التطبيق التالي في تطبيق الوظائف:
"PYTHON_ENABLE_INIT_INDEXING": "1"
إذا كنت تقوم بالنشر إلى Linux Consumption، أضف أيضا
"PYTHON_ISOLATE_WORKER_DEPENDENCIES": "1"
عند التشغيل محليا، تحتاج أيضا إلى إضافة هذه الإعدادات نفسها إلى
local.settings.json
ملف المشروع.
أمثلة على تدفقات HTTP
بعد تمكين ميزة تدفق HTTP، يمكنك إنشاء وظائف تقوم ببث البيانات عبر HTTP.
هذا المثال هو دالة مشغلة من HTTP تقوم ببث بيانات استجابة HTTP. قد تستخدم هذه الإمكانات لدعم سيناريوهات مثل إرسال بيانات الحدث من خلال البنية الأساسية لبرنامج ربط العمليات التجارية للتصور في الوقت الحقيقي أو الكشف عن الحالات الشاذة في مجموعات كبيرة من البيانات وتوفير إعلامات فورية.
import time
import azure.functions as func
from azurefunctions.extensions.http.fastapi import Request, StreamingResponse
app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)
def generate_sensor_data():
"""Generate real-time sensor data."""
for i in range(10):
# Simulate temperature and humidity readings
temperature = 20 + i
humidity = 50 + i
yield f"data: {{'temperature': {temperature}, 'humidity': {humidity}}}\n\n"
time.sleep(1)
@app.route(route="stream", methods=[func.HttpMethod.GET])
async def stream_sensor_data(req: Request) -> StreamingResponse:
"""Endpoint to stream real-time sensor data."""
return StreamingResponse(generate_sensor_data(), media_type="text/event-stream")
هذا المثال هو دالة مشغلة من HTTP تتلقى البيانات المتدفقة وتعالجها من عميل في الوقت الفعلي. وهو يوضح قدرات تحميل الدفق التي يمكن أن تكون مفيدة لسيناريوهات مثل معالجة تدفقات البيانات المستمرة ومعالجة بيانات الحدث من أجهزة IoT.
import azure.functions as func
from azurefunctions.extensions.http.fastapi import JSONResponse, Request
app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)
@app.route(route="streaming_upload", methods=[func.HttpMethod.POST])
async def streaming_upload(req: Request) -> JSONResponse:
"""Handle streaming upload requests."""
# Process each chunk of data as it arrives
async for chunk in req.stream():
process_data_chunk(chunk)
# Once all data is received, return a JSON response indicating successful processing
return JSONResponse({"status": "Data uploaded and processed successfully"})
def process_data_chunk(chunk: bytes):
"""Process each data chunk."""
# Add custom processing logic here
pass
استدعاء تدفقات HTTP
يجب استخدام مكتبة عميل HTTP لإجراء مكالمات دفق إلى نقاط نهاية FastAPI للدالة. قد لا تدعم أداة العميل أو المستعرض الذي تستخدمه البث في الأصل أو قد ترجع المجموعة الأولى من البيانات فقط.
يمكنك استخدام برنامج نصي للعميل مثل هذا لإرسال البيانات المتدفقة إلى نقطة نهاية HTTP:
import httpx # Be sure to add 'httpx' to 'requirements.txt'
import asyncio
async def stream_generator(file_path):
chunk_size = 2 * 1024 # Define your own chunk size
with open(file_path, 'rb') as file:
while chunk := file.read(chunk_size):
yield chunk
print(f"Sent chunk: {len(chunk)} bytes")
async def stream_to_server(url, file_path):
timeout = httpx.Timeout(60.0, connect=60.0)
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.post(url, content=stream_generator(file_path))
return response
async def stream_response(response):
if response.status_code == 200:
async for chunk in response.aiter_raw():
print(f"Received chunk: {len(chunk)} bytes")
else:
print(f"Error: {response}")
async def main():
print('helloworld')
# Customize your streaming endpoint served from core tool in variable 'url' if different.
url = 'http://localhost:7071/api/streaming_upload'
file_path = r'<file path>'
response = await stream_to_server(url, file_path)
print(response)
if __name__ == "__main__":
asyncio.run(main())
المخرجات
يمكن التعبير عن الإخراج في كل من قيمة الإرجاع ومعلمات الإخراج. إذا كان هناك مخرج واحد فقط، فإننا نوصي باستخدام القيمة المعادة. بالنسبة إلى مخرجات متعددة، يجب استخدام معلمات الإخراج.
لاستخدام القيمة المرجعة لدالة كقيمة لربط إخراج، name
يجب تعيين خاصية الربط إلى $return
في ملف function.json .
لإنتاج مخرجات متعددة، استخدم set()
الأسلوب الذي توفره الواجهة azure.functions.Out
لتعيين قيمة للربط. على سبيل المثال، يمكن للوظيفة التالية دفع رسالة إلى قائمة انتظار وإرجاع استجابة HTTP أيضًا.
{
"scriptFile": "__init__.py",
"bindings": [
{
"name": "req",
"direction": "in",
"type": "httpTrigger",
"authLevel": "anonymous"
},
{
"name": "msg",
"direction": "out",
"type": "queue",
"queueName": "outqueue",
"connection": "STORAGE_CONNECTION_STRING"
},
{
"name": "$return",
"direction": "out",
"type": "http"
}
]
}
import azure.functions as func
def main(req: func.HttpRequest,
msg: func.Out[func.QueueMessage]) -> str:
message = req.params.get('body')
msg.set(message)
return message
يمكن التعبير عن الإخراج في كل من قيمة الإرجاع ومعلمات الإخراج. إذا كان هناك مخرج واحد فقط، فإننا نوصي باستخدام القيمة المعادة. بالنسبة للمخرجات المتعددة، سيتعين عليك استخدام معلمات الإخراج.
لإنتاج مخرجات متعددة، استخدم set()
الأسلوب الذي توفره الواجهة azure.functions.Out
لتعيين قيمة للربط. على سبيل المثال، يمكن للوظيفة التالية دفع رسالة إلى قائمة انتظار وإرجاع استجابة HTTP أيضًا.
# function_app.py
import azure.functions as func
app = func.FunctionApp()
@app.write_blob(arg_name="msg", path="output-container/{name}",
connection="CONNECTION_STRING")
def test_function(req: func.HttpRequest,
msg: func.Out[str]) -> str:
message = req.params.get('body')
msg.set(message)
return message
تسجيل الدخول
يتوفر الوصول إلى مسجل وقت تشغيل Azure Functions عبر معالج جذر logging
في تطبيق الوظائف. هذا المسجل مرتبط بـ Application Insights ويسمح لك بوضع علامة على التحذيرات والأخطاء التي تحدث أثناء تنفيذ الوظيفة.
يسجل المثال التالي رسالة معلومات عند استدعاء الوظيفة عبر مشغل HTTP.
import logging
def main(req):
logging.info('Python HTTP trigger function processed a request.')
تتوفر المزيد من طرق التسجيل التي تتيح لك الكتابة إلى وحدة التحكم بمستويات تتبع مختلفة:
الطريقة | الوصف |
---|---|
critical(_message_) |
يكتب رسالة بمستوى CRITICAL على مسجل الجذر. |
error(_message_) |
يكتب رسالة بمستوى ERROR على مسجل الجذر. |
warning(_message_) |
يكتب رسالة بمستوى WARNING على مسجل الجذر. |
info(_message_) |
يكتب رسالة بمستوى INFO على مسجل الجذر. |
debug(_message_) |
يكتب رسالة بمستوى DEBUG على مسجل الجذر. |
لمعرفة المزيد حول التسجيل، راجع مراقبة وظائف Azure.
تسجيل الدخول من مؤشرات الترابط التي تم إنشاؤها
لمشاهدة السجلات القادمة من مؤشرات الترابط التي تم إنشاؤها، قم بتضمين الوسيطة context
في توقيع الدالة. تحتوي هذه الوسيطة على سمة thread_local_storage
تخزن محلي .invocation_id
يمكن تعيين هذا إلى تيار invocation_id
الدالة لضمان تغيير السياق.
import azure.functions as func
import logging
import threading
def main(req, context):
logging.info('Python HTTP trigger function processed a request.')
t = threading.Thread(target=log_function, args=(context,))
t.start()
def log_function(context):
context.thread_local_storage.invocation_id = context.invocation_id
logging.info('Logging from thread.')
تسجيل القياس عن بُعد المخصص
بشكل افتراضي، يجمع وقت تشغيل الوظائف السجلات وبيانات القياس عن بعد الأخرى التي يتم إنشاؤها بواسطة وظائفك. ينتهي هذا القياس عن بعد كتتبعات في رؤى التطبيق. يتم أيضا جمع بيانات تتبع الاستخدام للطلب والتبعية لبعض خدمات Azure بشكل افتراضي بواسطة المشغلات والروابط.
لتجميع طلب مخصص وبيانات تتبع استخدام التبعية المخصصة خارج الروابط، يمكنك استخدام OpenCensus Python Extensions. يرسل هذا الملحق بيانات القياس عن بعد المخصصة إلى مثيل Application Insights. يمكنك العثور على قائمة بالملحقات المدعومة في مستودع OpenCensus.
إشعار
لاستخدام امتدادات OpenCensus Python، يلزمك تمكين امتدادات Python للعاملين في تطبيق الوظيفة الخاص بك عن طريق تعيين PYTHON_ENABLE_WORKER_EXTENSIONS
إلى 1
. كما تحتاج إلى التبديل إلى استخدام سلسلة اتصال Application Insights عن طريق إضافة الإعداد APPLICATIONINSIGHTS_CONNECTION_STRING
إلى إعدادات التطبيق الخاص بك، إذا لم يكن موجودًا بالفعل.
// requirements.txt
...
opencensus-extension-azure-functions
opencensus-ext-requests
import json
import logging
import requests
from opencensus.extension.azure.functions import OpenCensusExtension
from opencensus.trace import config_integration
config_integration.trace_integrations(['requests'])
OpenCensusExtension.configure()
def main(req, context):
logging.info('Executing HttpTrigger with OpenCensus extension')
# You must use context.tracer to create spans
with context.tracer.span("parent"):
response = requests.get(url='http://example.com')
return json.dumps({
'method': req.method,
'response': response.status_code,
'ctx_func_name': context.function_name,
'ctx_func_dir': context.function_directory,
'ctx_invocation_id': context.invocation_id,
'ctx_trace_context_Traceparent': context.trace_context.Traceparent,
'ctx_trace_context_Tracestate': context.trace_context.Tracestate,
'ctx_retry_context_RetryCount': context.retry_context.retry_count,
'ctx_retry_context_MaxRetryCount': context.retry_context.max_retry_count,
})
مشغل HTTP
يتم تعريف مشغل HTTP في ملف function.json. name
يجب أن يتطابق الربط مع المعلمة المسماة في الدالة .
في الأمثلة السابقة، يتم استخدام اسم req
الربط. هذه المعلمة هي كائن HttpRequest ، ويتم إرجاع كائن HttpResponse .
من كائن HttpRequest، يمكنك الحصول على رؤوس الطلبات ومعلمات الاستعلام ومعلمات المسار والنص الأساسي للرسالة.
المثال التالي من قالب مشغل HTTP ل Python.
def main(req: func.HttpRequest) -> func.HttpResponse:
headers = {"my-http-header": "some-value"}
name = req.params.get('name')
if not name:
try:
req_body = req.get_json()
except ValueError:
pass
else:
name = req_body.get('name')
if name:
return func.HttpResponse(f"Hello {name}!", headers=headers)
else:
return func.HttpResponse(
"Please pass a name on the query string or in the request body",
headers=headers, status_code=400
)
في هذه الدالة، يمكنك الحصول على قيمة معلمة name
الاستعلام من معلمة params
كائن HttpRequest . يمكنك قراءة نص الرسالة المشفرة JSON باستخدام get_json
الأسلوب .
وبالمثل، يمكنك تعيين status_code
و headers
لرسالة الاستجابة في كائن HttpResponse الذي تم إرجاعه.
يتم تعريف مشغل HTTP كأسلوب يأخذ معلمة ربط مسماة، وهو كائن HttpRequest ، ويعيد كائن HttpResponse . يمكنك تطبيق function_name
مصمم الديكور على الأسلوب لتعريف اسم الدالة، بينما يتم تعيين نقطة نهاية HTTP عن طريق تطبيق route
مصمم الديكور.
هذا المثال من قالب مشغل HTTP لنموذج برمجة Python v2، حيث يكون اسم معلمة الربط هو req
. إنها عينة التعليمات البرمجية التي يتم توفيرها عند إنشاء دالة باستخدام Azure Functions Core Tools أو Visual Studio Code.
@app.function_name(name="HttpTrigger1")
@app.route(route="hello")
def test_function(req: func.HttpRequest) -> func.HttpResponse:
logging.info('Python HTTP trigger function processed a request.')
name = req.params.get('name')
if not name:
try:
req_body = req.get_json()
except ValueError:
pass
else:
name = req_body.get('name')
if name:
return func.HttpResponse(f"Hello, {name}. This HTTP-triggered function executed successfully.")
else:
return func.HttpResponse(
"This HTTP-triggered function executed successfully. Pass a name in the query string or in the request body for a personalized response.",
status_code=200
)
من كائن HttpRequest، يمكنك الحصول على رؤوس الطلبات ومعلمات الاستعلام ومعلمات المسار والنص الأساسي للرسالة. في هذه الدالة، يمكنك الحصول على قيمة معلمة name
الاستعلام من معلمة params
كائن HttpRequest . يمكنك قراءة نص الرسالة المشفرة JSON باستخدام get_json
الأسلوب .
وبالمثل، يمكنك تعيين status_code
و headers
لرسالة الاستجابة في كائن HttpResponse الذي تم إرجاعه.
لتمرير اسم في هذا المثال، الصق عنوان URL الذي يتم توفيره عند تشغيل الدالة، ثم قم بإلحاقه ب "?name={name}"
.
أطر عمل الويب
يمكنك استخدام أطر عمل متوافقة مع واجهة بوابة خادم الويب (WSGI) ومتوافقة مع واجهة بوابة الخادم (ASGI)، مثل Flask وFastAPI، مع وظائف Python التي يتم تشغيلها بواسطة HTTP. هذا القسم يوضح كيفية تعديل دالاتك لدعم أطر العمل هذه.
أولا، يجب تحديث ملف function.json لتضمين route
في مشغل HTTP، كما هو موضح في المثال التالي:
{
"scriptFile": "__init__.py",
"bindings": [
{
"authLevel": "anonymous",
"type": "httpTrigger",
"direction": "in",
"name": "req",
"methods": [
"get",
"post"
],
"route": "{*route}"
},
{
"type": "http",
"direction": "out",
"name": "$return"
}
]
}
يجب أيضا تحديث ملف host.json لتضمين HTTP routePrefix
، كما هو موضح في المثال التالي:
{
"version": "2.0",
"logging":
{
"applicationInsights":
{
"samplingSettings":
{
"isEnabled": true,
"excludedTypes": "Request"
}
}
},
"extensionBundle":
{
"id": "Microsoft.Azure.Functions.ExtensionBundle",
"version": "[3.*, 4.0.0)"
},
"extensions":
{
"http":
{
"routePrefix": ""
}
}
}
قم بتحديث ملف التعليمات البرمجية ل Python init.py، اعتمادا على الواجهة التي يستخدمها إطار العمل الخاص بك. المثال التالي يوضح إما نهج معالج ASGI أو نهج تضمين WSGI لـ Flask:
يمكنك استخدام أطر عمل متوافقة مع واجهة بوابة الخادم غير المتزامنة (ASGI) ومتوافقة مع واجهة بوابة خادم الويب (WSGI)، مثل Flask وFastAPI، مع وظائف Python التي يتم تشغيلها بواسطة HTTP. يجب أولا تحديث ملف host.json لتضمين HTTP routePrefix
، كما هو موضح في المثال التالي:
{
"version": "2.0",
"logging":
{
"applicationInsights":
{
"samplingSettings":
{
"isEnabled": true,
"excludedTypes": "Request"
}
}
},
"extensionBundle":
{
"id": "Microsoft.Azure.Functions.ExtensionBundle",
"version": "[2.*, 3.0.0)"
},
"extensions":
{
"http":
{
"routePrefix": ""
}
}
}
تبدو التعليمات البرمجية لإطار العمل مثل المثال التالي:
AsgiFunctionApp
هو فئة تطبيق الوظائف ذات المستوى الأعلى لإنشاء وظائف ASGI HTTP.
# function_app.py
import azure.functions as func
from fastapi import FastAPI, Request, Response
fast_app = FastAPI()
@fast_app.get("/return_http_no_body")
async def return_http_no_body():
return Response(content="", media_type="text/plain")
app = func.AsgiFunctionApp(app=fast_app,
http_auth_level=func.AuthLevel.ANONYMOUS)
التحجيم والأداء
للحصول على أفضل ممارسات التحجيم والأداء لتطبيقات وظائف Python، راجع مقالة تحجيم Python والأداء .
السياق
للحصول على سياق استدعاء دالة عند تشغيلها، قم بتضمين الوسيطة context
في توقيعها.
على سبيل المثال:
import azure.functions
def main(req: azure.functions.HttpRequest,
context: azure.functions.Context) -> str:
return f'{context.invocation_id}'
Context
تحتوي الفئة على سمات السلسلة التالية:
السمة | الوصف |
---|---|
function_directory |
الدليل الذي يتم تشغيل الدالة فيه. |
function_name |
اسم الدالة |
invocation_id |
معرف استدعاء الدالة الحالية. |
thread_local_storage |
تخزين مؤشر الترابط المحلي للدالة. يحتوي على محلي invocation_id للتسجيل من مؤشرات الترابط التي تم إنشاؤها. |
trace_context |
سياق التتبع الموزع. لمزيد من المعلومات، انظر Trace Context . |
retry_context |
سياق إعادة المحاولة إلى الدالة. لمزيد من المعلومات، انظر retry-policies . |
المتغيرات العمومية
ليس من المضمون الحفاظ على حالة تطبيقك لعمليات التنفيذ المستقبلية. ومع ذلك، فإن وقت تشغيل Azure Functions غالبًا ما يعيد استخدام نفس العملية لعمليات تنفيذ متعددة لنفس التطبيق. لتخزين نتائج حساب مكلف مؤقتا، قم بتعريفه كمتغير عمومي.
CACHED_DATA = None
def main(req):
global CACHED_DATA
if CACHED_DATA is None:
CACHED_DATA = load_json()
# ... use CACHED_DATA in code
متغيرات البيئة
في Azure Functions، يتم عرض إعدادات التطبيق، مثل سلسلة الاتصال الخدمة، كمتغيرات بيئة عند تشغيلها. هناك طريقتان رئيسيتان للوصول إلى هذه الإعدادات في التعليمات البرمجية الخاصة بك.
الطريقة | الوصف |
---|---|
os.environ["myAppSetting"] |
يحاول الحصول على إعداد التطبيق حسب اسم المفتاح، ويثير خطأ عندما يكون غير ناجح. |
os.getenv("myAppSetting") |
يحاول الحصول على إعداد التطبيق حسب اسم المفتاح، ويرجع None عندما يكون غير ناجح. |
تتطلب كلتا الطريقتين منك الإعلان عن import os
.
يستخدم os.environ["myAppSetting"]
المثال التالي للحصول على إعداد التطبيق، مع المفتاح المسمى myAppSetting
:
import logging
import os
import azure.functions as func
def main(req: func.HttpRequest) -> func.HttpResponse:
# Get the setting named 'myAppSetting'
my_app_setting_value = os.environ["myAppSetting"]
logging.info(f'My app setting value:{my_app_setting_value}')
للتطوير المحلي، يتم الاحتفاظ بإعدادات التطبيق في ملف local.settings.json.
في Azure Functions، يتم عرض إعدادات التطبيق، مثل سلسلة الاتصال الخدمة، كمتغيرات بيئة عند تشغيلها. هناك طريقتان رئيسيتان للوصول إلى هذه الإعدادات في التعليمات البرمجية الخاصة بك.
الطريقة | الوصف |
---|---|
os.environ["myAppSetting"] |
يحاول الحصول على إعداد التطبيق حسب اسم المفتاح، ويثير خطأ عندما يكون غير ناجح. |
os.getenv("myAppSetting") |
يحاول الحصول على إعداد التطبيق حسب اسم المفتاح، ويرجع None عندما يكون غير ناجح. |
تتطلب كلتا الطريقتين منك الإعلان عن import os
.
يستخدم os.environ["myAppSetting"]
المثال التالي للحصول على إعداد التطبيق، مع المفتاح المسمى myAppSetting
:
import logging
import os
import azure.functions as func
app = func.FunctionApp()
@app.function_name(name="HttpTrigger1")
@app.route(route="req")
def main(req: func.HttpRequest) -> func.HttpResponse:
# Get the setting named 'myAppSetting'
my_app_setting_value = os.environ["myAppSetting"]
logging.info(f'My app setting value:{my_app_setting_value}')
للتطوير المحلي، يتم الاحتفاظ بإعدادات التطبيق في ملف local.settings.json.
إصدار Python
تدعم وظائف Azure إصدارات Python التالية:
إصدار الدوال | إصدارات Python* |
---|---|
4.x | 3.11 3.10 3.9 3.8 3.7 |
3.x | 3.9 3.8 3.7 |
* توزيعات Python الرسمية
لطلب إصدار Python معين عند إنشاء تطبيق الوظائف في Azure، استخدم --runtime-version
خيار az functionapp create
الأمر . يتم تعيين إصدار وقت تشغيل الوظائف بواسطة --functions-version
الخيار . يتم تعيين إصدار Python عند إنشاء تطبيق الوظائف، ولا يمكن تغييره للتطبيقات التي تعمل في خطة الاستهلاك.
يستخدم وقت التشغيل إصدار Python المتوفر عند تشغيله محليا.
تغيير نسخة Python
لتعيين تطبيق دالة Python إلى إصدار لغة معين، تحتاج إلى تحديد اللغة وإصدار اللغة في LinuxFxVersion
الحقل في تكوين الموقع. على سبيل المثال، لتغيير تطبيق Python لاستخدام Python 3.8، قم بتعيين linuxFxVersion
إلى python|3.8
.
لمعرفة كيفية عرض إعداد الموقع وتغييره linuxFxVersion
، راجع كيفية استهداف إصدارات وقت تشغيل Azure Functions.
لمزيد من المعلومات العامة، راجع نهج دعم وقت تشغيل Azure Functions واللغات المدعومة في Azure Functions.
إدارة الحزمة
عند التطوير محليا باستخدام Core Tools أو Visual Studio Code، أضف أسماء الحزم المطلوبة وإصداراتها إلى ملف requirements.txt ، ثم قم بتثبيتها باستخدام pip
.
على سبيل المثال، يمكنك استخدام ملف requirements.txt التالي والأمر pip
لتثبيت الحزمة requests
من PyPI.
requests==2.19.1
pip install -r requirements.txt
عند تشغيل الوظائف في خطة App Service، يتم إعطاء التبعيات التي تحددها في requirements.txt الأسبقية على وحدات Python المضمنة، مثل logging
. يمكن أن تسبب هذه الأسبقية تعارضات عندما يكون للوحدات النمطية المضمنة نفس أسماء الدلائل في التعليمات البرمجية الخاصة بك. عند التشغيل في خطة استهلاك أو خطة Elastic Premium، تكون التعارضات أقل احتمالا لأن تبعياتك لا يتم تحديد أولوياتها بشكل افتراضي.
لمنع المشكلات قيد التشغيل في خطة App Service، لا تقم بتسمية دلائلك بنفس اسم أي وحدات Python أصلية ولا تقم بتضمين مكتبات Python الأصلية في ملف requirements.txt الخاص بمشروعك.
النشر إلى Azure
عندما تكون جاهزًا للنشر، تأكد من إدراج جميع التبعيات المتاحة للجمهور في ملف requirements.txt. يمكنك تحديد موقع هذا الملف في جذر دليل المشروع.
يمكنك العثور على ملفات المشروع والمجلدات المستبعدة من النشر، بما في ذلك مجلد البيئة الظاهرية، في الدليل الجذر لمشروعك.
هناك ثلاثة إجراءات بناء مدعومة لنشر مشروع Python الخاص بك على Azure: الإنشاء عن بُعد، والبناء المحلي، والإنشاءات باستخدام التبعيات المخصصة.
يمكنك أيضًا استخدام Azure Pipelines لبناء التبعيات الخاصة بك والنشر باستخدام التسليم المستمر (CD). لمعرفة المزيد، راجع التسليم المستمر باستخدام Azure Pipelines.
بناء بعيد
عند استخدام الإنشاء عن بعد، تتطابق التبعيات التي تتم استعادتها على الخادم والتبعيات الأصلية مع بيئة الإنتاج. ينتج عن هذا حزمة نشر أصغر للتحميل. استخدم الإنشاء عن بعد عند تطوير تطبيقات Python على Windows. إذا كان مشروعك يحتوي على تبعيات مخصصة، يمكنك استخدام الإنشاء عن بعد مع عنوان URL إضافي للفهرس.
يتم الحصول على التبعيات عن بُعد بناءً على محتويات ملف requirements.txt. الإنشاء عن بعد هو أسلوب البناء الموصى به. بشكل افتراضي، تطلب Core Tools إنشاء عن بعد عند استخدام الأمر التالي func azure functionapp publish
لنشر مشروع Python الخاص بك إلى Azure.
func azure functionapp publish <APP_NAME>
تذكر استبدال <APP_NAME>
باسم تطبيق الوظائف في Azure.
يطلب ملحق Azure Functions ل Visual Studio Code أيضا إنشاء عن بعد بشكل افتراضي.
بناء محلي
يتم الحصول على التبعيات محليًا بناءً على محتويات ملف requirements.txt. يمكنك منع إجراء بنية عن بعد باستخدام الأمر التالي func azure functionapp publish
للنشر باستخدام بنية محلية:
func azure functionapp publish <APP_NAME> --build local
تذكر استبدال <APP_NAME>
باسم تطبيق الوظائف في Azure.
عند استخدام --build local
الخيار، تتم قراءة تبعيات المشروع من ملف requirements.txt ، ويتم تنزيل هذه الحزم التابعة وتثبيتها محليا. يتم توزيع ملفات المشروع والتبعيات من الكمبيوتر المحلي الخاص بك إلى Azure. ينتج عن هذا تحميل حزمة نشر أكبر إلى Azure. إذا لم تتمكن لسبب ما من الحصول على ملف requirements.txt باستخدام Core Tools، فيجب عليك استخدام خيار التبعيات المخصصة للنشر.
لا نوصي باستخدام البنيات المحلية عند التطوير محليًا على Windows.
التبعيات المخصصة
عندما يكون لمشروعك تبعيات غير موجودة في فهرس حزمة Python، هناك طريقتان لإنشاء المشروع. تعتمد الطريقة الأولى، أسلوب البناء ، على كيفية إنشاء المشروع.
بناء عن بعد مع فهرس إضافي URL
عندما تتوفر الحزم الخاصة بك من فهرس حزمة مخصص يمكن الوصول إليه، استخدم بنية بعيدة. قبل النشر، تأكد من إنشاء إعداد تطبيق يسمى PIP_EXTRA_INDEX_URL
. قيمة هذا الإعداد هي عنوان URL لفهرس الحزمة المخصص الخاص بك. يؤدي استخدام هذا الإعداد إلى إعلام البنية البعيدة بالتشغيل pip install
باستخدام --extra-index-url
الخيار . لمعرفة المزيد، راجع وثائق Pythonpip install
.
يمكنك أيضًا استخدام بيانات اعتماد المصادقة الأساسية مع عناوين URL الإضافية لفهرس الحزمة. لمعرفة المزيد، راجع بيانات اعتماد المصادقة الأساسية في وثائق Python.
تثبيت الحزم المحلية
إذا كان مشروعك يستخدم حزما غير متوفرة بشكل عام لأدواتنا، فيمكنك توفيرها لتطبيقك عن طريق وضعها في دليل __app__/.python_packages . قبل النشر، قم بتشغيل الأمر التالي لتثبيت التبعيات محليا:
pip install --target="<PROJECT_DIR>/.python_packages/lib/site-packages" -r requirements.txt
عند استخدام تبعيات مخصصة، يجب استخدام --no-build
خيار النشر، لأنك قمت بالفعل بتثبيت التبعيات في مجلد المشروع.
func azure functionapp publish <APP_NAME> --no-build
تذكر استبدال <APP_NAME>
باسم تطبيق الوظائف في Azure.
اختبار الوحدة
يمكن اختبار الوظائف المكتوبة بلغة Python مثل تعليمة Python البرمجية الأخرى باستخدام أطر عمل الاختبار القياسية. بالنسبة لمعظم الروابط، من الممكن إنشاء كائن إدخال وهمي عن طريق إنشاء مثيل لفئة مناسبة من الحزمة azure.functions
. نظرا لأن الحزمة azure.functions
غير متوفرة على الفور، تأكد من تثبيتها عبر ملف requirements.txt كما هو موضح في قسم إدارة الحزمة أعلاه.
مع my_second_function كمثال، ما يلي هو اختبار وهمي للدالة التي يتم تشغيلها بواسطة HTTP:
أولا، قم بإنشاء <ملف project_root>/my_second_function/function.json ، ثم حدد هذه الدالة كمشغل HTTP.
{
"scriptFile": "__init__.py",
"entryPoint": "main",
"bindings": [
{
"authLevel": "function",
"type": "httpTrigger",
"direction": "in",
"name": "req",
"methods": [
"get",
"post"
]
},
{
"type": "http",
"direction": "out",
"name": "$return"
}
]
}
بعد ذلك، يمكنك تنفيذ my_second_function
و shared_code.my_second_helper_function
.
# <project_root>/my_second_function/__init__.py
import azure.functions as func
import logging
# Use absolute import to resolve shared_code modules
from shared_code import my_second_helper_function
# Define an HTTP trigger that accepts the ?value=<int> query parameter
# Double the value and return the result in HttpResponse
def main(req: func.HttpRequest) -> func.HttpResponse:
logging.info('Executing my_second_function.')
initial_value: int = int(req.params.get('value'))
doubled_value: int = my_second_helper_function.double(initial_value)
return func.HttpResponse(
body=f"{initial_value} * 2 = {doubled_value}",
status_code=200
)
# <project_root>/shared_code/__init__.py
# Empty __init__.py file marks shared_code folder as a Python package
# <project_root>/shared_code/my_second_helper_function.py
def double(value: int) -> int:
return value * 2
يمكنك البدء في كتابة حالات الاختبار لمشغل HTTP الخاص بك.
# <project_root>/tests/test_my_second_function.py
import unittest
import azure.functions as func
from my_second_function import main
class TestFunction(unittest.TestCase):
def test_my_second_function(self):
# Construct a mock HTTP request.
req = func.HttpRequest(method='GET',
body=None,
url='/api/my_second_function',
params={'value': '21'})
# Call the function.
resp = main(req)
# Check the output.
self.assertEqual(resp.get_body(), b'21 * 2 = 42',)
.venv
داخل مجلد بيئة Python الظاهرية، قم بتثبيت إطار عمل اختبار Python المفضل لديك، مثل pip install pytest
. ثم قم بتشغيل pytest tests
للتحقق من نتيجة الاختبار.
أولا، قم بإنشاء <ملف project_root>/function_app.py وتنفيذ الدالة my_second_function
كمشغل HTTP و shared_code.my_second_helper_function
.
# <project_root>/function_app.py
import azure.functions as func
import logging
# Use absolute import to resolve shared_code modules
from shared_code import my_second_helper_function
app = func.FunctionApp()
# Define the HTTP trigger that accepts the ?value=<int> query parameter
# Double the value and return the result in HttpResponse
@app.function_name(name="my_second_function")
@app.route(route="hello")
def main(req: func.HttpRequest) -> func.HttpResponse:
logging.info('Executing my_second_function.')
initial_value: int = int(req.params.get('value'))
doubled_value: int = my_second_helper_function.double(initial_value)
return func.HttpResponse(
body=f"{initial_value} * 2 = {doubled_value}",
status_code=200
)
# <project_root>/shared_code/__init__.py
# Empty __init__.py file marks shared_code folder as a Python package
# <project_root>/shared_code/my_second_helper_function.py
def double(value: int) -> int:
return value * 2
يمكنك البدء في كتابة حالات الاختبار لمشغل HTTP الخاص بك.
# <project_root>/tests/test_my_second_function.py
import unittest
import azure.functions as func
from function_app import main
class TestFunction(unittest.TestCase):
def test_my_second_function(self):
# Construct a mock HTTP request.
req = func.HttpRequest(method='GET',
body=None,
url='/api/my_second_function',
params={'value': '21'})
# Call the function.
func_call = main.build().get_user_function()
resp = func_call(req)
# Check the output.
self.assertEqual(
resp.get_body(),
b'21 * 2 = 42',
)
داخل مجلد البيئة الظاهرية .venv Python، قم بتثبيت إطار عمل اختبار Python المفضل لديك، مثل pip install pytest
. ثم قم بتشغيل pytest tests
للتحقق من نتيجة الاختبار.
ملفات مؤقتة
ترجع الطريقة tempfile.gettempdir()
مجلد مؤقت، وهو /tmp في Linux. يمكن للتطبيق الخاص بك استخدام هذا الدليل لتخزين الملفات المؤقتة التي يتم إنشاؤها واستخدامها بواسطة الوظائف الخاصة بك عند تشغيلها.
هام
لا يمكن ضمان استمرار الملفات المكتوبة في الدليل المؤقت عبر الاستدعاءات. أثناء التوسع، لا تتم مشاركة الملفات المؤقتة بين المثيلات.
ينشئ المثال التالي ملف مؤقت مسمى في الدليل المؤقت (/tmp):
import logging
import azure.functions as func
import tempfile
from os import listdir
#---
tempFilePath = tempfile.gettempdir()
fp = tempfile.NamedTemporaryFile()
fp.write(b'Hello world!')
filesDirListInTemp = listdir(tempFilePath)
نوصي بأن تحتفظ باختباراتك في مجلد منفصل عن مجلد المشروع. هذا الإجراء يمنعك من توزيع التعليمة البرمجية الاختبارية مع تطبيقك.
مكتبات مثبتة مسبقًا
تأتي بعض المكتبات مع وقت تشغيل وظائف Python.
مكتبة Python القياسية
تحتوي مكتبة Python القياسية على قائمة بوحدات Python المضمنة التي يتم شحنها مع كل توزيع Python. تساعدك معظم هذه المكتبات على الوصول إلى وظائف النظام، مثل إدخال/إخراج الملف (الإدخال/الإخراج). على أنظمة Windows، يتم تثبيت هذه المكتبات باستخدام Python. على الأنظمة المستندة إلى Unix، يتم توفيرها بواسطة مجموعات الحزم.
لعرض مكتبة إصدار Python، انتقل إلى:
- مكتبة Python 3.8 القياسية
- مكتبة Python 3.9 القياسية
- مكتبة Python 3.10 القياسية
- مكتبة Python 3.11 القياسية
وظائف Azure تبعيات عامل Python
يتطلب عامل Azure Functions Python مجموعة معينة من المكتبات. يمكنك أيضًا استخدام هذه المكتبات في وظائفك، لكنها ليست جزءًا من معيار Python. إذا كانت وظائفك تعتمد على أي من هذه المكتبات، فقد تكون غير متوفرة للتعليمات البرمجية الخاصة بك عند تشغيلها خارج Azure Functions.
إشعار
إذا احتوى ملف requirements.txt الخاص بتطبيق دالتك على إدخال azure-functions-worker
، فقم بإزالته. تتم إدارة عامل الوظائف تلقائيا بواسطة النظام الأساسي ل Azure Functions، ونحدثه بانتظام بميزات جديدة وإصلاحات الأخطاء. قد يؤدي تثبيت إصدار قديم من العامل يدويا في ملف requirements.txt إلى حدوث مشكلات غير متوقعة.
إشعار
إذا كانت الحزمة تحتوي على مكتبات معينة قد تتصادم مع تبعيات العامل (على سبيل المثال، protobuf أو tensorflow أو grpcio)، قم بالتكوين PYTHON_ISOLATE_WORKER_DEPENDENCIES
إلى 1
في إعدادات التطبيق لمنع تطبيقك من الإشارة إلى تبعيات العامل.
مكتبة Azure Functions Python
يتضمن كل تحديث لعامل Python إصدارا جديدا من مكتبة Azure Functions Python (azure.functions). يسهّل هذا الأسلوب تحديث تطبيقات وظائف Python باستمرار، لأن كل تحديث متوافق مع الإصدارات السابقة. للحصول على قائمة بإصدارات هذه المكتبة، انتقل إلى azure-functions PyPi.
تم إصلاح إصدار مكتبة وقت التشغيل بواسطة Azure، ولا يمكن تجاوزه بواسطة requirements.txt. يعد الإدخال azure-functions
في requirements.txt من أجل الفحص وإدراك العملاء فحسب.
استخدم التعليمات البرمجية التالية لتعقب الإصدار الفعلي من مكتبة وظائف Python في وقت التشغيل الخاص بك:
getattr(azure.functions, '__version__', '< 1.2.1')
مكتبات نظام وقت التشغيل
للحصول على قائمة بمكتبات النظام المثبتة مسبقا في صور Docker لعامل Python، راجع ما يلي:
وقت تشغيل الوظائف | نسخة دبيان | إصدارات Python |
---|---|---|
الإصدار 3.x | المغفل | Python 3.7 Python 3.8 Python 3.9 |
ملحقات عامل Python
تتيح لك عملية عامل Python التي يتم تشغيلها في Azure Functions دمج مكتبات الجهات الخارجية في تطبيق وظيفتك. تعمل مكتبات الامتدادات هذه كبرامج وسيطة يمكنها ضخ عمليات محددة أثناء دورة حياة تنفيذ وظيفتك.
يتم استيراد الامتدادات في كود وظيفتك مثل وحدة مكتبة Python القياسية. يتم تشغيل الملحقات استنادا إلى النطاقات التالية:
Scope | الوصف |
---|---|
مستوى التطبيق | عند استيرادها إلى أي مشغل وظيفة، يتم تطبيق الامتداد على كل وظيفة يتم تنفيذها في التطبيق. |
مستوى الدالة | يقتصر التنفيذ على مشغل الوظيفة المحددة التي يتم استيرادها إليه. |
راجع المعلومات الخاصة بكل ملحق لمعرفة المزيد حول النطاق الذي يتم تشغيل الملحق فيه.
الملحقات تنفذ واجهة ملحق عامل Python. يتيح هذا الإجراء لعملية عامل Python استدعاء التعليمات البرمجية للملحق أثناء دورة حياة تنفيذ الوظيفة. لمعرفة المزيد، راجع إنشاء ملحقات.
باستخدام الامتدادات
يمكنك استخدام مكتبة ملحق عامل Python في وظائف Python الخاصة بك عن طريق القيام بما يلي:
- أضف حزمة الامتداد في ملف requirements.txt لمشروعك.
- ثبّت المكتبة في تطبيقك.
- أضف إعدادات التطبيق التالية:
- قم باستيراد وحدة الامتداد إلى مشغل الوظيفة الخاص بك.
- قم بتكوين مثيل الامتداد، إذا لزم الأمر. يجب ذكر متطلبات التكوين في وثائق الامتداد.
هام
مكتبات ملحق عامل Python التابعة لجهة خارجية غير مدعومة أو مضمونة من قبل Microsoft. يجب التأكد من أن أي ملحقات تستخدمها في تطبيق الوظائف الخاص بك جديرة بالثقة، وأنك تتحمل المخاطر الكاملة لاستخدام ملحق ضار أو مكتوب بشكل سيئ.
يجب أن توفر الجهات الخارجية وثائق محددة حول كيفية تثبيت ملحقاتها واستهلاكها في تطبيق الوظائف الخاص بك. للحصول على مثال أساسي حول كيفية استهلاك ملحق، راجع استهلاك الملحق الخاص بك.
فيما يلي أمثلة لاستخدام الإضافات في تطبيق وظيفي حسب النطاق:
# <project_root>/requirements.txt
application-level-extension==1.0.0
# <project_root>/Trigger/__init__.py
from application_level_extension import AppExtension
AppExtension.configure(key=value)
def main(req, context):
# Use context.app_ext_attributes here
إنشاء ملحقات
يتم إنشاء الإضافات بواسطة مطوري مكتبات تابعين لجهات خارجية قاموا بإنشاء وظائف يمكن دمجها في وظائف Azure. يقوم مطور الإضافات بتصميم وتنفيذ وإصدار حزم Python التي تحتوي على منطق مخصص مصمم خصيصًا ليتم تشغيله في سياق تنفيذ الوظيفة. يمكن نشر هذه الامتدادات إما في سجل PyPI أو في مستودعات GitHub.
لمعرفة كيفية إنشاء حزمة ملحق عامل Python وحزمها ونشرها واستهلاكها، راجع تطوير ملحقات عامل Python لوظائف Azure.
امتدادات على مستوى التطبيق
يتم تشغيل ملحق موروث من AppExtensionBase
في نطاق تطبيق .
AppExtensionBase
يعرض أساليب الفئة المجردة التالية لتنفيذها:
الطريقة | الوصف |
---|---|
init |
تم استدعاؤه بعد استيراد الامتداد. |
configure |
يتم استدعاؤه من تعليمة برمجية الدالة عند الحاجة لتكوين الامتداد. |
post_function_load_app_level |
يتم استدعاؤه بعد تحميل الوظيفة مباشرة. يتم تمرير اسم الوظيفة ودليل الوظيفة إلى الامتداد. ضع في اعتبارك أن دليل الدالة للقراءة فقط، وأن أي محاولة للكتابة إلى ملف محلي في هذا الدليل تفشل. |
pre_invocation_app_level |
يتم استدعاؤها قبل تشغيل الوظيفة مباشرة. سياق الوظيفة ووسائط استدعاء الدالة يتم تمريرها إلى الامتداد. يمكنك عادةً تمرير سمات أخرى في كائن السياق لكي يستهلكها رمز الوظيفة. |
post_invocation_app_level |
يتم استدعاؤه مباشرة بعد انتهاء تنفيذ الدالة. سياق الدالة، وسيطات استدعاء الدالة، وكائن إرجاع استدعاء يتم تمريرها إلى الامتداد. يعد هذا التطبيق مكانًا جيدًا للتحقق مما إذا كان تنفيذ خطافات دورة الحياة قد نجح أم لا. |
ملحقات على مستوى الوظيفة
يتم تشغيل ملحق يرث من FuncExtensionBase في مشغل دالة معين.
FuncExtensionBase
يعرض أساليب الفئة المجردة التالية للتطبيقات:
الطريقة | الوصف |
---|---|
__init__ |
الدالة الإنشائية للملحق. يتم استدعاؤه عند تهيئة مثيل ملحق في وظيفة معينة. عند تنفيذ هذه الطريقة المجردة، قد ترغب في قبول معلمة filename وتمريرها إلى أسلوب super().__init__(filename) الأصل لتسجيل الملحق المناسب. |
post_function_load |
يتم استدعاؤه بعد تحميل الوظيفة مباشرة. يتم تمرير اسم الوظيفة ودليل الوظيفة إلى الامتداد. ضع في اعتبارك أن دليل الدالة للقراءة فقط، وأن أي محاولة للكتابة إلى ملف محلي في هذا الدليل تفشل. |
pre_invocation |
يتم استدعاؤها قبل تشغيل الوظيفة مباشرة. سياق الوظيفة ووسائط استدعاء الدالة يتم تمريرها إلى الامتداد. يمكنك عادةً تمرير سمات أخرى في كائن السياق لكي يستهلكها رمز الوظيفة. |
post_invocation |
يتم استدعاؤه مباشرة بعد انتهاء تنفيذ الدالة. سياق الدالة، وسيطات استدعاء الدالة، وكائن إرجاع استدعاء يتم تمريرها إلى الامتداد. يعد هذا التطبيق مكانًا جيدًا للتحقق مما إذا كان تنفيذ خطافات دورة الحياة قد نجح أم لا. |
مشاركة الموارد عبر المنشأ
تدعم Azure Functions مشاركة الموارد عبر المنشأ (CORS). يتم تكوين CORS في المدخل ومن خلال Azure CLI. تنطبق قائمة الأصول المسموح بها لـ CORS على مستوى تطبيق الدالات. مع تمكين CORS، تتضمن Access-Control-Allow-Origin
الاستجابات العنوان. لمزيد من المعلومات، راجع مشاركة الموارد عبر المنشأ.
يتم دعم مشاركة الموارد عبر المنشأ (CORS) بشكل كامل لتطبيقات وظائف Python.
غير متزامنة
بشكل افتراضي، يمكن لمثيل مضيف لـ Python معالجة استدعاء دالة واحدة فقط في كل مرة. وذلك لأن Python هو وقت تشغيل مترابط واحد. بالنسبة لتطبيق دالة يعالج عددًا كبيرًا من أحداث الإدخال/إخراج و/أو ارتباط الإدخال/إخراج، يمكنك تحسين الأداء بشكل ملحوظ عن طريق تشغيل الدالات بشكل غير متزامن. لمزيد من المعلومات، راجع تحسين أداء الإنتاجية لتطبيقات Python في وظائف Azure.
الذاكرة المشتركة (معاينة)
لتحسين معدل النقل، تتيح Azure Functions لعامل لغة Python خارج العملية مشاركة الذاكرة مع عملية مضيف الوظائف. عندما يصطدم تطبيق الدالات بالازدحام، يمكنك تمكين الذاكرة المشتركة عن طريق إضافة إعداد تطبيق يسمى FUNCTIONS_WORKER_SHARED_MEMORY_DATA_TRANSFER_ENABLED بقيمة 1
. مع تمكين الذاكرة المشتركة، يمكنك بعد ذلك استخدام إعداد DOCKER_SHM_SIZE لتعيين الذاكرة المشتركة إلى شيء مثل 268435456
، وهو ما يعادل 256 ميغابايت.
على سبيل المثال، قد تقوم بتمكين الذاكرة المشتركة لتقليل الازدحامات عند استخدام روابط Blob Storage لنقل حمولات أكبر من 1 ميغابايت.
تتوفر هذه الوظيفة فقط لتطبيقات الوظائف التي تعمل في خطط Premium و Dedicated (Azure App Service). لمعرفة المزيد، راجع الذاكرة المشتركة.
المشكلات المعروفة والأسئلة الشائعة
فيما يلي دليلان لاستكشاف الأخطاء وإصلاحها للمشكلات الشائعة:
فيما يلي دليلان لاستكشاف الأخطاء وإصلاحها للمشكلات المعروفة مع نموذج البرمجة v2:
يتم تعقب جميع المشكلات المعروفة وطلبات الميزات في قائمة مشكلات GitHub. إذا واجهت مشكلة ولم تتمكن من العثور على المشكلة في GitHub، فافتح مشكلة جديدة، وقم بتضمين وصف مفصل للمشكلة.
الخطوات التالية
لمزيد من المعلومات، راجع الموارد التالية: