### Initialize and Configure Local Development Environment Source: https://github.com/googlecloudplatform/functions-framework-python/blob/main/examples/skaffold/README.md Commands to verify installations, start the Minikube cluster, and enable the ingress addon to handle external traffic for the deployed functions. ```bash minikube version minikube start minikube addons enable ingress ``` -------------------------------- ### Install Functions Framework Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt Instructions for installing the Functions Framework library using pip or adding it to a requirements file for deployment. ```bash pip install functions-framework ``` ```text functions-framework==3.* ``` -------------------------------- ### Install tox for Environment Setup Source: https://github.com/googlecloudplatform/functions-framework-python/blob/main/DEVELOPMENT.md Installs tox, a command-line tool for managing and automating testing in Python projects. This is a prerequisite for running linters and tests. ```bash python -m pip install -U tox ``` -------------------------------- ### Run Functions Framework via CLI Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt Basic command-line usage for starting a local development server for a specified function target. ```bash functions-framework --target=hello ``` -------------------------------- ### Dockerfile for Python Function Source: https://github.com/googlecloudplatform/functions-framework-python/blob/main/examples/cloud_run_http/README.md Defines the environment for building a Docker container. It installs Python, copies the application code, installs dependencies (functions-framework and requirements.txt), and sets the command to run the function. ```dockerfile # Use the official Python image. # https://hub.docker.com/_/python FROM python:3.7-slim # Copy local code to the container image. ENV APP_HOME /app WORKDIR $APP_HOME COPY . . # Install production dependencies. RUN pip install functions-framework RUN pip install -r requirements.txt # Run the web service on container startup. CMD exec functions-framework --target=hello ``` -------------------------------- ### Deploy and Run Functions with Skaffold Source: https://github.com/googlecloudplatform/functions-framework-python/blob/main/examples/skaffold/README.md Commands to verify the Skaffold installation and initiate the development loop, which watches for code changes and automatically redeploys the functions to the cluster. ```bash skaffold version skaffold dev ``` -------------------------------- ### Background Functions (Pub/Sub and GCS) Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt Examples of background functions triggered by Pub/Sub messages and Cloud Storage events. ```APIDOC ## Background Functions This section covers background functions triggered by events like Pub/Sub messages or Cloud Storage changes. ### Pub/Sub Triggered Function #### Description A background function triggered by Pub/Sub messages. It accesses event data and context metadata, decodes the Pub/Sub message, and prints attributes. ### Method Event-driven (triggered by Pub/Sub) ### Endpoint Not applicable (event-driven) ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body - **event** (dict) - Dictionary containing the Pub/Sub message data. - **context** (Context) - Context object with event metadata. ### Request Example ```json { "context": { "eventId": "1234567890", "timestamp": "2024-01-15T12:00:00Z", "eventType": "google.pubsub.topic.publish", "resource": {"name": "projects/my-project/topics/my-topic"} }, "data": {"data": "SGVsbG8gV29ybGQh", "attributes": {"key": "value"}} } ``` ### Response #### Success Response (200) None (typically logs output) #### Response Example ``` Event ID: 1234567890 Event Type: google.pubsub.topic.publish Timestamp: 2024-01-15T12:00:00Z Resource: projects/my-project/topics/my-topic Pub/Sub message: Hello World! Attribute key: value ``` ### Cloud Storage Triggered Function #### Description A background function triggered by Cloud Storage events. It accesses event data related to the GCS object and context metadata. ### Method Event-driven (triggered by Cloud Storage) ### Endpoint Not applicable (event-driven) ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body - **event** (dict) - Dictionary containing GCS object metadata. - **context** (Context) - Context object with event metadata. ### Response #### Success Response (200) None (typically logs output) #### Response Example ``` Event Type: google.storage.object.finalize Bucket: my-bucket File: my-file.txt Content Type: text/plain Size: 1024 bytes Created: 2024-01-15T12:00:00Z ``` ### Running Locally ```bash functions-framework --target=hello_pubsub --signature-type=event --debug ``` ``` -------------------------------- ### Run Linting with tox Source: https://github.com/googlecloudplatform/functions-framework-python/blob/main/DEVELOPMENT.md Executes the linting checks for the project using tox. This command ensures code quality and adherence to style guides. ```bash python -m tox -e lint ``` -------------------------------- ### Build Deployable Container with Functions Framework Source: https://github.com/googlecloudplatform/functions-framework-python/blob/main/README.md Builds a container image for a Python function using the Functions Framework buildpacks and the 'pack' tool. It requires Docker and the 'pack' CLI to be installed. The resulting container can be run locally. ```sh pack build \ --builder gcr.io/buildpacks/builder:v1 \ --env GOOGLE_FUNCTION_SIGNATURE_TYPE=http \ --env GOOGLE_FUNCTION_TARGET=hello \ my-first-function ``` ```sh docker run --rm -p 8080:8080 my-first-function ``` ```sh curl localhost:8080 ``` -------------------------------- ### Sending CloudEvents to Python Functions Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt Provides Python code examples for creating and sending CloudEvents to a function using the CloudEvents SDK. Demonstrates sending events in both structured and binary formats via HTTP POST requests. ```python # send_cloud_event.py import requests from cloudevents.http import CloudEvent, to_structured, to_binary # Create a CloudEvent attributes = { "type": "com.example.user.created", "source": "https://example.com/users", "subject": "user-123", } data = { "user_id": "123", "email": "user@example.com", "name": "John Doe", "created_at": "2024-01-15T10:30:00Z" } event = CloudEvent(attributes, data) # Send as structured CloudEvent (JSON body with all attributes) headers, body = to_structured(event) response = requests.post( "http://localhost:8080", headers=headers, data=body ) print(f"Structured response: {response.status_code}") # Send as binary CloudEvent (attributes in headers, data in body) headers, body = to_binary(event) response = requests.post( "http://localhost:8080", headers=headers, data=body ) print(f"Binary response: {response.status_code}") # Using curl for structured format: # curl -X POST http://localhost:8080 \ # -H "Content-Type: application/cloudevents+json" \ # -d '{ # "specversion": "1.0", # "type": "com.example.user.created", # "source": "https://example.com/users", # "id": "event-uuid-here", # "time": "2024-01-15T10:30:00Z", # "subject": "user-123", # "datacontenttype": "application/json", # "data": {"user_id": "123", "email": "user@example.com"} # }' # Using curl for binary format: # curl -X POST http://localhost:8080 \ # -H "Content-Type: application/json" \ # -H "ce-specversion: 1.0" \ # -H "ce-type: com.example.user.created" \ # -H "ce-source: https://example.com/users" \ # -H "ce-id: event-uuid-here" \ # -H "ce-time: 2024-01-15T10:30:00Z" \ # -d '{"user_id": "123", "email": "user@example.com"}' ``` -------------------------------- ### Streaming HTTP Responses with Python Functions Framework Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt Implements synchronous and asynchronous HTTP functions that support streaming responses using the Functions Framework. Useful for large responses or server-sent events. Includes examples for both Flask (sync) and Starlette (async) integration. ```python # main.py import time import asyncio import functions_framework import functions_framework.aio from starlette.responses import StreamingResponse # Synchronous streaming with Flask def generate_data_sync(count: int): """Generator function for sync streaming.""" yield "data: Starting stream\n\n" for i in range(1, count + 1): yield f"data: Item {i}\n\n" time.sleep(0.5) yield "data: Stream complete\n\n" @functions_framework.http def stream_sync(request): """Sync HTTP function with streaming response. Returns a generator that Flask streams to the client. """ count = int(request.args.get("count", "5")) return generate_data_sync(count), { "Content-Type": "text/event-stream", "Cache-Control": "no-cache", "X-Accel-Buffering": "no" } # Asynchronous streaming with Starlette async def generate_data_async(count: int): """Async generator for streaming.""" yield "data: Starting async stream\n\n" for i in range(1, count + 1): yield f"data: Async Item {i}\n\n" await asyncio.sleep(0.3) yield "data: Async stream complete\n\n" @functions_framework.aio.http async def stream_async(request): """Async HTTP function with streaming response.""" count = int(request.query_params.get("count", "5")) return StreamingResponse( generate_data_async(count), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no" } ) # Run locally: # functions-framework --target=stream_sync --debug # functions-framework --target=stream_async --debug # Test with curl (shows streaming output): # curl -N http://localhost:8080?count=10 ``` -------------------------------- ### GET /stream_response Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt An asynchronous endpoint that streams an HTML list of numbers to the client. ```APIDOC ## GET /stream_response ### Description Streams an HTML response containing a list of numbers up to the specified maximum. ### Method GET ### Endpoint /stream ### Parameters #### Query Parameters - **max** (integer) - Optional - The maximum number to list. Defaults to 10. ### Response #### Success Response (200) - **Content-Type** (text/html) - Streamed HTML content. ``` -------------------------------- ### HTTP Function with Error Handling Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt An example of an HTTP Cloud Function written in Python that includes custom error handling for validation, not found, and division by zero errors. ```APIDOC ## HTTP Function with Error Handling ### Description This endpoint handles HTTP requests for mathematical operations (add, subtract, multiply, divide) and includes custom error handling for missing parameters, invalid input, unknown operations, and division by zero. ### Method GET ### Endpoint / ### Query Parameters - **op** (string) - Required - The operation to perform (e.g., 'add', 'subtract', 'multiply', 'divide'). - **a** (number) - Optional - The first operand (defaults to 0). - **b** (number) - Optional - The second operand (defaults to 0). ### Request Example ```json { "example": "curl \"http://localhost:8080?op=add&a=5&b=3\"" } ``` ### Response #### Success Response (200) - **result** (number) - The result of the operation. #### Error Responses - **400 Bad Request**: Returned if 'op' parameter is missing or if 'a'/'b' are not valid numbers. - **404 Not Found**: Returned if the specified operation is unknown. - **418 I'm a teapot**: Returned for division by zero errors. #### Response Example (Success) ```json { "example": "{\"result\": 8.0}" } ``` #### Response Example (Division by Zero) ```json { "example": "I'm a teapot - cannot divide by zero" } ``` #### Response Example (Unknown Operation) ```json { "example": "{\"error\": \"Not Found\", \"message\": \"Unknown operation: unknown\"}" } ``` #### Response Example (Validation Error) ```json { "example": "{\"error\": \"Validation Error\", \"message\": \"Missing 'op' parameter\"}" } ``` ``` -------------------------------- ### Running Functions Framework with CLI Options Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt Demonstrates how to run the Functions Framework using command-line flags or environment variables. Covers options like target function, source file, signature type, host, port, and debug mode. Also shows alternative command names and ASGI mode. ```bash functions-framework \ --target=my_function \ --source=src/main.py \ --signature-type=cloudevent \ --host=127.0.0.1 \ --port=9000 \ --debug ``` ```bash export FUNCTION_TARGET=hello export FUNCTION_SOURCE=main.py export FUNCTION_SIGNATURE_TYPE=http export HOST=0.0.0.0 export PORT=8080 export DEBUG=true functions-framework ``` ```bash functions-framework --target=hello_async --asgi ``` ```bash functions-framework-python --target=hello ``` ```bash ff --target=hello ``` -------------------------------- ### CLI Execution and Configuration Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt Explains how to run functions locally using the functions-framework CLI and configure them via flags or environment variables. ```APIDOC ## CLI Usage ### Description Run your Python function locally using the `functions-framework` CLI. You can configure the target function, source file, signature type, and server settings. ### Parameters - **--target** (string) - Required - The name of the function to invoke. - **--source** (string) - Optional - Path to the source file (default: main.py). - **--signature-type** (string) - Optional - Type of function: http, event, cloudevent, or typed (default: http). - **--host** (string) - Optional - Host to bind (default: 0.0.0.0). - **--port** (integer) - Optional - Port to bind (default: 8080). - **--debug** (boolean) - Optional - Enable debug mode with auto-reload. - **--asgi** (boolean) - Optional - Use ASGI server for async functions. ### Request Example functions-framework --target=my_function --source=src/main.py --signature-type=cloudevent --port=9000 --debug ``` -------------------------------- ### Build and Deploy Container to Cloud Run Source: https://github.com/googlecloudplatform/functions-framework-python/blob/main/examples/cloud_run_http/README.md Commands to build the Docker image, push it to Google Container Registry, and deploy it to Cloud Run. Replace `[PROJECT-ID]` with your actual project ID. ```bash docker build -t gcr.io/[PROJECT-ID]/helloworld . docker push gcr.io/[PROJECT-ID]/helloworld gcloud run deploy helloworld --image gcr.io/[PROJECT-ID]/helloworld --region us-central1 ``` -------------------------------- ### Build and Deploy Containerized Functions Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt Demonstrates how to build container images for various function types using the pack CLI and deploy them to Google Cloud Run. It also provides an alternative approach using a standard Dockerfile. ```bash pack build my-http-function --builder gcr.io/buildpacks/builder:v1 --env GOOGLE_FUNCTION_SIGNATURE_TYPE=http --env GOOGLE_FUNCTION_TARGET=hello pack build my-cloudevent-function --builder gcr.io/buildpacks/builder:v1 --env GOOGLE_FUNCTION_SIGNATURE_TYPE=cloudevent --env GOOGLE_FUNCTION_TARGET=process_event pack build my-event-function --builder gcr.io/buildpacks/builder:v1 --env GOOGLE_FUNCTION_SIGNATURE_TYPE=event --env GOOGLE_FUNCTION_TARGET=hello_pubsub docker run --rm -p 8080:8080 my-http-function curl http://localhost:8080 gcloud run deploy my-service --image my-http-function --platform managed --region us-central1 --allow-unauthenticated ``` ```dockerfile FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . ENV FUNCTION_TARGET=hello CMD ["functions-framework", "--target=hello", "--port=8080"] ``` -------------------------------- ### Define and Run an HTTP Function Source: https://github.com/googlecloudplatform/functions-framework-python/blob/main/README.md Demonstrates how to create a simple HTTP-triggered function using the @functions_framework.http decorator and how to execute it locally using the functions-framework CLI. ```python import flask import functions_framework @functions_framework.http def hello(request: flask.Request) -> flask.typing.ResponseReturnValue: return "Hello world!" ``` ```bash functions-framework --target hello --debug ``` -------------------------------- ### Programmatic ASGI App Creation Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt Illustrates how to create an ASGI application using `functions_framework.aio.create_asgi_app` for asynchronous Cloud Functions, compatible with ASGI servers like Uvicorn. ```APIDOC ## Programmatic ASGI App Creation ### Description This section demonstrates using `functions_framework.aio.create_asgi_app()` to create an ASGI application for asynchronous Cloud Functions. This allows integration with ASGI servers such as Uvicorn. ### Method N/A (Programmatic App Creation) ### Endpoint N/A (App is created programmatically) ### Parameters - **target** (string) - Required - The name of the asynchronous Cloud Function to target. - **source** (string) - Optional - The path to the source file containing the function (defaults to `main.py`). - **signature_type** (string) - Optional - The type of function signature (e.g., `http`, `event`, `cloudevent`, `typed`). Defaults to `http`. ### Request Example ```python import functions_framework.aio from functions_framework.aio import create_asgi_app @functions_framework.aio.http async def async_handler(request): return {"async": True, "method": request.method} app = create_asgi_app( target="async_handler", source="async_app.py", signature_type="http" ) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8080) ``` ### Response #### Success Response (200) N/A (This is for app creation, not an endpoint response) #### Response Example N/A ``` -------------------------------- ### Configure Functions via Environment Variables Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt Shows how to access environment variables within a Python function and how to run the Functions Framework with custom configuration settings. ```python import os import functions_framework @functions_framework.http def configurable_function(request): api_key = os.environ.get("API_KEY", "not-set") database_url = os.environ.get("DATABASE_URL", "sqlite:///local.db") debug_mode = os.environ.get("DEBUG", "false").lower() == "true" return { "api_key_set": api_key != "not-set", "database_url": database_url, "debug_mode": debug_mode, "function_target": os.environ.get("FUNCTION_TARGET"), "function_signature_type": os.environ.get("FUNCTION_SIGNATURE_TYPE") } ``` ```bash FUNCTION_TARGET=configurable_function \ FUNCTION_SIGNATURE_TYPE=http \ API_KEY=my-secret-key \ DATABASE_URL=postgresql://user:pass@host/db \ functions-framework ``` -------------------------------- ### Programmatic WSGI App Creation Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt Demonstrates how to programmatically create a Flask WSGI application using `functions_framework.create_app` for testing or custom server configurations. ```APIDOC ## Programmatic WSGI App Creation ### Description This section shows how to use `functions_framework.create_app()` to generate a WSGI application for a specified Cloud Function. This is useful for local testing, embedding in other applications, or custom server setups. ### Method N/A (Programmatic App Creation) ### Endpoint N/A (App is created programmatically) ### Parameters - **target** (string) - Required - The name of the Cloud Function to target. - **source** (string) - Optional - The path to the source file containing the function (defaults to `main.py`). - **signature_type** (string) - Optional - The type of function signature (e.g., `http`, `event`, `cloudevent`, `typed`). Defaults to `http`. ### Request Example ```python import functions_framework from functions_framework import create_app @functions_framework.http def my_function(request): return {"status": "ok", "path": request.path} app = create_app( target="my_function", source="app.py", signature_type="http" ) if __name__ == "__main__": app.run(host="0.0.0.0", port=8080, debug=True) ``` ### Response #### Success Response (200) N/A (This is for app creation, not an endpoint response) #### Response Example N/A ``` -------------------------------- ### Functions Framework Configuration Options Source: https://github.com/googlecloudplatform/functions-framework-python/blob/main/README.md Configure the Functions Framework using command-line flags or environment variables. Environment variables are ignored if both are specified. ```APIDOC ## Functions Framework Configuration You can configure the Functions Framework using command-line flags or environment variables. If you specify both, the environment variable will be ignored. ### Configuration Table | Command-line flag | Environment variable | Description | | ------------------ | ------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `--host` | `HOST` | The host on which the Functions Framework listens for requests. Default: `0.0.0.0` | | `--port` | `PORT` | The port on which the Functions Framework listens for requests. Default: `8080` | | `--target` | `FUNCTION_TARGET` | The name of the exported function to be invoked in response to requests. Default: `function` | | `--signature-type` | `FUNCTION_SIGNATURE_TYPE` | The signature used when writing your function. Controls unmarshalling rules and determines which arguments are used to invoke your function. Default: `http`; accepted values: `http`, `event` or `cloudevent` | | `--source` | `FUNCTION_SOURCE` | The path to the file containing your function. Default: `main.py` (in the current working directory) | | `--debug` | `DEBUG` | A flag that allows to run functions-framework to run in debug mode, including live reloading. Default: `False` | ``` -------------------------------- ### Local Container Build and Run Source: https://github.com/googlecloudplatform/functions-framework-python/blob/main/examples/cloud_run_http/README.md Commands to build a Docker image locally and run it as a container. This allows for testing the function's behavior in a containerized environment before deployment. ```bash docker build -t helloworld . && docker run --rm -p 8080:8080 -e PORT=8080 helloworld ``` -------------------------------- ### Configure Docker Credential Helper Source: https://github.com/googlecloudplatform/functions-framework-python/blob/main/examples/cloud_run_http/README.md Configures the `gcloud` CLI to use Docker, enabling authentication for pushing container images to Google Container Registry. ```bash gcloud auth configure-docker ``` -------------------------------- ### Create WSGI Application Programmatically Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt Shows how to use functions_framework.create_app to instantiate a Flask WSGI application for a specific target function, enabling custom server configurations or unit testing. ```python from functions_framework import create_app app = create_app( target="my_function", source="app.py", signature_type="http" ) if __name__ == "__main__": app.run(host="0.0.0.0", port=8080, debug=True) ``` -------------------------------- ### Create ASGI Application for Async Functions Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt Explains how to use functions_framework.aio.create_asgi_app to create a Starlette ASGI application, allowing for asynchronous request handling compatible with Uvicorn. ```python from functions_framework.aio import create_asgi_app app = create_asgi_app( target="async_handler", source="async_app.py", signature_type="http" ) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8080) ``` -------------------------------- ### Python Pub/Sub Function with Emulator Source: https://github.com/googlecloudplatform/functions-framework-python/blob/main/README.md Defines a Python function that processes Pub/Sub messages and demonstrates how to run it locally using the Functions Framework and Pub/Sub emulator. It requires the Functions Framework and Google Cloud SDK. ```python def hello(event, context): print("Received", context.event_id) ``` ```sh functions-framework --target=hello --signature-type=event --debug --port=8080 ``` ```sh export PUBSUB_PROJECT_ID=my-project gcloud beta emulators pubsub start \ --project=$PUBSUB_PROJECT_ID \ --host-port=localhost:8085 ``` ```sh export PUBSUB_PROJECT_ID=my-project export TOPIC_ID=my-topic export PUSH_SUBSCRIPTION_ID=my-subscription $(gcloud beta pubsub env-init) git clone https://github.com/googleapis/python-pubsub.git cd python-pubsub/samples/snippets/ pip install -r requirements.txt python publisher.py $PUBSUB_PROJECT_ID create $TOPIC_ID python subscriber.py $PUBSUB_PROJECT_ID create-push $TOPIC_ID $PUSH_SUBSCRIPTION_ID http://localhost:8080 python publisher.py $PUBSUB_PROJECT_ID publish $TOPIC_ID ``` -------------------------------- ### Implement HTTP Functions with Custom Error Handling Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt Demonstrates how to use the @functions_framework.http decorator to handle incoming requests and @functions_framework.errorhandler to manage custom exceptions like validation errors and resource not found scenarios. ```python import functions_framework class ValidationError(Exception): pass class NotFoundError(Exception): pass @functions_framework.errorhandler(ValidationError) def handle_validation_error(error): return {"error": "Validation Error", "message": str(error)}, 400 @functions_framework.errorhandler(NotFoundError) def handle_not_found_error(error): return {"error": "Not Found", "message": str(error)}, 404 @functions_framework.http def calculate(request): operation = request.args.get("op") if not operation: raise ValidationError("Missing 'op' parameter") # Logic implementation... ``` -------------------------------- ### Async HTTP Functions with @functions_framework.aio.http Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt Information on using the `@aio.http` decorator for asynchronous HTTP functions, leveraging Starlette for non-blocking I/O. ```APIDOC ## @functions_framework.aio.http Async HTTP Decorator The `@aio.http` decorator registers an async HTTP function that uses Starlette's ASGI server, enabling non-blocking I/O operations and better concurrency for I/O-bound workloads. ### Description This decorator is used for defining asynchronous HTTP Cloud Functions. It allows you to write non-blocking code, which is particularly beneficial for functions that involve waiting for external resources like databases or other APIs. ### Method HTTP (GET, POST, PUT, DELETE, etc.) ### Endpoint `/` (default, or as configured) ### Parameters #### Path Parameters Depends on the route definition. #### Query Parameters Depends on the route definition. #### Request Body Depends on the route definition and the request content type. ### Request Example ```python import functions_framework from starlette.requests import Request from starlette.responses import JSONResponse @functions_framework.aio.http async def async_http_function(request: Request) -> JSONResponse: """An asynchronous HTTP function. Args: request: Starlette Request object. Returns: Starlette JSONResponse. """ try: data = await request.json() name = data.get("name", "World") return JSONResponse({"message": f"Hello, {name}! (async)"}) except Exception as e: return JSONResponse({"error": str(e)}, status_code=400) # Run locally: # functions-framework --target=async_http_function --signature-type=http # Test with curl: # curl -X POST http://localhost:8080 \ # -H "Content-Type: application/json" \ # -d '{"name": "Bob"}' # Expected output: {"message": "Hello, Bob! (async)"} ``` ### Response #### Success Response (200) - **message** (str) - A response message from the async function. #### Response Example ```json { "message": "Hello, World! (async)" } ``` ``` -------------------------------- ### Functions Framework CLI Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt Basic usage of the Functions Framework CLI for local development and testing of Cloud Functions. ```APIDOC ## Functions Framework CLI ### Description The Functions Framework CLI allows you to run and test your Cloud Functions locally. It supports features like automatic reloading for development. ### Method N/A (CLI Command) ### Endpoint N/A (CLI Command) ### Parameters - **--target** (string) - Required - The name of the function to execute. - **--source** (string) - Optional - The source file containing the function (defaults to `main.py`). - **--port** (integer) - Optional - The port to run the server on (defaults to 8080). - **--debug** - Optional - Enables debug mode. ### Request Example ```bash functions-framework --target=hello ``` ### Response N/A (CLI output varies based on function execution) ``` -------------------------------- ### Run All Tests with tox Source: https://github.com/googlecloudplatform/functions-framework-python/blob/main/DEVELOPMENT.md Executes all defined test suites within the project using tox. This provides a comprehensive check of the project's functionality. ```bash python -m tox ``` -------------------------------- ### GET/POST /hello_async_http Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt An asynchronous HTTP endpoint that accepts a name parameter via query string or JSON body and returns a greeting. ```APIDOC ## GET/POST /hello_async_http ### Description Processes an asynchronous HTTP request to return a personalized greeting. ### Method GET or POST ### Endpoint / ### Parameters #### Query Parameters - **name** (string) - Optional - The name to greet. Defaults to 'World'. #### Request Body - **name** (string) - Optional - The name to greet (used in POST requests). ### Request Example { "name": "Alice" } ### Response #### Success Response (200) - **message** (string) - The greeting message. - **async** (boolean) - Confirmation of async execution. #### Response Example { "message": "Hello, Alice!", "async": true } ``` -------------------------------- ### Define and Invoke a CloudEvent Function Source: https://github.com/googlecloudplatform/functions-framework-python/blob/main/README.md Shows how to create a function that processes CloudEvents using the @functions_framework.cloud_event decorator and how to trigger it via a POST request. ```python import functions_framework from cloudevents.http.event import CloudEvent @functions_framework.cloud_event def hello_cloud_event(cloud_event: CloudEvent) -> None: print(f"Received event with ID: {cloud_event['id']} and data {cloud_event.data}") ``` ```bash curl -X POST localhost:8080 \ -H "Content-Type: application/cloudevents+json" \ -d '{ "specversion" : "1.0", "type" : "example.com.cloud.event", "source" : "https://example.com/cloudevents/pull", "subject" : "123", "id" : "A234-1234-1234", "time" : "2018-04-05T17:31:00Z", "data" : "hello world" }' ``` -------------------------------- ### Streaming HTTP Responses Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt Demonstrates how to implement streaming responses for both synchronous (Flask) and asynchronous (Starlette) functions. ```APIDOC ## POST / (Streaming) ### Description Streams data back to the client using a generator function. Useful for server-sent events or large payloads. ### Method POST ### Endpoint / ### Parameters #### Query Parameters - **count** (integer) - Optional - Number of items to stream. ### Response #### Success Response (200) - **Content-Type** (string) - text/event-stream ### Response Example data: Starting stream data: Item 1 data: Item 2 ``` -------------------------------- ### Run Tests for Current Python Version with tox Source: https://github.com/googlecloudplatform/functions-framework-python/blob/main/DEVELOPMENT.md Runs tests specifically for the Python version currently active in the environment using tox. This is useful for quick checks. ```bash python -m tox -e py ``` -------------------------------- ### Implement Custom Error Handling Source: https://github.com/googlecloudplatform/functions-framework-python/blob/main/README.md Illustrates how to use the @functions_framework.errorhandler decorator to catch specific exceptions and return custom HTTP responses. ```python import functions_framework @functions_framework.errorhandler(ZeroDivisionError) def handle_zero_division(e): return "I'm a teapot", 418 def function(request): 1 / 0 return "Success", 200 ``` -------------------------------- ### Test Function Locally Source: https://github.com/googlecloudplatform/functions-framework-python/blob/main/examples/cloud_run_http/README.md Uses `curl` to send a request to the locally running containerized function and displays the output. ```bash curl localhost:8080 # Output: Hello world! ``` -------------------------------- ### POST /process_event_async Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt An asynchronous CloudEvent handler for processing event-driven data. ```APIDOC ## POST /process_event_async ### Description Processes incoming CloudEvents asynchronously. ### Method POST ### Endpoint / ### Request Body - **specversion** (string) - Required - CloudEvent version. - **type** (string) - Required - The event type. - **source** (string) - Required - The event source. - **id** (string) - Required - Unique event identifier. - **data** (object) - Required - The event payload. ### Request Example { "specversion": "1.0", "type": "com.example.async.event", "source": "/async/source", "id": "async-event-123", "data": {"action": "process", "item_id": "12345"} } ``` -------------------------------- ### Background Event Functions (Legacy) Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt This section describes the legacy signature type for background event functions, which accept separate `event` and `context` parameters. This is primarily for backward compatibility. ```APIDOC ## POST / (Background Event - Legacy) ### Description This endpoint handles background events using the legacy Google Cloud Functions event format, providing separate `event` and `context` arguments to the function. ### Method POST ### Endpoint `/` (root endpoint for background event functions) ### Parameters #### Request Body - **event** (object) - Required - The event data payload. - **context** (object) - Required - The event context information. ### Request Example ```json { "event": {"data": "some event data"}, "context": {"eventId": "12345", "eventType": "google.cloud.pubsub.topic.v1.messagePublished"} } ``` ### Response #### Success Response (200) - **None** - Background event functions typically do not return a value directly; they perform actions based on the event. #### Response Example ```json { "example": "(No explicit response body for background event functions)" } ``` ``` -------------------------------- ### CloudEvents Integration Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt Details how to send structured and binary CloudEvents to a function using the CloudEvents SDK or standard HTTP requests. ```APIDOC ## POST / (CloudEvent) ### Description Sends a CloudEvent to the function. Supports both structured (JSON body) and binary (headers) formats. ### Method POST ### Endpoint / ### Request Body - **specversion** (string) - Required - The version of the CloudEvents specification. - **type** (string) - Required - The type of event. - **source** (string) - Required - The source of the event. - **id** (string) - Required - Unique event identifier. - **data** (object) - Required - The event payload. ### Request Example { "specversion": "1.0", "type": "com.example.user.created", "source": "https://example.com/users", "id": "event-uuid-here", "data": {"user_id": "123"} } ``` -------------------------------- ### Typed Functions with @functions_framework.typed Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt Demonstrates how to use the `@typed` decorator for strongly-typed Cloud Functions, enabling automatic serialization and deserialization. ```APIDOC ## @functions_framework.typed Decorator The `@typed` decorator enables strongly-typed event functions where input and output are automatically serialized/deserialized using custom classes with `from_dict` and `to_dict` methods. ### Description This decorator allows you to define Cloud Functions that accept and return specific data structures, enhancing type safety and reducing boilerplate code for data handling. ### Method HTTP (POST by default for typed functions) ### Endpoint `/` (default, or as configured) ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body - **user** (UserRequest) - An object conforming to the `UserRequest` dataclass structure, automatically deserialized from JSON. ### Request Example ```json { "name": "Alice", "email": "alice@example.com", "age": 30 } ``` ### Response #### Success Response (200) - **message** (str) - A welcome message. - **user_id** (str) - A unique identifier for the user. - **processed** (bool) - Indicates if the processing was successful. #### Response Example ```json { "message": "Welcome, Alice!", "user_id": "user_1234", "processed": true } ``` ### Running Locally ```bash functions-framework --target=process_user --signature-type=typed ``` ### Alternative Usage (Type Annotation) ```python import functions_framework from dataclasses import dataclass from typing import Optional @dataclass class UserRequest: name: str email: str age: Optional[int] = None @classmethod def from_dict(cls, data: dict) -> "UserRequest": return cls( name=data.get("name", ""), email=data.get("email", ""), age=data.get("age") ) @dataclass class UserResponse: message: str user_id: str processed: bool def to_dict(self) -> dict: return { "message": self.message, "user_id": self.user_id, "processed": self.processed } @functions_framework.typed def process_user_annotated(user: UserRequest) -> UserResponse: user_id = f"user_{hash(user.email) % 10000:04d}" return UserResponse( message=f"Hello, {user.name}!", user_id=user_id, processed=True ) ``` ``` -------------------------------- ### HTTP Functions with @functions_framework.http Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt This section details how to define HTTP-triggered functions using the `@functions_framework.http` decorator. It explains how to access request data (query parameters, JSON body, headers) and return responses. ```APIDOC ## POST /api/users ### Description This endpoint handles HTTP requests, supporting both GET and POST methods. It can access query parameters and JSON request bodies. ### Method GET, POST ### Endpoint `/` (root endpoint for HTTP functions) ### Parameters #### Query Parameters - **name** (string) - Optional - The name to greet. #### Request Body - **name** (string) - Optional - The name to greet, provided in JSON format for POST requests. ### Request Example ```json { "name": "Claude" } ``` ### Response #### Success Response (200) - **body** (string) - A greeting message including the provided name and the User-Agent. #### Response Example ```json { "example": "Hello, Claude! User-Agent: curl/7.64.1" } ``` ``` -------------------------------- ### Run Specific Test in File with tox Source: https://github.com/googlecloudplatform/functions-framework-python/blob/main/DEVELOPMENT.md Executes a single, specific test function (e.g., test_cli_no_arguments) within a given test file (e.g., tests/test_cli.py) using tox. This is ideal for debugging individual test cases. ```bash python -m tox -e py -- tests/test_cli.py::test_cli_no_arguments ``` -------------------------------- ### Python Function Definition Source: https://github.com/googlecloudplatform/functions-framework-python/blob/main/examples/cloud_run_http/README.md A simple Python function that returns 'Hello world!'. This is the core logic to be deployed. ```python def hello(request): return "Hello world!" ``` -------------------------------- ### Invoke Deployed Cloud Functions Source: https://github.com/googlecloudplatform/functions-framework-python/blob/main/examples/skaffold/README.md Commands to trigger the deployed functions using curl by querying the Minikube IP address. ```bash curl `minikube ip`/hello curl `minikube ip`/goodbye ``` -------------------------------- ### Implement Background Function with Event Signature Source: https://github.com/googlecloudplatform/functions-framework-python/blob/main/README.md Defines a Python function that accepts event and context objects for background processing. This signature requires setting the signature type to 'event' via configuration. ```python def hello(event, context): print(event) print(context) ``` -------------------------------- ### Run Tests for Specific Python Version with tox Source: https://github.com/googlecloudplatform/functions-framework-python/blob/main/DEVELOPMENT.md Executes tests against a specific Python version (e.g., 3.12) using tox. This allows for targeted testing across different Python environments. ```bash python -m tox -e py3.12 ``` -------------------------------- ### Run Specific Test File with tox Source: https://github.com/googlecloudplatform/functions-framework-python/blob/main/DEVELOPMENT.md Runs tests contained within a particular file (e.g., tests/test_cli.py) using tox and the current Python environment. This helps in isolating test failures. ```bash python -m tox -e py -- tests/test_cli.py ``` -------------------------------- ### Enable Google Cloud Run Function Events Source: https://github.com/googlecloudplatform/functions-framework-python/blob/main/README.md Enable automatic unmarshalling of Google Cloud Run function event payloads to `event` and `context` objects by setting the function signature type to `event`. ```APIDOC ## Enable Google Cloud Run function Events The Functions Framework can unmarshall incoming Google Cloud Run functions event payloads to `event` and `context` objects. These will be passed as arguments to your function when it receives a request. Note that your function must use the `event`-style function signature. ### Example Function Signature ```python def hello(event, context): print(event) print(context) ``` To enable automatic unmarshalling, set the function signature type to `event` using the `--signature-type` command-line flag or the `FUNCTION_SIGNATURE_TYPE` environment variable. By default, the HTTP signature will be used and automatic event unmarshalling will be disabled. For more details on this signature type, see the Google Cloud Functions documentation on background functions. ``` -------------------------------- ### CloudEvent Functions with @functions_framework.cloud_event Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt This section covers defining functions that process CloudEvents using the `@functions_framework.cloud_event` decorator. It shows how to access CloudEvent attributes and data payload. ```APIDOC ## POST / (CloudEvent) ### Description This endpoint handles CloudEvents, processing incoming event data and metadata. It's designed for event-driven architectures. ### Method POST ### Endpoint `/` (root endpoint for CloudEvent functions) ### Parameters #### Request Body - **CloudEvent object** (object) - Required - The CloudEvent payload conforming to the CloudEvents specification. - **specversion** (string) - Required - The version of the CloudEvents specification. - **type** (string) - Required - The type of the event. - **source** (string) - Required - The source of the event. - **subject** (string) - Optional - The subject of the event. - **id** (string) - Required - The unique identifier for the event. - **time** (string) - Optional - The timestamp of the event. - **data** (any) - Optional - The event payload data. ### Request Example ```json { "specversion": "1.0", "type": "example.com.cloud.event", "source": "https://example.com/source", "subject": "my-resource", "id": "A234-1234-1234", "time": "2024-01-15T17:31:00Z", "data": {"message": "Hello from CloudEvent!"} } ``` ### Response #### Success Response (200) - **None** - CloudEvent functions typically do not return a value directly; they perform actions based on the event. #### Response Example ```json { "example": "(No explicit response body for CloudEvent functions)" } ``` ``` -------------------------------- ### Async Function for Concurrent API Calls Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt This asynchronous function demonstrates how to make multiple HTTP requests concurrently using `aiohttp` and `asyncio.gather`. It fetches data from several URLs in parallel and returns a JSON response indicating the number of successful fetches. ```python import asyncio import functions_framework.aio from starlette.requests import Request from starlette.responses import JSONResponse, StreamingResponse @functions_framework.aio.http async def fetch_multiple_apis(request: Request): """Async function demonstrating concurrent API calls.""" import aiohttp async def fetch_url(session, url): async with session.get(url) as response: return await response.text() urls = [ "https://httpbin.org/get", "https://httpbin.org/headers", ] async with aiohttp.ClientSession() as session: # Fetch all URLs concurrently results = await asyncio.gather( *[fetch_url(session, url) for url in urls] ) return JSONResponse({"fetched": len(results)}) ``` -------------------------------- ### Async Pub/Sub CloudEvent Handler Source: https://context7.com/googlecloudplatform/functions-framework-python/llms.txt An asynchronous CloudEvent function specifically designed to handle Google Cloud Pub/Sub messages. It decodes base64-encoded message data and processes message attributes. The function includes a simulation of an asynchronous acknowledgment process. ```python import asyncio import functions_framework.aio from cloudevents.http.event import CloudEvent @functions_framework.aio.cloud_event async def handle_pubsub_message(cloud_event: CloudEvent) -> None: """Async handler for Pub/Sub CloudEvents.""" import base64 # Pub/Sub CloudEvents wrap message in 'message' field message_data = cloud_event.data.get("message", {}) # Decode base64 message data if "data" in message_data: decoded = base64.b64decode(message_data["data"]).decode("utf-8") print(f"Pub/Sub message: {decoded}") # Process message attributes attributes = message_data.get("attributes", {}) for attr_key, attr_value in attributes.items(): print(f"Attribute: {attr_key}={attr_value}") # Async acknowledgment simulation await asyncio.sleep(0.1) ```