API reference

aiohttp_asgi

class aiohttp_asgi.ASGIResource(app, root_path='/', name=None)

Bases: AbstractResource

An aiohttp resource which serves an ASGI application.

Register it in an aiohttp router, and the ASGI application answers every request which starts with root_path.

resource = ASGIResource(asgi_app, root_path="/api")
aiohttp_app.router.register_resource(resource)
resource.lifespan_mount(aiohttp_app)

Parameters:

• app (Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Any]], Coroutine[Any, Any, Any]]) – the ASGI application.

• root_path (str) – the path prefix of the application.

• name (Optional[str]) – the resource name for the aiohttp router.

SHUTDOWN_TIMEOUT = 60

Maximum time in seconds to wait for the lifespan task after shutdown.

__init__(app, root_path='/', name=None)

Parameters:

• app (Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Any]], Coroutine[Any, Any, Any]])

• root_path (str)

• name (Optional[str])

property canonical: str

Exposes the resource’s canonical path.

For example ‘/foo/bar/{name}’

url_for(**kwargs)

Construct url for resource with additional params.

Parameters:

kwargs (str)

Return type:

URL

add_prefix(prefix)

Add a prefix to processed URLs.

Required for subapplications support.

Parameters:

prefix (str)

Return type:

None

get_info()

Return a dict with additional info useful for introspection

Return type:

ResourceInfoDict

raw_match(path)

Return True when the path starts with the resource root_path.

Parameters:

path (str)

Return type:

bool

async resolve(request)

Resolve resource.

Return (UrlMappingMatchInfo, allowed_methods) pair.

Parameters:

request (Request)

Return type:

Tuple[Optional[ASGIMatchInfo], Set[str]]

lifespan_mount(app)

Send the ASGI lifespan events with the aiohttp application.

The method appends a cleanup context to the aiohttp application. The context sends lifespan.startup on startup, and lifespan.shutdown on shutdown. Without this call the ASGI application gets no lifespan events.

Parameters:

app (Application) – the aiohttp application which gives the events.

Return type:

None

aiohttp_asgi.resource

class aiohttp_asgi.resource.ASGIMatchInfo(handler)

Bases: AbstractMatchInfo

The match info which ASGIResource.resolve() returns.

aiohttp takes the handler, the application stack and the expect handler from this object.

Parameters:

handler (Callable[..., Any])

property handler: Callable[[Request], Awaitable[StreamResponse]]

Execute matched request handler

property expect_handler: Callable[[Request], Awaitable[Optional[StreamResponse]]]

Expect handler for 100-continue processing

property http_exception: Optional[HTTPException]

HTTPException instance raised on router’s resolving, or None

get_info()

Return a dict with additional info useful for introspection

Return type:

Dict[str, Any]

property apps: Tuple[Application, ...]

Stack of nested applications.

Top level application is left-most element.

add_app(app)

Add application to the nested apps stack.

Parameters:

app (Application)

Return type:

None

freeze()

Freeze the match info.

The method is called after route resolution.

After the call .add_app() is forbidden.

Return type:

None

class aiohttp_asgi.resource.ASGIContext(app, request, root_path)

Bases: object

The bridge between one aiohttp request and the ASGI application.

The context builds the ASGI scope, moves the body in both directions and makes the aiohttp response.

Parameters:

• app (Callable[..., Any])

• request (Request)

• root_path (str)

Type aliases

aiohttp_asgi.resource gives the type aliases of the ASGI protocol:

Alias	Meaning
ASGIScopeType	The scope: a mutable mapping with string keys
ASGIReceiveType	The receive callable, which gives the next message
ASGISendType	The send callable, which takes a message
ASGIApplicationType	The application: a callable which takes the scope, receive and send

aiohttp_asgi.main

The command line server. The functions are useful for a custom entry point which needs the same behaviour.

aiohttp_asgi.__main__.parse_app_spec(spec)

Check the “module_name:asgi_application” format.

This function does not import the module, because the import must happen in the worker process, after the fork.

Parameters:

spec (str)

Return type:

str

aiohttp_asgi.__main__.load_app(spec)

Import the ASGI application from the “module_name:asgi_application” spec.

Call this in the worker process. An application can start threads or event loops on import, and the fork does not copy them correctly.

Parameters:

spec (str)

Return type:

Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Any]], Coroutine[Any, Any, Any]]

aiohttp_asgi.__main__.new_event_loop(implementation='stdlib')

Create the event loop for the server.

Call this in the worker process so optional loops are imported and created after the fork.

Parameters:

implementation (str)

Return type:

AbstractEventLoop

aiohttp_asgi.__main__.bind_socket(*args, address, port, reuse_addr=True, reuse_port=False)

Parameters:

• args (Any)

• address (str)

• port (int)

• reuse_addr (bool)

• reuse_port (bool)

Return type:

socket

aiohttp_asgi.__main__.stop_on_signal(stop)

Set the stop event on SIGINT and SIGTERM.

The server stops from a signal callback, because an exception from a signal handler does not leave every event loop implementation, and in a worker it can also die inside the request handler task.

Parameters:

stop (Event)

Return type:

None

async aiohttp_asgi.__main__.run_server(app, sock)

Serve the application on the bound socket until a signal arrives.

Parameters:

• app (Application)

• sock (socket)

Return type:

None

aiohttp_asgi.__main__.serve(spec, sock, event_loop='stdlib')

Import the application and run the server on the bound socket.

This function runs in the worker process, therefore it creates the event loop after the fork.

Parameters:

• spec (str)

• sock (socket)

• event_loop (str)

Return type:

None

aiohttp_asgi.__main__.main()

Return type:

None