Azure Functions Python 개발자 가이드
이 가이드에서는 Python을 사용하여 Azure Functions를 개발하는 방법을 소개합니다. 이 문서에서는 Azure Functions 개발자 가이드를 이미 읽었다고 가정합니다.
Important
이 문서는 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 Portal에서 Python v1 함수를 만들 수도 있습니다.
팁
Windows에서 로컬로 Python 기반 Azure 함수를 개발할 수 있지만 Python은 Azure에서 실행될 때 Linux 기반 호스팅 플랜에서만 지원됩니다. 자세한 내용은 지원되는 운영 체제/런타임 조합 목록을 참조하세요.
프로그래밍 모델
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/, 로컬 Python 가상 환경을 무시하는 .venv/, 테스트 사례를 무시하는 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/, 로컬 Python 가상 환경을 무시하는 .venv/, 테스트 사례를 무시하는 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는 IoT, 전자상거래, 게임 등을 포함한 다양한 사용 사례에서 Azure Cosmos DB와 원활하게 통합됩니다.
예를 들어, 이벤트 소싱의 경우 두 서비스가 통합되어 Azure Cosmos DB의 변경 피드 기능을 사용하여 이벤트 기반 아키텍처를 구동합니다. 변경 피드는 다운스트림 마이크로 서비스에 삽입 및 업데이트(예: 주문 이벤트)를 증분 방식을 통해 안정적으로 읽을 수 있는 기능을 제공합니다. 이 기능은 상태 변경 이벤트를 위한 메시지 브로커로 영구 이벤트 저장소를 제공하고 많은 마이크로 서비스 간에 주문 처리 워크플로를 구동하는 데 사용할 수 있습니다( 서버리스 Azure Functions구현될 수 있습니다).
Azure Cosmos DB에 연결하려면 먼저 계정, 데이터베이스 및 컨테이너를 만듭니다. 그런 다음 트리거 및 바인딩을 사용하여 함수 코드를 Azure 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())
Blueprints
Python v2 프로그래밍 모델은 청사진의 개념을 소개합니다. 청사진은 핵심 함수 애플리케이션 외부에서 함수를 등록하기 위해 인스턴스화된 새 클래스입니다. 청사진 인스턴스에 등록된 함수는 함수 런타임에 의해 직접 인덱싱되지 않습니다. 이러한 청사진 함수를 인덱싱하려면 함수 앱이 청사진 인스턴스에서 함수를 등록해야 합니다.
청사진을 사용하면 다음과 같은 이점이 있습니다.
- 함수 앱을 모듈식 구성 요소로 분할하여 여러 Python 파일에서 함수를 정의하고 파일당 다른 구성 요소로 나눌 수 있습니다.
- 사용자 고유의 API를 빌드하고 다시 사용할 수 있는 확장 가능한 공용 함수 앱 인터페이스를 제공합니다.
다음 예제에서는 청사진을 사용하는 방법을 보여 줍니다.
먼저 HTTP 트리거 함수가 http_blueprint.py 파일에서 먼저 정의되고 청사진 개체에 추가됩니다.
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/ 폴더에 python 패키지로 표시하기 위해 __init__.py 파일이 포함되어야 합니다.
다음 __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 Storage 계정에서 검색되고 함수 본문에서 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로 함수에 전달됩니다. 항목은 경로 URL의 ID에 따라 Azure Blob Storage 계정에서 검색되고 함수 본문에서 obj(으)로 사용할 수 있습니다. 여기에서 지정된 스토리지 계정은 STORAGE_CONNECTION_STRING 앱 설정에서 찾은 연결 문자열입니다.
데이터 집약적 바인딩 작업의 경우 별도의 스토리지 계정을 사용할 수 있습니다. 자세한 내용은 스토리지 계정 지침을 참조하세요.
SDK 형식 바인딩(미리 보기)
선택한 트리거 및 바인딩의 경우 기본 Azure SDK 및 프레임워크에서 구현하는 데이터 형식을 사용할 수 있습니다. 이러한 SDK 형식 바인딩을은 기본 서비스 SDK를 사용하는 것처럼 바인딩 데이터를 상호 작용할 수 있습니다.
Important
SDK 형식 바인딩을 지원하려면 Python v2 프로그래밍 모델이 필요합니다.
Functions는 기본 BlobClient 형식을 사용하여 Blob 데이터를 사용할 수 있는 Azure Blob Storage에 대한 Python SDK 형식 바인딩을 지원합니다.
Important
Python에 대한 SDK 형식 바인딩 지원은 현재 미리 보기로 제공됩니다.
- Python v2 프로그래밍 모델을 사용해야 합니다.
- 현재 동기 SDK 형식만 지원됩니다.
필수 조건
- Azure Functions 런타임 버전 버전 4.34 이상 버전.
- Python 버전 3.9 이상에서 지원되는 버전.
Blob Storage 확장에 SDK 형식 바인딩 사용
프로젝트의
requirements.txt파일에azurefunctions-extensions-bindings-blob확장 패키지를 추가합니다. 이 파일은 적어도 다음 패키지를 포함해야 합니다.azure-functions azurefunctions-extensions-bindings-blobSDK 형식 바인딩을 가져오는 프로젝트의
function_app.py파일에 이 코드를 추가합니다.import azurefunctions.extensions.bindings.blob as blob
SDK 형식 바인딩 예제
이 예제에서는 Blob Storage 트리거(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 Storage에 대한 다른 SDK 형식 바인딩 샘플을 볼 수 있습니다.
HTTP 스트림(미리 보기)
HTTP 스트림을 사용하면 함수에서 사용하도록 설정된 FastAPI 요청 및 응답 API를 사용하여 HTTP 엔드포인트에서 데이터를 수락하고 반환할 수 있습니다. 이러한 API를 사용하면 호스트가 전체 메시지를 메모리로 읽는 대신 HTTP 메시지의 큰 데이터를 청크로 처리할 수 있습니다.
이 기능을 사용하면 대규모 데이터 스트림, OpenAI 통합을 처리하고, 동적 콘텐츠를 제공하고, HTTP를 통한 실시간 상호 작용이 필요한 다른 핵심 HTTP 시나리오를 지원할 수 있습니다. HTTP 스트림에서 FastAPI 응답 형식을 사용할 수도 있습니다. HTTP 스트림이 없으면 전체 메시지 페이로드를 처리할 때 발생할 수 있는 메모리 제한으로 인해 HTTP 요청 및 응답의 크기가 제한됩니다.
Important
HTTP 스트림을 지원하려면 Python v2 프로그래밍 모델이 필요합니다.
Important
Python에 대한 HTTP 스트림 지원은 현재 미리 보기 상태이며 Python v2 프로그래밍 모델을 사용해야 합니다.
필수 조건
- Azure Functions 런타임 버전 4.34.1 이상 버전.
- Python 버전 3.8 이상에서 지원되는 버전.
HTTP 스트림 사용
HTTP 스트림은 기본적으로 사용하지 않도록 설정됩니다. 애플리케이션 설정에서 이 기능을 사용하도록 설정하고 FastAPI 패키지를 사용하도록 코드를 업데이트해야 합니다. HTTP 스트림을 사용하도록 설정하면 함수 앱은 기본적으로 HTTP 스트리밍을 사용하게 되며 원래 HTTP 기능은 작동하지 않습니다.
프로젝트의
requirements.txt파일에azurefunctions-extensions-http-fastapi확장 패키지를 추가합니다. 이 파일은 적어도 다음 패키지를 포함해야 합니다.azure-functions azurefunctions-extensions-http-fastapiFastAPI 확장을 가져오는 프로젝트의
function_app.py파일에 다음 코드를 추가합니다.from azurefunctions.extensions.http.fastapi import Request, StreamingResponseAzure에 배포할 때 함수 앱에 다음 애플리케이션 설정을 추가합니다.
"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())
출력
출력은 반환 값 및 출력 매개 변수 둘 다로 표현될 수 있습니다. 출력이 하나만 있는 경우 반환 값을 사용하는 것이 좋습니다. 여러 출력의 경우 출력 매개 변수를 사용해야 합니다.
함수의 반환 값을 출력 바인딩의 값으로 사용하려면 바인딩의 name 속성을 function.json의 $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_) |
루트 로거에 위험 수준의 메시지를 기록합니다. |
error(_message_) |
루트 로거에 오류 수준의 메시지를 기록합니다. |
warning(_message_) |
루트 로거에 경고 수준의 메시지를 기록합니다. |
info(_message_) |
루트 로거에 정보 수준의 메시지를 기록합니다. |
debug(_message_) |
루트 로거에 디버그 수준의 메시지를 기록합니다. |
로깅에 대한 자세한 내용은 Azure Functions 모니터링을 참조하세요.
생성된 스레드에서 로깅
생성된 스레드에서 들어오는 로그를 보려면 함수의 서명에 context 인수를 포함합니다. 이 인수에는 로컬 invocation_id에 저장하는 특성 thread_local_storage(이)가 포함되어 있습니다. 컨텍스트가 변경되도록 함수의 현재 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.')
로그 사용자 지정 원격 분석
기본적으로 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
)
이 함수에서 name 쿼리 매개 변수의 값은 HttpRequest 개체의 params 매개 변수에서 가져옵니다. get_json 메서드를 사용하여 JSON으로 인코딩된 메시지 본문을 읽습니다.
마찬가지로, 반환된 HttpResponse 개체의 응답 메시지에 대한 status_code 및 headers를 설정할 수 있습니다.
HTTP 트리거는 명명된 바인딩 매개 변수(HttpRequest 개체)를 사용하고 HttpResponse 개체를 반환하는 메서드로 정의됩니다. 메서드에 function_name 데코레이터를 적용하여 함수 이름을 정의하고, route 데코레이터를 적용하여 HTTP 엔드포인트를 설정합니다.
이 예제는 바인딩 매개 변수 이름이 req인 Python v2 프로그래밍 모델에 대한 HTTP 트리거 템플릿에서 가져온 것입니다. 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 쿼리 매개 변수의 값은 HttpRequest 개체의 params 매개 변수에서 가져옵니다. get_json 메서드를 사용하여 JSON으로 인코딩된 메시지 본문을 읽습니다.
마찬가지로, 반환된 HttpResponse 개체의 응답 메시지에 대한 status_code 및 headers를 설정할 수 있습니다.
이 예제에서 이름을 전달하려면 함수를 실행할 때 제공된 URL을 붙여넣고 "?name={name}"으로 추가합니다.
웹 프레임워크
HTTP 트리거 Python 함수와 함께 WSGI(Web Server Gateway Interface) 호환 및 ASGI(비동기 서버 게이트웨이 인터페이스) 호환 프레임워크(예: Flask 및 FastAPI)를 사용할 수 있습니다. 이 섹션에서는 이러한 프레임워크를 지원하도록 함수를 수정하는 방법을 보여 줍니다.
먼저 다음 예와 같이 HTTP 트리거에 route를 포함하도록 function.json 파일을 업데이트해야 합니다.
{
"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 래퍼 방법을 보여 줍니다.
HTTP 트리거 Python 함수와 함께 ASGI(비동기 서버 게이트웨이 인터페이스) 호환 및 WSGI(웹 서버 게이트웨이 인터페이스) 호환 프레임워크(예: Flask 및 FastAPI)를 사용할 수 있습니다. 다음 예제와 같이 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
실행 중에 함수의 호출 컨텍스트를 가져오려면 해당 서명에 context 인수를 포함해야 합니다.
예시:
import azure.functions
def main(req: azure.functions.HttpRequest,
context: azure.functions.Context) -> str:
return f'{context.invocation_id}'
Context 클래스에는 다음과 같은 문자열 특성이 있습니다.
| attribute | 설명 |
|---|---|
function_directory |
함수가 실행되는 디렉터리입니다. |
function_name |
함수의 이름. |
invocation_id |
현재 함수 호출의 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을 선언해야 합니다.
다음 예제에서는 myAppSetting라는 키가 있는 애플리케이션 설정을 가져오기 위해 os.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}')
로컬에서 개발하는 경우 애플리케이션 설정은 local.settings.json 파일에 유지됩니다.
Azure Functions에서 서비스 연결 문자열 같은 애플리케이션 설정은 실행 중에 환경 변수로 공개됩니다. 코드에서 이러한 설정에 액세스하는 두 가지 주요 방법이 있습니다.
| 메서드 | 설명 |
|---|---|
os.environ["myAppSetting"] |
키 이름으로 애플리케이션 설정을 설정하려고 시도하여 실패할 때 오류가 발생합니다. |
os.getenv("myAppSetting") |
키 이름으로 애플리케이션 설정을 구하려고 시도하여 실패하면, None을 반환합니다. |
이러한 두 가지 방법 모두 import os을 선언해야 합니다.
다음 예제에서는 myAppSetting라는 키가 있는 애플리케이션 설정을 가져오기 위해 os.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}')
로컬에서 개발하는 경우 애플리케이션 설정은 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 3.8을 사용하도록 Python 앱을 변경하려면 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에 정의한 종속성은 logging과 같은 기본 제공 Python 모듈보다 우선적으로 적용됩니다. 이 우선 순위는 기본 제공 모듈의 이름이 코드의 디렉터리 이름과 같을 때 충돌을 일으킬 수 있습니다. 사용 계획 또는 Elastic Premium 계획에서 실행하는 경우 종속성이 기본적으로 우선 순위가 지정되지 않으므로 충돌이 발생할 가능성이 적습니다.
App Service 요금제에서 실행되는 문제를 방지하려면 디렉터리의 이름을 Python 네이티브 모듈과 동일하게 지정하지 말고 프로젝트의 requirements.txt 파일에 Python 네이티브 라이브러리를 포함하지 마세요.
Azure에 게시
게시할 준비가 되면 공개적으로 제공되는 모든 종속성이 requirements.txt 파일에 나열되는지 확인합니다. 프로젝트 디렉터리의 루트에서 이 파일을 찾을 수 있습니다.
가상 환경 폴더를 비롯하여 게시에서 제외되는 프로젝트 파일과 폴더는 프로젝트의 루트 디렉터리에 있습니다.
Python 프로젝트를 Azure에 게시하는 데 지원되는 세 가지 빌드 작업은 원격 빌드, 로컬 빌드 및 사용자 지정 종속성을 사용하는 빌드입니다.
또한 Azure Pipelines를 사용하여 종속성을 빌드하고 CD(지속적인 업데이트)를 통해 게시할 수 있습니다. 자세히 알아보려면 Azure Pipelines를 사용한 지속적인 배달을 참조하세요.
원격 빌드
원격 빌드를 사용하는 경우 서버에서 복원된 종속성과 기본 종속성은 프로덕션 환경과 일치합니다. 따라서 업로드할 배포 패키지가 작아질 수 있습니다. Windows에서 Python 앱을 개발할 때 원격 빌드를 사용합니다. 프로젝트에 사용자 지정 종속성이 있는 경우 추가 인덱스 URL로 원격 빌드를 사용할 수 있습니다.
requirements.txt 파일의 내용에 따라 원격으로 종속성을 가져옵니다. 원격 빌드는 권장 빌드 방법입니다. 기본적으로 Core Tools는 다음과 같은 func azure functionapp publish 명령을 사용하여 Python 프로젝트를 Azure에 게시할 때 원격 빌드를 요청합니다.
func azure functionapp publish <APP_NAME>
<APP_NAME>을 Azure의 함수 앱 이름으로 바꾸어야 합니다.
Visual Studio Code용 Azure Functions 확장 역시 기본적으로 원격 빌드를 요청합니다.
로컬 빌드
requirements.txt 파일의 내용에 따라 로컬로 종속성을 가져옵니다. 다음 func azure functionapp publish 명령을 사용하여 로컬 빌드로 게시하면 원격 빌드를 방지할 수 있습니다.
func azure functionapp publish <APP_NAME> --build local
<APP_NAME>을 Azure의 함수 앱 이름으로 바꾸어야 합니다.
--build local 옵션을 사용하면 requirements.txt 파일에서 프로젝트 종속성을 읽고 해당하는 종속 패키지를 로컬로 다운로드하여 설치합니다. 프로젝트 파일과 종속성은 로컬 컴퓨터에서 Azure로 배포됩니다. 결과적으로 더 큰 배포 패키지가 Azure에 업로드됩니다. 어떤 이유로든 Core Tools를 사용하여 requirements.txt 파일을 가져올 수 없는 경우 게시에 사용자 지정 종속성 옵션을 사용해야 합니다.
Windows에서 로컬로 개발할 때는 로컬 빌드를 사용하지 않는 것이 좋습니다.
사용자 지정 종속성
Python 패키지 인덱스에 없는 종속성이 프로젝트에 있는 경우 두 가지 방법으로 프로젝트를 빌드할 수 있습니다. 첫 번째 방법인 build 메서드는 프로젝트를 빌드하는 방법에 따라 달라집니다.
추가 인덱스 URL을 사용하여 원격 빌드
액세스할 수 있는 사용자 지정 패키지 인덱스에서 패키지를 사용할 수 있는 경우 원격 빌드를 사용합니다. 게시하기 전에 PIP_EXTRA_INDEX_URL이라는 앱 설정을 만들어야 합니다. 이 설정값은 사용자 지정 패키지 인덱스의 URL입니다. 이 설정을 사용하면 원격 빌드가 --extra-index-url 옵션을 사용하여 pip install을 실행하도록 지시합니다. 자세한 내용은 Python pip 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() 메서드는 임시 폴더를 반환하며, 이 폴더는 Linux에서 /tmp입니다. 애플리케이션에서는 함수가 실행 중에 만들어서 사용하는 임시 파일을 이 디렉터리에 저장할 수 있습니다.
Important
임시 디렉터리에 기록된 파일은 호출 간에 유지되지 않을 수 있습니다. 스케일 아웃 도중에 임시 파일은 인스턴스 간에 공유되지 않습니다.
다음은 임시 디렉터리(/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 모듈 목록이 포함되어 있습니다. 이러한 라이브러리는 대부분 파일 I/O(입/출력)와 같은 시스템 기능에 액세스하는 데 유용합니다. Windows 시스템에서 이러한 라이브러리는 Python과 함께 설치됩니다. Unix 기반 시스템에서는 패키지 컬렉션으로 제공됩니다.
Python 버전의 라이브러리를 보려면 다음으로 이동합니다.
Azure Functions Python 작업자 종속성
Azure Functions Python 작업자에는 특정 라이브러리 집합이 필요합니다. 함수에서 이러한 라이브러리를 사용할 수도 있지만 Python 표준의 일부가 아닙니다. 함수가 이러한 라이브러리에 의존하는 경우 Azure Functions 외부에서 실행할 때 코드에서 사용하지 못할 수 있습니다.
참고 항목
함수 앱의 requirements.txt 파일에 azure-functions-worker 항목이 포함되어 있으면 제거합니다. Functions 작업자는 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 항목은 린팅 및 고객 인식에만 사용됩니다.
런타임에 Python 함수 라이브러리의 실제 버전을 추적하려면 다음 코드를 사용합니다.
getattr(azure.functions, '__version__', '< 1.2.1')
런타임 시스템 라이브러리
Python 작업자 Docker 이미지에 사전 설치된 시스템 라이브러리 목록을 보려면 다음을 참조하세요.
| 함수 런타임 | 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 파일의
- 확장 모듈을 함수 트리거로 가져옵니다.
- 필요한 경우 확장 인스턴스를 구성합니다. 확장 설명서에서 구성 요구 사항을 설명해야 합니다.
Important
타사 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 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(원본 간 리소스 공유)를 지원합니다. CORS는 포털 및 Azure CLI를 통해 구성됩니다. CORS에 허용된 원본 목록은 함수 앱 수준에서 적용됩니다. CORS를 사용하면 응답에 Access-Control-Allow-Origin 헤더가 포함됩니다. 자세한 내용은 크로스-원본 자원 공유(CORS)를 참조하십시오.
CORS(원본 간 리소스 공유)는 Python 함수 앱에 대해 완전히 지원됩니다.
Async
기본적으로 Python용 호스트 인스턴스는 한 번에 하나의 함수 호출만 처리할 수 있습니다. 이는 Python이 단일 스레드 런타임이기 때문입니다. 많은 수의 I/O 이벤트를 처리하거나 I/O 바운드되는 함수 앱의 경우 함수를 비동기식으로 실행하여 성능을 크게 개선시킬 수 있습니다. 자세한 내용은 Azure Functions에서 Python 앱의 전체 성능 개선을 참조하세요.
공유 메모리(미리 보기)
처리량을 개선하기 위해 Azure Functions는 out-of-process Python 언어 작업자가 Functions 호스트 프로세스와 메모리를 공유할 수 있도록 합니다. 함수 앱에 병목 현상이 발생하면 값이 1인 FUNCTIONS_WORKER_SHARED_MEMORY_DATA_TRANSFER_ENABLED라는 애플리케이션 설정을 추가하여 공유 메모리를 사용하도록 설정할 수 있습니다. 공유 메모리가 사용하도록 설정된 상태에서 DOCKER_SHM_SIZE 설정을 사용하여 공유 메모리를 268435456(예: 256MB)으로 설정할 수 있습니다.
예를 들어 Blob Storage 바인딩을 사용하여 1MB보다 큰 페이로드를 전송할 때 병목 현상을 줄이기 위해 공유 메모리를 사용하도록 설정할 수 있습니다.
이 기능은 프리미엄 및 전용(Azure App Service) 플랜에서 실행되는 기능 앱에만 사용할 수 있습니다. 자세한 내용은 공유 메모리를 참조하세요.
알려진 문제 및 FAQ
일반적인 문제에 대한 두 가지 문제 해결 가이드는 다음과 같습니다.
다음은 v2 프로그래밍 모델의 알려진 문제에 대한 문제 해결 가이드입니다.
모든 알려진 문제 및 기능 요청은 GitHub 문제 목록에서 추적됩니다. 문제가 발생하여 GitHub에서 해당 문제를 찾을 수 없는 경우 새 문제를 열고 해당 문제에 대한 자세한 설명을 제공해 주세요.
다음 단계
자세한 내용은 다음 리소스를 참조하세요.