Azure Functions Python 開發人員指南
本指南是使用 Python 開發 Azure Functions 的簡介。 本文假設您已閱讀過 Azure Functions 開發人員指南。
重要
此文章支援 Azure Functions 中適用於 Python 的 v1 和 v2 程式設計模型。 Python v1 模型使用 functions.json 檔案來定義函式,但新的 v2 模型可讓您改用裝飾項目型方法。 這個新方法會產生更簡單的檔案結構,並且更以程式碼為中心。 選擇文章頂端的 v2 選取器,以了解這個新的程式設計模型。
身為 Python 開發人員,您可能也對下列主題感興趣:
- Visual Studio Code:使用 Visual Studio Code 建立您的第一個 Python 應用程式。
- 終端機或命令提示字元:使用 Azure Functions Core Tools 從命令提示字元建立您的第一個 Python 應用程式。
- 範例:在 Learn 範例瀏覽器中檢閱一些現有的 Python 應用程式。
- Visual Studio Code:使用 Visual Studio Code 建立您的第一個 Python 應用程式。
- 終端機或命令提示字元:使用 Azure Functions Core Tools 從命令提示字元建立您的第一個 Python 應用程式。
- 範例:在 Learn 範例瀏覽器中檢閱一些現有的 Python 應用程式。
開發選項
這兩種 Python Functions 程式設計模型都支援下列其中一個環境中的本機開發:
Python v2 程式設計模型:
Python v1 程式設計模型:
您也可以在 Azure 入口網站中建立 Python v1 函式。
提示
雖然您可以在本地 Windows 上開發以 Python 為基礎的 Azure Functions,但只有在 Azure 中執行時,Linux 型主控方案才支援 Python。 如需詳細資訊,請參閱受支援的作業系統/執行階段組合清單。
程式設計模型
Azure Functions 預期函式是 Python 指令碼中的無狀態方法,用來處理輸入及產生輸出。 根據預設,執行階段預期此方法將會實作為全域方法 (在 __init__.py 檔案中稱為 main()
)。 您也可以指定替代進入點。
您可以將來自觸發程序和繫結的資料,透過方法屬性 (其使用 function.json 檔案中定義的 name
屬性) 繫結至函式。 例如,下列 function.json 檔案說明由名為 req
HTTP 要求所觸發的簡單函式:
{
"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 型別註釋,在函式中明確地宣告參數類型並傳回型別。 此動作可協助您使用許多 Python 程式碼編輯器提供的 IntelliSense 和自動完成功能。
import azure.functions
def main(req: azure.functions.HttpRequest) -> str:
user = req.params.get('user')
return f'Hello, {user}!'
請使用 azure.functions.* 套件中所包含的 Python 註釋,以將輸入和輸出繫結至方法。
Azure Functions 預期函式是 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 型別註釋,在函式中明確地宣告參數類型並傳回型別。 此動作可協助您使用許多 Python 程式碼編輯器提供的 IntelliSense 和自動完成功能。
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 模型的已知限制及其因應措施,請參閱針對 Azure Functions中的 Python 錯誤進行疑難排解。
替代進入點
您可以在 function.json 檔案中選擇性地指定 scriptFile
和 entryPoint
屬性,來變更函式的預設行為。 例如,下列 function.json 會告訴執行階段使用在 main.py 檔案中的 customentry()
方法,以作為 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:包含在發佈至 Azure 時系統安裝的 Python 套件清單。
- host.json:包含會對函數應用程式執行個體中所有函式產生影響的設定選項。 此檔案會發行至 Azure。 在本機執行時,不支援所有選項。 若要深入了解,請參閱host.json。
- .vscode/:(選用) 包含已儲存的 Visual Studio Code 設定。 若要深入了解,請參閱 Visual Studio Code 設定。
- .venv/:(選擇性) 包含本地開發所使用的 Python 虛擬環境。
- Dockerfile:(選擇性) 在自訂容器中發佈專案時使用。
- tests/:(選擇性) 包含函數應用程式的測試案例。
- .funcignore:(選擇性) 宣告不應發佈至 Azure 的檔案。 此檔案通常包含 .vscode/ 來忽略您的編輯器設定、.venv/ 來忽略本機 Python 虛擬環境、tests/ 來忽略測試案例,以及 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:(選擇性) 透過藍圖在 function_app.py 中參考的函式 (通常適用於邏輯群組) 包含在內的其他任何 Python 檔案。
- tests/:(選擇性) 包含函數應用程式的測試案例。
- .funcignore:(選擇性) 宣告不應發佈至 Azure 的檔案。 此檔案通常包含 .vscode/ 來忽略您的編輯器設定、.venv/ 來忽略本機 Python 虛擬環境、tests/ 來忽略測試案例,以及 local.settings.json 來防止發佈本機應用程式設定。
- host.json:包含會對函數應用程式執行個體中所有函式產生影響的設定選項。 此檔案會發行至 Azure。 在本機執行時,不支援所有選項。 若要深入了解,請參閱host.json。
- local.settings.json:可在本機執行時用來儲存應用程式設定和連接字串。 此檔案不會發行至 Azure。 若要深入瞭解,請參閱 local.settings.file。
- requirements.txt:包含在發佈至 Azure 時系統安裝的 Python 套件清單。
- Dockerfile:(選擇性) 在自訂容器中發佈專案時使用。
將您的專案部署到 Azure 中的函式應用程式時,主要專案資料夾 (<project_root>) 的全部內容應包含在套件中,但不應包含資料夾本身,意即 host.json 應位於套件根目錄中。 建議您在與其他函式相同的資料夾中保存您的測試 (在此範例中,指的是 tests/)。 如需詳細資訊,請參閱單元測試。
連線至資料庫
Azure Functions 在許多使用案例中均可適當整合 Azure Cosmos DB,包括物聯網、電子商務、遊戲等。
例如,用於事件來源時,可以使用 Azure Cosmos DB 的變更摘要功能來整合這兩個服務以強化事件驅動架構。 變更摘要可讓下游微服務能夠確實累加讀取插入和更新 (例如訂單事件)。 這項功能可以用來提供持續事件存放區,以作為許多微服務 (能夠以無伺服器 Azure Functions 實作) 之間狀態變更事件和訂單處理工作流程的訊息代理程式。
若要連線至 Azure Cosmos DB,請先建立帳戶、資料庫和容器。 然後,您可以使用觸發程序和繫結,將函式程式碼連線到 Cosmos DB,如此範例所示。
若要實作更複雜的應用程式邏輯,您也可以使用適用於 Cosmos DB 的 Python 程式庫。 非同步 I/O 實作看起來如下:
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())
藍圖
Python v2 程式設計模型引進藍圖的概念。 藍圖是具現化的新類別,可在核心函式應用程式外部註冊函式。 在藍圖執行個體中註冊的函式不會由函式執行階段直接編製索引。 若要編製索引這些藍圖函式,函式應用程式必須從藍圖執行個體註冊函式。
使用藍圖提供下列優點:
- 可讓您將函式應用程式分成模組化元件,以便您在多個 Python 檔案中定義函式,並將它們分割成每個檔案的不同元件。
- 提供可延伸的公用函式應用程式介面,以組建及重複使用您自己的 API。
下列範例示範如何使用藍圖:
首先,在 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
形式傳遞至函式。 系統會根據路由 URL 中的 ID 從 Azure Blob 儲存體帳戶擷取輸入,並在函式主體中以 obj
形式提供。 在此處,指定的儲存體帳戶是 CONNECTION_STRING
應用程式設定中找到的連接字串。
在 Azure Functions 中輸入會分成兩個類別:觸發程序輸入和其他輸入。 雖然是使用不同的裝飾項目來定義,但使用方式在 Python 程式碼中很類似。 在本機執行時,觸發程序和輸入來源的連接字串或祕密會對應至 local.settings.json 檔案中的值,但在 Azure 中執行時,則會對應至應用程式設定。
例如,下列程式碼會示範如何定義 Blob 儲存體輸入繫結:
// 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
形式傳遞至函式。 系統會根據路由 URL 中的 ID 從 Azure Blob 儲存體帳戶擷取輸入,並在函式主體中以 obj
形式提供。 在這裡,指定的儲存體帳戶是 STORAGE_CONNECTION_STRING
應用程式設定中找到的連接字串。
針對資料密集的繫結作業,您可能想要使用個別的儲存體帳戶。 如需詳細資訊,請參閱儲存體帳戶指南。
SDK 類型繫結 (預覽)
若要選取觸發程序和繫結,您可以使用基礎 Azure SDK 和架構所實作的資料類型。 這些 SDK 類型繫結可讓您像使用基礎服務 SDK 一樣與繫結資料互動。
重要
SDK 類型繫結支援需要 Python v2 程式設計模型。
Functions 支援適用於 Azure Blob 儲存體的 Python SDK 類型繫結,可讓您使用基礎 BlobClient
類型來處理 Blob 資料。
重要
適用於 Python 的 SDK 類型繫結支援目前處於預覽狀態:
- 您必須使用 Python v2 程式設計模型。
- 目前僅支援同步 SDK 類型。
必要條件
- Azure Functions 執行階段 4.34 版或更新版本。
- Python 3.9 版或更新版本支援的版本。
啟用 Blob 儲存體延伸模組的 SDK 類型繫結
將
azurefunctions-extensions-bindings-blob
延伸模組套件新增至專案中的requirements.txt
檔案,其中至少應包含下列套件:azure-functions azurefunctions-extensions-bindings-blob
將此程式碼新增至專案中的
function_app.py
檔案,以匯入 SDK 類型繫結:import azurefunctions.extensions.bindings.blob as blob
SDK 類型繫結範例
此範例示範如何從 Blob 儲存體觸發程序 (blob_trigger
) 以及 HTTP 觸發程序上的輸入繫結 (blob_input
) 中取得 BlobClient
:
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"
您可以在 Python 延伸模組存放庫中檢視適用於 Blob 儲存體的其他 SDK 類型繫結範例:
HTTP 串流 (預覽)
HTTP 串流可讓您使用函式中啟用的 FastAPI 要求和回應 API,接受和傳回 HTTP 端點的資料。 這些 API 可讓主機以區塊處理 HTTP 訊息中的大型資料,而不是將整個訊息讀入記憶體。
這項功能可讓您處理大型資料流、OpenAI 整合、提供動態內容,以及支援需要透過 HTTP 進行即時互動的其他核心 HTTP 案例。 您也可以搭配 HTTP 串流使用 FastAPI 回應類型。 如果沒有 HTTP 串流,則在整合訊息承載的處理全都在記憶體中時,HTTP 要求和回應的大小會因記憶體限制而受限。
重要
HTTP 串流支援需要 Python v2 程式設計模型。
重要
適用於 Python 的 HTTP 串流支援目前處於預覽狀態,因此您必須使用 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 使用量,也請新增
"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())
輸出
輸出可以使用傳回值和輸出參數來表示。 如果只有一個輸出,我們建議使用傳回的值。 若為多個輸出,您必須使用輸出參數。
若要使用函式的傳回值作為輸出繫結的值,應該將 function.json 檔案中的繫結 name
屬性設定為 $return
。
若要產生多個輸出,請使用 azure.functions.Out
介面提供的 set()
方法為繫結指派值。 例如,下列函式可以將訊息推送至佇列,同時也會傳回 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
輸出可以使用傳回值和輸出參數來表示。 如果只有一個輸出,我們建議使用傳回的值。 若為多個輸出,您必須使用輸出參數。
若要產生多個輸出,請使用 azure.functions.Out
介面提供的 set()
方法為繫結指派值。 例如,下列函式可以將訊息推送至佇列,同時也會傳回 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
記錄
可以透過函式應用程式中的根 logging
處理常式來存取 Azure Functions 執行階段記錄器。 這個記錄器會與 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 Functions。
從已建立的執行緒記錄
若要查看已建立執行緒的記錄,請在函式簽章中包含 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 Functions 執行階段會收集函式產生的記錄和其他遙測資料, 而遙測資料最終會成為 Application Insights 的追蹤。 觸發程序和繫結預設也會收集特定 Azure 服務的要求和相依性遙測資料。
若要收集繫結外部的自訂要求和自訂相依性遙測,您可以使用 OpenCensus Python 延伸模組。 該延伸模組會將自訂遙測資料傳送至 Application Insights 執行個體。 您可以在 OpenCensus 存放庫 (英文) 找到支援的延伸模組清單。
注意
如要使用 OpenCensus Python 延伸模組,您必須將 PYTHON_ENABLE_WORKER_EXTENSIONS
設定為 1
,在函數應用程式中啟用 Python 背景工作角色延伸模組。 如果還沒將APPLICATIONINSIGHTS_CONNECTION_STRING
設定新增至您的應用程式設定,則也需切換至使用 Application Insights 連接字串。
// 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 物件中,您可以取得要求標題、查詢參數、路由參數和訊息內文。
下列範例來自 Python 的 HTTP 觸發程序範本。
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
)
在此函數中,從 HttpRequest 物件的 params
參數取得 name
查詢參數的值。 使用 get_json
方法來讀取 JSON 編碼的訊息內文。
同樣地,您可以在傳回的 HttpResponse 物件中設定回應訊息的 status_code
和 headers
。
HTTP 觸發程序定義為採用具名繫結參數 (HttpRequest 物件) 的方法,並傳回 HttpResponse 物件。 將 function_name
裝飾項目套用至方法來定義函式名稱,而 HTTP 端點則是藉由套用 route
裝飾項目來設定。
此範例來自 Python v2 程式設計模型的 HTTP 觸發程序範本,其中繫結參數名稱是 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 物件中,您可以取得要求標題、查詢參數、路由參數和訊息內文。 在此函數中,從 HttpRequest 物件的 params
參數取得 name
查詢參數的值。 使用 get_json
方法來讀取 JSON 編碼的訊息內文。
同樣地,您可以在傳回的 HttpResponse 物件中設定回應訊息的 status_code
和 headers
。
若要在此範例中傳入名稱,請貼上執行函式時提供的 URL,並附加 "?name={name}"
。
Web 架構
您可以使用與 Web 伺服器閘道介面 (WSGI) 相容的架構,以及與非同步伺服器閘道介面 (ASGI) 相容的架構,例如 Flask 和 FastAPI,搭配 HTTP 觸發的 Python 函式。 本節說明如何修改函式來支援這些架構。
首先必須更新 function.json 檔案,才能在 HTTP 觸發程序中包含 route
,如下列範例所示:
{
"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。 下列範例顯示 Flask 的 ASGI 處理常式方法或 WSGI 包裝函式方法:
您可以使用與非同步伺服器閘道介面 (ASGI) 相容的架構,以及與 Web 伺服器閘道介面 (WSGI) 相容的架構,例如 Flask 和 FastAPI,搭配 HTTP 觸發的 Python 函式。 您必須先更新 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 Functions 支援下列 Python 版本:
Functions 版本 | Python* 版本 |
---|---|
4.x | 3.11 3.10 3.9 3.8 3.7 |
3.x | 3.9 3.8 3.7 |
*官方 Python 發行版本
若要在 Azure 中建立函式應用程式時要求特定的 Python 版本,請使用 az functionapp create
命令的 --runtime-version
選項。 Functions 執行階段版本是由 --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
命令,從 PyPI 安裝 requests
套件。
requests==2.19.1
pip install -r requirements.txt
在 App Service 方案中執行函式時,您在 requirements.txt 中定義的相依性優先於內建 Python 模組,例如 logging
。 當內建模組的名稱與程式碼中的目錄相同時,此優先順序可能會導致衝突。 在使用量方案 或彈性進階方案中執行時,發生衝突的可能性較小,因為您的相依性預設不會設為優先。
若要避免在 App Service 方案中執行時發生問題,請勿將目錄命名為與任何 Python 原生模組相同,也不要在專案的 requirements.txt 檔案中包含 Python 原生程式庫。
發行至 Azure
當您準備發佈時,請確保所有可公開使用的相依性都列在 requirements.txt 檔案中。 您可以在專案目錄的根目錄找到此檔案。
排除在發佈之外的專案檔和資料夾,包括虛擬環境資料夾,您可以在專案的根目錄中找到。
有三個組建動作支援將 Python 專案發佈至 Azure:遠端組建、本地組建以及使用自訂相依性組建。
您也可以使用 Azure Pipelines 組建相依性,並使用持續傳遞 (CD) 加以發佈。 若要深入了解,請參閱設定 Azure Pipelines 的持續傳遞。
遠端組建
當您使用遠端組建時,伺服器上還原的相依性與原生相依性均符合實際執行環境。 因此會上傳較小的部署套件。 當您在 Windows 上開發 Python 應用程式時,請使用遠端組建。 如果您的專案具有自訂相依性,則可以使用遠端組建搭配額外索引 URL。
相依性是根據 requirements.txt 檔案的內容從遠端取得。 遠端組建是建議的組建方法。 根據預設,當您使用下列 func azure functionapp publish
命令,將 Python 專案發佈至 Azure 時,Core Tools 會要求遠端組建。
func azure functionapp publish <APP_NAME>
在 Azure 中,請記得以您的函式應用程式名稱取代 <APP_NAME>
。
根據預設,適用於 Visual Studio Code 的 Azure Functions 擴充功能也會要求遠端組建。
本機組建
系統會根據 requirements.txt 檔案的內容,在本機取得相依性。 您可以使用下列 func azure functionapp publish
命令,搭配本機組建進行發佈,以防止執行遠端組建:
func azure functionapp publish <APP_NAME> --build local
在 Azure 中,請記得以您的函式應用程式名稱取代 <APP_NAME>
。
您使用 --build local
選項時,會從 requirements.txt 檔案讀取專案相依性,以及在本機下載並安裝這些相依套件。 專案檔案和相依性會從您的本機電腦部署至 Azure。 這會導致更大的部署套件上傳至 Azure。 如果您因為一些原因而無法使用 Core Tools 取得 requirements.txt 檔案,則必須使用自訂相依性選項進行發佈。
當您在 Windows 上本機開發時,不建議使用本機組建。
自訂相依性
若 Python 套件索引中找不到您專案的相依性,則可透過兩種方式組建專案。 第一種方式是組建方法,需依組建專案的方式而定。
具額外索引 URL 的遠端組建
若您的套件能從可存取的自訂套件索引中獲取,請使用遠端組建。 發佈前,請務必建立名為 PIP_EXTRA_INDEX_URL
的應用程式設定。 此設定的值是自訂套件索引的 URL。 使用此設定會指示遠端組建使用 --extra-index-url
選項執行 pip install
。 若要深入了解,請參閱 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
在 Azure 中,請記得以您的函式應用程式名稱取代 <APP_NAME>
。
單元測試
以 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()
方法會傳回暫存資料夾,在 Linux 上是 /tmp。 您的應用程式可以使用此目錄,來儲存您函式在執行期間所產生及使用的暫存檔案。
重要
寫入臨時目錄的檔案不保證會在叫用之間持續存在。 在擴增期間,暫存檔案不會在執行個體之間共用。
下列範例會在暫存目錄 (/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 Functions 執行階段會附帶幾個程式庫。
Python 標準程式庫
Python 標準程式庫包含隨附於每個 Python 散發套件的內建 Python 模組清單。 這些程式庫大多能協助您存取系統功能,例如檔案輸入/輸出 (I/O)。 在 Windows 系統上,這些程式庫會伴隨 Python 一同安裝。 在以 Unix 為基礎的系統上,程式庫則會以套件集合的方式提供。
若要檢視 Python 版本的程式庫,請前往:
Azure Functions 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 加以覆寫。 requirements.txt 中的 azure-functions
輸入僅供 Lint 分析與客戶感知之用。
請用下列程式碼追蹤執行階段中 Python Functions 程式庫的實際版本:
getattr(azure.functions, '__version__', '< 1.2.1')
執行階段系統程式庫
如需 Python 背景工作角色 Docker 映射中預先安裝系統程式庫的清單,請參閱下方:
Functions 執行階段 | Debian 版本 | Python 版本 |
---|---|---|
3.x 版 | Buster | Python 3.7 Python 3.8 Python 3.9 (英文) |
Python 背景工作角色延伸模組
在 Azure Functions 中執行的 Python 背景工作處理序可讓您將協力廠商的程式庫整合到函數應用程式中。 這些延伸模組程式庫可作為中介軟體,在函式的執行生命週期內插入特定作業。
延伸模組匯入函式程式碼的方式,跟標準 Python 程式庫模組並無二致。 延伸模組會依據下列範圍執行:
影響範圍 | 描述 |
---|---|
應用程式層級 | 匯入任何函式觸發程序時,延伸模組會套用至應用程式中每個執行的函式。 |
函數層級 | 僅匯入的特定函式觸發程序會套用執行。 |
請檢閱每個延伸模組的資訊,以深入瞭解延伸模組執行的範圍。
延伸模組會實作 Python 背景工作角色延伸模組介面。 此動作可讓 Python 背景工作處理序在函式執行生命週期期間呼叫延伸模組程式碼。 若要深入瞭解,請參閱建立延伸模組。
使用擴充功能
請遵循下列步驟,在 Python 函式中使用 Python 背景工作角色延伸模組程式庫:
- 在專案的 requirements.txt 檔案中新增延伸模組套件。
- 將程式庫安裝至您的應用程式。
- 新增下列應用程式設定:
- 本地:在 local.settings.json 檔案的
Values
區段中輸入"PYTHON_ENABLE_WORKER_EXTENSIONS": "1"
。 - Azure:在 [應用程式設定] 中輸入
PYTHON_ENABLE_WORKER_EXTENSIONS=1
。
- 本地:在 local.settings.json 檔案的
- 將延伸模組匯入您的函式觸發程序。
- 視需要設定延伸模組執行個體。 延伸模組文件應會指明相關設定需求。
重要
Microsoft 不支援或保證協力廠商 Python 背景工作角色延伸模組程式庫。 請確定您在函數應用程式中使用的任何延伸模組都值得信任,且您願意為使用惡意或撰寫不夠周全的延伸模組承擔所有風險。
協力廠商應提供特定文件,說明如何在函數應用程式中安裝並取用其延伸模組。 如需取用延伸模組的基本範例,請參閱取用延伸模組。
以下是依範圍在函數應用程式中使用延伸模組的範例:
# <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 Functions。 延伸模組開發人員會設計、實作並發行 Python 套件,其中包含專為在函式執行內容中執行而設計的自訂邏輯。 這些延伸模組可發佈至 PyPI 登錄或 GitHub 存放庫。
欲了解如何建立、封裝、發佈及取用 Python 背景工作角色延伸模組套件,請參閱開發適用於 Azure Functions 的 Python 背景工作角色延伸模組。
應用等級延伸模組
繼承自 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)。 您可在入口網站中,以及透過 Azure CLI 來設定 CORS。 CORS 允許的原始來源清單適用於函數應用程式層級。 在啟用 CORS 的情況下,回應會包含 Access-Control-Allow-Origin
標頭。 如需詳細資訊,請參閱跨原始資源共用。
Python 函式應用程式完全支援跨原始資源共用 (CORS)。
Async
根據預設,Python 的主機執行個體一次只能處理一個函式的叫用, 因為 Python 屬於單一執行緒的執行階段。 針對處理大量 I/O 事件和/或擁有 I/O 繫結的函數應用程式,您可以非同步方式執行函式,以便大幅提升效能。 如需詳細資訊,請參閱在 Azure Functions 中改善 Python 應用程式的整體效能。
共用記憶體 (預覽版)
為了改善輸送量,Azure Functions 可讓您的跨處理序 Python 語言背景工作角色與 Functions 主機處理序共用記憶體。 若函數應用程式達到瓶頸,則可新增名為 FUNCTIONS_WORKER_SHARED_MEMORY_DATA_TRANSFER_ENABLED 且值為 1
的應用程式設定,藉此啟用共用記憶體。 啟用共用記憶體後,您可用 DOCKER_SHM_SIZE 設定將共用記憶體設定為268435456
之類的數值,相當於 256 MB。
例如,您可在使用 Blob 儲存體繫結傳輸大於 1 MB 的承載時,啟用共用記憶體以減少瓶頸。
這項功能僅適用於在「進階版」和「專用 (Azure App Service) 方案」中執行的函數應用程式。 若要深入了解,請參閱共用記憶體。
已知問題和常見問題集
下列為常見問題的兩個疑難排解指南:
以下是 v2 程式設計模型已知問題的兩個疑難排解指南:
所有已知問題和功能要求則會使用 GitHub 問題清單追蹤。 如果您遇到問題,但在 GitHub 中卻找不到此問題,請開啟新的問題,並包含問題的詳細說明。
下一步
如需詳細資訊,請參閱以下資源: