Azure İşlevleri Python geliştirici kılavuzu

Azure İşlevleri, Python betiğinizde girişi işleyen ve çıkış üreten durum bilgisi olmayan bir yöntem işlevi olmasını bekler. Varsayılan olarak, çalışma zamanı yönteminin __init__.py dosyasında çağrılan main() genel bir yöntem olarak uygulanmasını bekler. Alternatif bir giriş noktası da belirtebilirsiniz.

function.json dosyasında tanımlanan özelliğini kullanan name yöntem öznitelikleri aracılığıyla tetikleyicilerden ve bağlamalardan işleve veri bağlarsınız. Örneğin, aşağıdaki function.json dosyasında adlı reqbir HTTP isteği tarafından tetiklenen basit bir işlev açıklanır:

{
    "scriptFile": "__init__.py",
    "bindings": [
        {
            "authLevel": "function",
            "type": "httpTrigger",
            "direction": "in",
            "name": "req",
            "methods": [
                "get",
                "post"
            ]
        },
        {
            "type": "http",
            "direction": "out",
            "name": "$return"
        }
    ]
}

Bu tanıma bağlı olarak, işlev kodunu içeren __init__.py dosyası aşağıdaki örneğe benzer olabilir:

def main(req):
    user = req.params.get('user')
    return f'Hello, {user}!'

Ayrıca Python türü ek açıklamalarını kullanarak işlevde öznitelik türlerini ve dönüş türünü açıkça bildirebilirsiniz. Bunu yapmak, birçok Python kod düzenleyicisi tarafından sağlanan IntelliSense ve otomatik tamamlama özelliklerini kullanmanıza yardımcı olur.

import azure.functions

def main(req: azure.functions.HttpRequest) -> str:
    user = req.params.get('user')
    return f'Hello, {user}!'

Giriş ve çıkışları yöntemlerinize bağlamak için azure.functions.* paketine dahil edilen Python ek açıklamalarını kullanın.

Azure İşlevleri, Python betiğinizde girişi işleyen ve çıkış üreten durum bilgisi olmayan bir yöntem işlevi olmasını bekler. Çalışma zamanı varsayılan olarak yönteminin function_app.py dosyasında genel bir yöntem olarak uygulanmasını bekler.

Tetikleyiciler ve bağlamalar, dekoratör tabanlı bir yaklaşımda bir işlevde bildirilebilir ve kullanılabilir. Bunlar, işlevlerle aynı dosyada function_app.py tanımlanır. Örneğin, aşağıdaki function_app.py dosyası BIR HTTP isteği tarafından bir işlev tetikleyicisini temsil eder.

@app.function_name(name="HttpTrigger1")
@app.route(route="req")
def main(req):
    user = req.params.get("user")
    return f"Hello, {user}!"

Ayrıca Python türü ek açıklamalarını kullanarak işlevde öznitelik türlerini ve dönüş türünü açıkça bildirebilirsiniz. Bunu yapmak, birçok Python kod düzenleyicisi tarafından sağlanan IntelliSense ve otomatik tamamlama özelliklerini kullanmanıza yardımcı olur.

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 modeliyle ilgili bilinen sınırlamalar ve bunların geçici çözümleri hakkında bilgi edinmek için bkz. Azure İşlevleri Python hatalarını giderme.

İsteğe bağlı olarak function.json dosyasında ve entryPoint özelliklerini belirterek scriptFile bir işlevin varsayılan davranışını değiştirebilirsiniz. Örneğin, aşağıdaki function.json çalışma zamanına azure işlevinizin giriş noktası olarak main.py dosyasındaki yöntemini kullanmasını customentry() söyler.

{
  "scriptFile": "main.py",
  "entryPoint": "customentry",
  "bindings": [
      ...
  ]
}

Giriş noktası yalnızca function_app.py dosyasındadır. Ancak, şemaları kullanarak veya içeri aktararak proje içindeki işlevlere function_app.py başvurabilirsiniz.

Python işlevleri projesi için önerilen klasör yapısı aşağıdaki örneğe benzer:

 <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> ana proje klasörü <aşağıdaki dosyaları içerebilir:

• local.settings.json: Yerel olarak çalışırken uygulama ayarlarını ve bağlantı dizesi depolamak için kullanılır. Bu dosya Azure'da yayımlanmaz. Daha fazla bilgi için bkz . local.settings.file.
• requirements.txt: Azure'da yayımlarken sistemin yüklemiş olduğu Python paketlerinin listesini içerir.
• host.json: İşlev uygulaması örneğindeki tüm işlevleri etkileyen yapılandırma seçeneklerini içerir. Bu dosya Azure'da yayımlanır. Yerel olarak çalıştırılırken tüm seçenekler desteklenmez. Daha fazla bilgi edinmek için bkz . host.json.
• .vscode/: (İsteğe bağlı) Depolanan Visual Studio Code yapılandırmasını içerir. Daha fazla bilgi edinmek için bkz . Visual Studio Code ayarları.
• .venv/: (İsteğe bağlı) Yerel geliştirme tarafından kullanılan bir Python sanal ortamı içerir.
• Dockerfile: (İsteğe bağlı) Projenizi özel bir kapsayıcıda yayımlarken kullanılır.
• tests/: (İsteğe bağlı) İşlev uygulamanızın test çalışmalarını içerir.
• .funcignore: (İsteğe bağlı) Azure'da yayımlanmaması gereken dosyaları bildirir. Bu dosya genellikle düzenleyici ayarınızı yoksaymak için .vscode/, yerel Python sanal ortamını yoksaymak için .venv/, test çalışmalarını yoksaymak için testler/ ve yerel uygulama ayarlarının yayımlanmasını önlemek için local.settings.json içerir.

Her işlevin kendi kod dosyası ve bağlama yapılandırma dosyası function.json.

Python işlevleri projesi için önerilen klasör yapısı aşağıdaki örneğe benzer:

 <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> ana proje klasörü <aşağıdaki dosyaları içerebilir:

• .venv/: (İsteğe bağlı) Yerel geliştirme tarafından kullanılan bir Python sanal ortamı içerir.
• .vscode/: (İsteğe bağlı) Depolanan Visual Studio Code yapılandırmasını içerir. Daha fazla bilgi edinmek için bkz . Visual Studio Code ayarları.
• function_app.py: Tüm işlevler ve bunların ilgili tetikleyicileri ve bağlamaları için varsayılan konum.
• additional_functions.py: (İsteğe bağlı) şemalar aracılığıyla function_app.py başvuruda bulunılan işlevler içeren diğer Python dosyaları (genellikle mantıksal gruplandırma için).
• tests/: (İsteğe bağlı) İşlev uygulamanızın test çalışmalarını içerir.
• .funcignore: (İsteğe bağlı) Azure'da yayımlanmaması gereken dosyaları bildirir. Bu dosya genellikle düzenleyici ayarınızı yoksaymak için .vscode/ içerir, yerel Python sanal ortamını yoksaymak için .venv/ , test çalışmalarını yoksaymak için testler/ ve yerel uygulama ayarlarının yayımlanmasını önlemek için local.settings.json .
• host.json: İşlev uygulaması örneğindeki tüm işlevleri etkileyen yapılandırma seçeneklerini içerir. Bu dosya Azure'da yayımlanır. Yerel olarak çalıştırılırken tüm seçenekler desteklenmez. Daha fazla bilgi edinmek için bkz . host.json.
• local.settings.json: Yerel olarak çalışırken uygulama ayarlarını ve bağlantı dizesi depolamak için kullanılır. Bu dosya Azure'da yayımlanmaz. Daha fazla bilgi için bkz . local.settings.file.
• requirements.txt: Sistemin Azure'da yayımlandığında yüklemiş olduğu Python paketlerinin listesini içerir.
• Dockerfile: (İsteğe bağlı) Projenizi özel bir kapsayıcıda yayımlarken kullanılır.

Blueprints

Python v2 programlama modeli şema kavramını tanıtır. Şema, işlevleri çekirdek işlev uygulamasının dışında kaydetmek için örneği oluşturulmuş yeni bir sınıftır. Şema örneklerine kaydedilen işlevler, işlev çalışma zamanı tarafından doğrudan dizine kaydedilmez. Bu şema işlevlerini dizine almak için işlev uygulamasının şema örneklerinden işlevleri kaydetmesi gerekir.

Şemaları kullanmak aşağıdaki avantajları sağlar:

• İşlev uygulamasını modüler bileşenlere ayırmanıza olanak tanır. Bu sayede birden çok Python dosyasında işlev tanımlayabilir ve bunları dosya başına farklı bileşenlere bölebilirsiniz.
• Kendi API'lerinizi oluşturmak ve yeniden kullanmak için genişletilebilir genel işlev uygulaması arabirimleri sağlar.

Aşağıdaki örnekte şemaların nasıl kullanılacağı gösterilmektedir:

İlk olarak, bir http_blueprint.py dosyasında, http ile tetiklenen bir işlev önce tanımlanır ve şema nesnesine eklenir.

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 
        ) 

Ardından, function_app.py dosyasında şema nesnesi içeri aktarılır ve işlevleri işlev uygulamasına kaydedilir.

import azure.functions as func 
from http_blueprint import bp

app = func.FunctionApp() 

app.register_functions(bp) 

İçeri aktarma davranışı

Hem mutlak hem de göreli başvuruları kullanarak işlev kodunuzda modülleri içeri aktarabilirsiniz. Daha önce açıklanan klasör yapısına bağlı olarak, aşağıdaki içeri aktarmalar project_root\my_first_function>\__init__.py işlev dosyasının <içinden çalışır:

from shared_code import my_first_helper_function #(absolute)

import shared_code.my_second_helper_function #(absolute)

from . import example #(relative)

Aşağıdaki __app__ içeri aktarma ve üst düzey göreli içeri aktarma işlemleri, statik tür denetleyicisi tarafından desteklenmediğinden ve Python test çerçeveleri tarafından desteklenmediğinden kullanım dışıdır:

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)

Girişler Azure İşlevleri iki kategoriye ayrılır: tetikleyici girişi ve diğer girişler. function.json dosyasında farklı olsalar da Python kodunda kullanımları aynıdır. Tetikleyici ve giriş kaynakları için bağlantı dizeleri veya gizli diziler, yerel olarak çalıştırılırken local.settings.json dosyasındaki değerlerle eşler ve Azure'da çalışırken uygulama ayarlarına eşler.

Örneğin, aşağıdaki kod iki giriş arasındaki farkı gösterir:

// 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()}')

İşlev çağrıldığında HTTP isteği işlevine olarak reqgeçirilir. Azure Blob Depolama hesabından yol URL'sindeki kimliğe göre bir giriş alınır ve işlev gövdesinde olduğu gibi obj kullanılabilir hale getirilir. Burada, belirtilen depolama hesabı uygulama ayarında CONNECTION_STRING bulunan bağlantı dizesi.

Girişler Azure İşlevleri iki kategoriye ayrılır: tetikleyici girişi ve diğer girişler. Farklı dekoratörler kullanılarak tanımlanmış olsalar da Python kodunda kullanımları benzerdir. Tetikleyici ve giriş kaynakları için bağlantı dizeleri veya gizli diziler, yerel olarak çalıştırılırken local.settings.json dosyasındaki değerlerle eşler ve Azure'da çalışırken uygulama ayarlarına eşler.

Örneğin aşağıdaki kodda Blob Depolama giriş bağlamasının nasıl tanımlanacağı gösterilmektedir:

// 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()}')

İşlev çağrıldığında HTTP isteği işlevine olarak reqgeçirilir. Azure Blob Depolama hesabından yol URL'sindeki kimliğe göre bir giriş alınır ve işlev gövdesinde olduğu gibi obj kullanılabilir hale getirilir. Burada, belirtilen depolama hesabı uygulama ayarında STORAGE_CONNECTION_STRING bulunan bağlantı dizesi.

İşlevler, temel BlobClient alınan türü kullanarak blob verileriyle çalışmanıza olanak tanıyan Azure Blob depolama için Python SDK türü bağlamalarını destekler.

Önkoşullar

• Azure İşlevleri çalışma zamanı sürümü 4.34 veya sonraki bir sürüm.
• Python sürüm 3.9 veya sonraki bir desteklenen sürüm.

Blob depolama uzantısı için SDK türü bağlamalarını etkinleştirme

1. azurefunctions-extensions-bindings-blob Uzantı paketini projedeki requirements.txt dosyaya ekleyin; en azından şu paketleri içermelidir:

   azure-functions
   azurefunctions-extensions-bindings-blob
   

2. Bu kodu projedeki dosyaya function_app.py ekleyin ve SDK türü bağlamalarını içeri aktarır:

   import azurefunctions.extensions.bindings.blob as blob
   

SDK türü bağlama örnekleri

Bu örnekte BlobClient , hem Blob depolama tetikleyicisinden (blob_trigger) hem de http tetikleyicisi üzerindeki giriş bağlamasından (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"

Python uzantıları deposunda Blob depolama için diğer SDK türü bağlama örneklerini görüntüleyebilirsiniz:

• ContainerClient türü
• StorageStreamDownloader türü

Önkoşullar

• Azure İşlevleri çalışma zamanı sürümü 4.34.1 veya sonraki bir sürüm.
• Python sürüm 3.8 veya sonraki bir desteklenen sürüm.

HTTP akışlarını etkinleştirme

HTTP akışları varsayılan olarak devre dışı bırakılır. Bu özelliği uygulama ayarlarınızda etkinleştirmeniz ve ayrıca kodunuzu FastAPI paketini kullanacak şekilde güncelleştirmeniz gerekir. HTTP akışlarını etkinleştirirken işlev uygulamasının varsayılan olarak HTTP akışını kullanacağını ve özgün HTTP işlevinin çalışmayacağını unutmayın.

1. azurefunctions-extensions-http-fastapi Uzantı paketini projedeki requirements.txt dosyaya ekleyin; en azından şu paketleri içermelidir:

   azure-functions
   azurefunctions-extensions-http-fastapi
   

2. Bu kodu projedeki dosyaya function_app.py ekleyin. Bu kod, FastAPI uzantısını içeri aktarır:

   from azurefunctions.extensions.http.fastapi import Request, StreamingResponse
   

3. Azure'a dağıttığınızda, işlev uygulamanıza aşağıdaki uygulama ayarını ekleyin:

   "PYTHON_ENABLE_INIT_INDEXING": "1"

   Linux Tüketimi'ne dağıtıyorsanız,

   "PYTHON_ISOLATE_WORKER_DEPENDENCIES": "1"

   Yerel olarak çalışırken bu ayarları local.settings.json proje dosyasına da eklemeniz gerekir.

HTTP akışları örnekleri

HTTP akış özelliğini etkinleştirdikten sonra, HTTP üzerinden veri akışı sağlayan işlevler oluşturabilirsiniz.

Bu örnek, HTTP yanıt verilerinin akışını sağlayan HTTP ile tetiklenen bir işlevdir. Gerçek zamanlı görselleştirme için işlem hattı üzerinden olay verileri gönderme veya büyük veri kümelerindeki anomalileri algılama ve anlık bildirimler sağlama gibi senaryoları desteklemek için bu özellikleri kullanabilirsiniz.

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

Bu örnek, bir istemciden gerçek zamanlı olarak akış verilerini alan ve işleyen HTTP ile tetiklenen bir işlevdir. Sürekli veri akışlarını işleme ve IoT cihazlarından olay verilerini işleme gibi senaryolar için yararlı olabilecek akış karşıya yükleme özelliklerini gösterir.

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 akışlarını çağırma

bir işlevin FastAPI uç noktalarına akış çağrıları yapmak için HTTP istemci kitaplığı kullanmanız gerekir. Kullandığınız istemci aracı veya tarayıcı, akışı yerel olarak desteklemeyebilir veya yalnızca ilk veri öbeklerini döndürebilir.

Akış verilerini bir HTTP uç noktasına göndermek için bunun gibi bir istemci betiği kullanabilirsiniz:

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

Çıkış hem dönüş değeri hem de çıkış parametreleriyle ifade edilebilir. Yalnızca bir çıkış varsa, dönüş değerini kullanmanızı öneririz. Birden çok çıkış için çıkış parametrelerini kullanmanız gerekir.

Bir işlevin dönüş değerini çıkış bağlamasının değeri olarak kullanmak için bağlamanın name özelliği function.json dosyasında olarak ayarlanmalıdır $return .

Birden çok çıkış üretmek için, bağlamaya set() azure.functions.Out bir değer atamak için arabirimi tarafından sağlanan yöntemini kullanın. Örneğin, aşağıdaki işlev kuyruğa ileti göndererek bir HTTP yanıtı döndürebilir.

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

Çıkış hem dönüş değeri hem de çıkış parametreleriyle ifade edilebilir. Yalnızca bir çıkış varsa, dönüş değerini kullanmanızı öneririz. Birden çok çıkış için çıkış parametrelerini kullanmanız gerekir.

Birden çok çıkış üretmek için, bağlamaya set() azure.functions.Out bir değer atamak için arabirimi tarafından sağlanan yöntemini kullanın. Örneğin, aşağıdaki işlev kuyruğa ileti göndererek bir HTTP yanıtı döndürebilir.

# 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

HTTP tetikleyicisi function.json dosyasında tanımlanır. name bağlamasının işlevindeki adlandırılmış parametreyle eşleşmesi gerekir. Önceki örneklerde bağlama adı req kullanılmıştır. Bu parametre bir HttpRequest nesnesidir ve bir HttpResponse nesnesi döndürülür.

HttpRequest nesnesinden istek üst bilgilerini, sorgu parametrelerini, yol parametrelerini ve ileti gövdesini alabilirsiniz.

Aşağıdaki örnek Python için HTTP tetikleyici şablonundan alınılmıştır.

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
        )

Bu işlevde sorgu parametresinin name değerini HttpRequest nesnesinin parametresinden params alırsınız. yöntemini kullanarak JSON ile kodlanmış ileti gövdesini get_json okursunuz.

Benzer şekilde, döndürülen HttpResponse nesnesinde yanıt iletisi için ve headers değerini ayarlayabilirsinizstatus_code.

HTTP tetikleyicisi, httpRequest nesnesi olan adlandırılmış bağlama parametresini alan ve bir HttpResponse nesnesi döndüren bir yöntem olarak tanımlanır. function_name İşlev adını tanımlamak için yöntemine dekoratör uygularken, HTTP uç noktası dekoratör uygulanarak route ayarlanır.

Bu örnek Python v2 programlama modelinin HTTP tetikleyici şablonundan alınmıştı ve burada bağlama parametresi adı şeklindedir req. bu, Azure İşlevleri Core Tools veya Visual Studio Code kullanarak işlev oluşturduğunuzda sağlanan örnek koddır.

@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 nesnesinden istek üst bilgilerini, sorgu parametrelerini, yol parametrelerini ve ileti gövdesini alabilirsiniz. Bu işlevde sorgu parametresinin name değerini HttpRequest nesnesinin parametresinden params alırsınız. yöntemini kullanarak JSON ile kodlanmış ileti gövdesini get_json okursunuz.

Benzer şekilde, döndürülen HttpResponse nesnesinde yanıt iletisi için ve headers değerini ayarlayabilirsinizstatus_code.

Bu örnekte bir ad geçirmek için, işlevi çalıştırırken sağlanan URL'yi yapıştırın ve ile ekleyin "?name={name}".

Http ile tetiklenen Python işlevlerinizle Flask ve FastAPI gibi Web Sunucusu Ağ Geçidi Arabirimi (WSGI) uyumlu ve Zaman Uyumsuz Sunucu Ağ Geçidi Arabirimi (ASGI) uyumlu çerçeveleri kullanabilirsiniz. Bu bölümde, işlevlerinizi bu çerçeveleri destekleyecek şekilde nasıl değiştireceğiniz gösterilmektedir.

İlk olarak, aşağıdaki örnekte gösterildiği gibi function.json dosyası HTTP tetikleyicisine eklenecek route şekilde güncelleştirilmelidir:

{
  "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 dosyası, aşağıdaki örnekte gösterildiği gibi http routePrefixiçerecek şekilde de güncelleştirilmelidir:

{
  "version": "2.0",
  "logging": 
  {
    "applicationInsights": 
    {
      "samplingSettings": 
      {
        "isEnabled": true,
        "excludedTypes": "Request"
      }
    }
  },
  "extensionBundle": 
  {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[3.*, 4.0.0)"
  },
  "extensions": 
  {
    "http": 
    {
        "routePrefix": ""
    }
  }
}

Çerçeveniz tarafından kullanılan arabirime bağlı olarak Python kod dosyasını init.py güncelleştirin. Aşağıdaki örnekte Flask için ASGI işleyicisi yaklaşımı veya WSGI sarmalayıcı yaklaşımı gösterilmektedir:

app = fastapi.FastAPI()

@app.get("hello/{name}")
async def get_name(name: str):
  return {"name": name}

def main(req: func.HttpRequest, context: func.Context) -> func.HttpResponse:
    return func.AsgiMiddleware(app).handle(req, context)

app = Flask("Test")

@app.route("hello/<name>", methods=["GET"])
def hello(name: str):
    return f"hello {name}"

def main(req: func.HttpRequest, context) -> func.HttpResponse:
  logging.info("Python HTTP trigger function processed a request.")
  return func.WsgiMiddleware(app).handle(req, context)

HTTP ile tetiklenen Python işlevlerinizle Flask ve FastAPI gibi Zaman Uyumsuz Sunucu Ağ Geçidi Arabirimi (ASGI) uyumlu ve Web Sunucusu Ağ Geçidi Arabirimi (WSGI) uyumlu çerçeveler kullanabilirsiniz. Aşağıdaki örnekte gösterildiği gibi, önce host.json dosyasını http routePrefixiçerecek şekilde güncelleştirmeniz gerekir:

{
  "version": "2.0",
  "logging": 
  {
    "applicationInsights": 
    {
      "samplingSettings": 
      {
        "isEnabled": true,
        "excludedTypes": "Request"
      }
    }
  },
  "extensionBundle": 
  {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[2.*, 3.0.0)"
  },
  "extensions": 
  {
    "http": 
    {
        "routePrefix": ""
    }
  }
}

Çerçeve kodu aşağıdaki örneğe benzer:

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

# function_app.py

import azure.functions as func 
from flask import Flask, Response 

flask_app = Flask(__name__) 

@flask_app.get("/return_http") 
def return_http(): 
    return Response("<h1>Hello World™</h1>", mimetype="text/html") 

app = func.WsgiFunctionApp(app=flask_app.wsgi_app, 
                           http_auth_level=func.AuthLevel.ANONYMOUS) 

Azure İşlevleri,hizmet bağlantı dizesi gibi uygulama ayarları çalışırken ortam değişkenleri olarak kullanıma sunulur. Kodunuzda bu ayarlara erişmenin iki ana yolu vardır.

Bu yolların her ikisi de bildirmenizi import osgerektirir.

Aşağıdaki örnek, adlı myAppSettinganahtarla uygulama ayarını almak için kullanıros.environ["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}')

Yerel geliştirme için uygulama ayarları local.settings.json dosyasında tutulur.

Azure İşlevleri,hizmet bağlantı dizesi gibi uygulama ayarları çalışırken ortam değişkenleri olarak kullanıma sunulur. Kodunuzda bu ayarlara erişmenin iki ana yolu vardır.

Bu yolların her ikisi de bildirmenizi import osgerektirir.

Aşağıdaki örnek, adlı myAppSettinganahtarla uygulama ayarını almak için kullanıros.environ["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}')

Yerel geliştirme için uygulama ayarları local.settings.json dosyasında tutulur.

İlk olarak bir <project_root>/my_second_function/function.json dosyası oluşturun ve ardından bu işlevi HTTP tetikleyicisi olarak tanımlayın.

{
  "scriptFile": "__init__.py",
  "entryPoint": "main",
  "bindings": [
    {
      "authLevel": "function",
      "type": "httpTrigger",
      "direction": "in",
      "name": "req",
      "methods": [
        "get",
        "post"
      ]
    },
    {
      "type": "http",
      "direction": "out",
      "name": "$return"
    }
  ]
}

Ardından ve shared_code.my_second_helper_functionuygulayabilirsinizmy_second_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 tetikleyiciniz için test çalışmaları yazmaya başlayabilirsiniz.

# <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',)

Python sanal ortamı klasörünüzde .venv , gibi pip install pytestsık kullandığınız Python test çerçevesini yükleyin. Ardından test sonucunu denetlemek için komutunu çalıştırın pytest tests .

İlk olarak, project_root/function_app.py dosyasını oluşturun <ve işlevini HTTP tetikleyicisi ve shared_code.my_second_helper_functionolarak uygulayınmy_second_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 tetikleyiciniz için test çalışmaları yazmaya başlayabilirsiniz.

# <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 sanal ortam klasörünüzün içine, gibi pip install pytestsık kullandığınız Python test çerçevesini yükleyin. Ardından test sonucunu denetlemek için komutunu çalıştırın pytest tests .