ASGI Adapter¶

ASGIAdapter¶

ASGIAdapter ¶

ASGIAdapter(app: RestApplication, metrics_publisher: Union[MetricsPublisher, None, _DefaultPublisher] = _DEFAULT_PUBLISHER, enable_metrics: Optional[bool] = None, namespace: Optional[str] = None, service_name: Optional[str] = None, metrics_resolution: int = 60)

ASGI 3.0 adapter for running RestMachine applications on ASGI servers.

This adapter enables RestMachine applications to run on any ASGI-compatible server (Uvicorn, Hypercorn, Daphne, etc.) while maintaining the synchronous programming model internally.

The adapter handles: - True streaming of request bodies (application receives data as it arrives) - Converting ASGI scope/receive/send to RestMachine Request - Executing the synchronous RestMachine application in a thread pool - Converting RestMachine Response to ASGI response format (streaming if needed) - Proper header normalization and encoding - Content-Type and Content-Length handling - Automatic AWS detection - Enables CloudWatch EMF metrics when running in AWS

Streaming behavior: - Request bodies: The adapter receives the first chunk, passes the stream to the application immediately, and continues receiving chunks in the background. The application can start processing before all data has arrived. - Response bodies: File-like objects are streamed in 8KB chunks with proper ASGI more_body signaling.

Metrics behavior: - AWS auto-detection: When AWS environment is detected (via AWS_REGION or AWS_EXECUTION_ENV), CloudWatch EMF metrics are automatically enabled - Manual configuration: Pass custom metrics_publisher for other platforms - Disable: Set enable_metrics=False to disable metrics

Example

from restmachine import RestApplication
from restmachine.adapters import ASGIAdapter

app = RestApplication()

@app.get("/")
def home(metrics):
    metrics.add_metric("requests", 1)
    return {"message": "Hello World"}

# Create ASGI application - metrics auto-enabled if in AWS
asgi_app = ASGIAdapter(app)

# Use with any ASGI server:
# uvicorn module:asgi_app
# hypercorn module:asgi_app

Initialize the ASGI adapter with optional metrics support.

Automatically detects AWS environment and enables CloudWatch EMF metrics when running on AWS (ECS, EC2, App Runner, etc.).

Parameters:

Name	Type	Description	Default
`app`	`RestApplication`	The RestMachine application to wrap	required
`metrics_publisher`	`Union[MetricsPublisher, None, _DefaultPublisher]`	Metrics publisher. Defaults to auto-detection: - CloudWatch EMF if AWS environment detected - None if not in AWS Pass explicit publisher to override, or None to disable.	`_DEFAULT_PUBLISHER`
`enable_metrics`	`Optional[bool]`	Explicitly enable/disable metrics (overrides auto-detection)	`None`
`namespace`	`Optional[str]`	CloudWatch namespace (used if AWS detected, default: "RestMachine")	`None`
`service_name`	`Optional[str]`	Service name dimension (default: from env or "asgi-app")	`None`
`metrics_resolution`	`int`	Metric resolution in seconds, 1 or 60 (default: 60)	`60`

Examples:

Auto-detect AWS and enable EMF¶

adapter = ASGIAdapter(app)

Custom namespace for AWS¶

adapter = ASGIAdapter(app, namespace="MyApp/API")

Custom publisher (Prometheus, etc.)¶

adapter = ASGIAdapter(app, metrics_publisher=PrometheusPublisher())

Disable metrics¶

adapter = ASGIAdapter(app, enable_metrics=False)

Functions¶

call `async` ¶

__call__(scope: Dict[str, Any], receive: Callable[[], Awaitable[Dict[str, Any]]], send: Callable[[Dict[str, Any]], Awaitable[None]])

ASGI 3.0 application entry point.

Parameters:

Name	Type	Description	Default
`scope`	`Dict[str, Any]`	ASGI connection scope dictionary	required
`receive`	`Callable[[], Awaitable[Dict[str, Any]]]`	Async callable to receive ASGI messages	required
`send`	`Callable[[Dict[str, Any]], Awaitable[None]]`	Async callable to send ASGI messages	required

Adapter (Base Class)¶

Adapter ¶

Bases: ABC

Abstract base class for synchronous event adapters.

Use this base class for platforms that use synchronous event handling, such as AWS Lambda, Azure Functions, Google Cloud Functions, etc.

For ASGI servers, use ASGIAdapter instead.

Functions¶

handle_event `abstractmethod` ¶

handle_event(event: Any, context: Optional[Any] = None) -> Any

Handle an external event and return the appropriate response format.

Parameters:

Name	Type	Description	Default
`event`	`Any`	The external event (e.g., AWS Lambda event)	required
`context`	`Optional[Any]`	Optional context (e.g., AWS Lambda context)	`None`

Returns:

Type	Description
`Any`	Response in the format expected by the external system

convert_to_request `abstractmethod` ¶

convert_to_request(event: Any, context: Optional[Any] = None) -> Request

Convert an external event to a Request object.

Parameters:

Name	Type	Description	Default
`event`	`Any`	The external event	required
`context`	`Optional[Any]`	Optional context	`None`

Returns:

Type	Description
`Request`	Request object that can be processed by the app

convert_from_response `abstractmethod` ¶

convert_from_response(response: Response, event: Any, context: Optional[Any] = None) -> Any

Convert a Response object to the format expected by the external system.

Parameters:

Name	Type	Description	Default
`response`	`Response`	Response from the app	required
`event`	`Any`	Original external event	required
`context`	`Optional[Any]`	Optional context	`None`

Returns:

Type	Description
`Any`	Response in the format expected by the external system

create_asgi_app¶

create_asgi_app ¶

create_asgi_app(app: RestApplication, **adapter_kwargs: Any) -> ASGIAdapter

Create an ASGI application from a RestMachine application.

This is a convenience function for creating ASGI adapters with optional metrics configuration.

Parameters:

Name	Type	Description	Default
`app`	`RestApplication`	The RestMachine application to wrap	required
`**adapter_kwargs`	`Any`	Additional arguments passed to ASGIAdapter: - metrics_publisher: Custom metrics publisher - enable_metrics: Explicitly enable/disable metrics - namespace: CloudWatch namespace (if AWS detected) - service_name: Service name dimension - metrics_resolution: Metric resolution (1 or 60 seconds)	`{}`

Returns:

Type	Description
`ASGIAdapter`	An ASGI-compatible application

Examples:

from restmachine import RestApplication
from restmachine.adapters import create_asgi_app

app = RestApplication()

@app.get("/")
def home(metrics):
    metrics.add_metric("requests", 1)
    return {"message": "Hello World"}

# Auto-detect AWS and enable EMF (default)
asgi_app = create_asgi_app(app)

# Custom namespace for AWS
asgi_app = create_asgi_app(app, namespace="MyApp/API")

# Disable metrics
asgi_app = create_asgi_app(app, enable_metrics=False)

# Run with uvicorn
# uvicorn module:asgi_app --reload

Overview¶

Adapters convert RestMachine applications to run on different platforms. The ASGIAdapter enables deployment on any ASGI-compatible server (Uvicorn, Hypercorn, Daphne, etc.).

Quick Start¶

from restmachine import RestApplication, ASGIAdapter

app = RestApplication()

@app.get('/hello')
def hello():
    return {"message": "Hello World"}

# Create ASGI app
asgi_app = ASGIAdapter(app)

# Run with uvicorn
# uvicorn myapp:asgi_app --reload

ASGI Adapter¶

ASGIAdapter¶

ASGIAdapter ¶

Auto-detect AWS and enable EMF¶

Custom namespace for AWS¶

Custom publisher (Prometheus, etc.)¶

Disable metrics¶

Functions¶

__call__ async ¶

Adapter (Base Class)¶

Adapter ¶

Functions¶

handle_event abstractmethod ¶

convert_to_request abstractmethod ¶

convert_from_response abstractmethod ¶

create_asgi_app¶

create_asgi_app ¶

Overview¶

Quick Start¶

See Also¶

call `async` ¶

handle_event `abstractmethod` ¶

convert_to_request `abstractmethod` ¶

convert_from_response `abstractmethod` ¶