API Reference#

This part of the documentation covers all the interfaces of ASGI-Tools.

Request#

class asgi_tools.Request(scope, receive, send)[source]#

Represent a HTTP Request.

Parameters:

  • scope (TASGIScope) – HTTP ASGI Scope

  • receive (TASGIReceive) – an asynchronous callable which lets the application receive event messages from the client

  • send (TASGISend) – an asynchronous callable which lets the application send event messages to the client

The class gives you an nicer interface to incoming HTTP request.

from asgi_tools import Request, Response

async def app(scope, receive, send):
    request = Request(scope, receive, send)
    content = f"{ request.method } { request.url.path }"
    response = Response(content)
    await response(scope, receive, send)

Requests are based on a given scope and represents a mapping interface.

request = Request(scope, receive, send)
assert request['version'] == scope['version']
assert request['method'] == scope['method']
assert request['scheme'] == scope['scheme']
assert request['path'] == scope['path']

# and etc

# ASGI Scope keys also are available as Request attrubutes.

assert request.version == scope['version']
assert request.method == scope['method']
assert request.scheme == scope['scheme']
headers#

A lazy property that parses the current scope’s headers, decodes them as strings and returns case-insensitive multi-dict multidict.CIMultiDict.

request = Request(scope)

assert request.headers['content-type']
assert request.headers['authorization']

See multidict documentation for futher reference.

cookies#

A lazy property that parses the current scope’s cookies and returns a dictionary.

request = Request(scope)
ses = request.cookies.get('session')
query#

A lazy property that parse the current query string and returns it as a multidict.MultiDict.

charset#

Get an encoding charset for the current scope.

content_type#

Get a content type for the current scope.

url#

A lazy property that parses the current URL and returns yarl.URL object.

request = Request(scope)
assert str(request.url) == '... the full http URL ..'
assert request.url.scheme
assert request.url.host
assert request.url.query is not None
assert request.url.query_string is not None

See yarl documentation for further reference.

async stream()[source]#

Stream the request’s body.

The method provides byte chunks without storing the entire body to memory. Any subsequent calls to body(), form(), json() or data() will raise an error.

Warning

You can only read stream once. Second call raises an error. Save a readed stream into a variable if you need.

from asgi_tools import Request, Response

async def app(scope, receive, send):
    request = Request(scope, receive, send)
    body = b''
    async for chunk in request.stream():
        body += chunk

    response = Response(body, content_type=request.content_type)
    await response(scope, receive, send)
Return type:

AsyncGenerator

async body()[source]#

Read and return the request’s body as bytes.

body = await request.body()

Return type:

bytes

async text()[source]#

Read and return the request’s body as a string.

text = await request.text()

Return type:

str

async form(max_size=0, upload_to=None, file_memory_limit=1048576)[source]#

Read and return the request’s multipart formdata as a multidict.

The method reads the request’s stream stright into memory formdata. Any subsequent calls to body(), json() will raise an error.

Parameters:

  • max_size (int) – The maximum size of the request body in bytes.

  • upload_to (Optional[Callable]) – A callable to be used to determine the upload path for files.

  • file_memory_limit (int) – The maximum size of the file to be stored in memory in bytes.

Return type:

MultiDict

formdata = await request.form()

async json()[source]#

Read and return the request’s body as a JSON.

json = await request.json()

Return type:

None | bool | int | float | str | List[TJSON] | Mapping[str, TJSON]

async data(*, raise_errors=False)[source]#

The method checks Content-Type Header and parse the request’s data automatically.

Parameters:

raise_errors (bool) – Raise an error if the given data is invalid.

Return type:

Union[str, bytes, MultiDict, TJSON]

data = await request.data()

If raise_errors is false (by default) and the given data is invalid (ex. invalid json) the request’s body would be returned.

Returns data from json() for application/json, form() for application/x-www-form-urlencoded, multipart/form-data and text() otherwise.

Responses#

class asgi_tools.Response(content, *, status_code=None, content_type=None, headers=None, cookies=None)[source]#

A base class to make ASGI responses.

Parameters:

  • content (str | bytes) – A response’s body

  • status_code (int) – An HTTP status code

  • headers (dict[str, str]) – A dictionary of HTTP headers

  • content_type (str) – A string with the content-type

  • cookies (dict[str, str]) – An initial dictionary of cookies

A helper to make http responses.

from asgi_tools import Response

async def app(scope, receive, send):
    response = Response('Hello, world!', content_type='text/plain')
    await response(scope, receive, send)
headers: MultiDict#

Multidict of response’s headers

cookies: SimpleCookie#

Set/Update cookies

  • response.cookies[name] = value str – set a cookie’s value

  • response.cookies[name][‘path’] = value str – set a cookie’s path

  • response.cookies[name][‘expires’] = value int – set a cookie’s expire

  • response.cookies[name][‘domain’] = value str – set a cookie’s domain

  • response.cookies[name][‘max-age’] = value int – set a cookie’s max-age

  • response.cookies[name][‘secure’] = value bool– is the cookie should only be sent if request is SSL

  • response.cookies[name][‘httponly’] = value bool – is the cookie should be available through HTTP request only (not from JS)

  • response.cookies[name][‘samesite’] = value str – set a cookie’s strategy (‘lax’|’strict’|’none’)

from asgi_tools import Response

async def app(scope, receive, send):
    response = Response('OK')
    response.cookies["rocky"] = "road"
    response.cookies["rocky"]["path"] = "/cookie"
    await response(scope, receive, send)
async __call__(_, __, send)[source]#

Behave as an ASGI application.

Parameters:

send (TASGISend) –

ResponseText#

class asgi_tools.ResponseText(content, *, status_code=None, content_type=None, headers=None, cookies=None)[source]#

A helper to return plain text responses (text/plain).

from asgi_tools import ResponseText

async def app(scope, receive, send):
    response = ResponseText('Hello, world!')
    await response(scope, receive, send)
Parameters:

  • status_code (int) –

  • content_type (str | None) –

  • headers (MultiDict) –

  • cookies (SimpleCookie) –

ResponseHTML#

class asgi_tools.ResponseHTML(content, *, status_code=None, content_type=None, headers=None, cookies=None)[source]#

A helper to return HTML responses (text/html).

from asgi_tools import ResponseHTML

async def app(scope, receive, send):
    response = ResponseHTML('<h1>Hello, world!</h1>')
    await response(scope, receive, send)
Parameters:

  • status_code (int) –

  • content_type (str | None) –

  • headers (MultiDict) –

  • cookies (SimpleCookie) –

ResponseJSON#

class asgi_tools.ResponseJSON(content, *, status_code=None, content_type=None, headers=None, cookies=None)[source]#

A helper to return JSON responses (application/json).

The class optionally supports ujson and orjson JSON libraries. Install one of them to use instead the standard library.

from asgi_tools import ResponseJSON

async def app(scope, receive, send):
    response = ResponseJSON({'hello': 'world'})
    await response(scope, receive, send)
Parameters:

  • status_code (int) –

  • content_type (str | None) –

  • headers (MultiDict) –

  • cookies (SimpleCookie) –

ResponseRedirect#

class asgi_tools.ResponseRedirect(url, status_code=None, **kwargs)[source]#

A helper to return HTTP redirects. Uses a 307 status code by default.

Parameters:

  • url (str) – A string with the new location

  • status_code (int) – 

from asgi_tools import ResponseRedirect

async def app(scope, receive, send):
    response = ResponseRedirect('/login')
    await response(scope, receive, send)

If you are using asgi_tools.App or asgi_tools.ResponseMiddleware you are able to raise the ResponseRedirect as an exception.

from asgi_tools import ResponseRedirect, Request, ResponseMiddleware

async def app(scope, receive, send):
    request = Request(scope, receive, send)
    if not request.headers.get('authorization):
        raise ResponseRedirect('/login')

    return 'OK'

app = ResponseMiddleware(app)

ResponseError#

class asgi_tools.ResponseError(message=None, status_code=500, **kwargs)[source]#

A helper to return HTTP errors. Uses a 500 status code by default.

Parameters:

message – A string with the error’s message (HTTPStatus messages will be used by default)

from asgi_tools import ResponseError

async def app(scope, receive, send):
    response = ResponseError('Timeout', 502)
    await response(scope, receive, send)

If you are using asgi_tools.App or asgi_tools.ResponseMiddleware you are able to raise the ResponseError as an exception.

from asgi_tools import ResponseError, Request, ResponseMiddleware

async def app(scope, receive, send):
    request = Request(scope, receive, send)
    if not request.method == 'POST':
        raise ResponseError('Invalid request data', 400)

    return 'OK'

app = ResponseMiddleware(app)

You able to use http.HTTPStatus properties with the ResponseError class

response = ResponseError.BAD_REQUEST('invalid data')
response = ResponseError.NOT_FOUND()
response = ResponseError.BAD_GATEWAY()
# and etc

ResponseStream#

class asgi_tools.ResponseStream(stream, **kwargs)[source]#

A helper to stream a response’s body.

Parameters:

  • content (AsyncGenerator) – An async generator to stream the response’s body

  • stream (AsyncGenerator[Any, None]) – 

from asgi_tools import ResponseStream
from asgi_tools.utils import aio_sleep  # for compatability with different async libs

async def stream_response():
    for number in range(10):
        await aio_sleep(1)
        yield str(number)

async def app(scope, receive, send):
    generator = stream_response()
    response = ResponseStream(generator, content_type='plain/text')
    await response(scope, receive, send)

ResponseSSE#

class asgi_tools.ResponseSSE(stream, **kwargs)[source]#

A helper to stream SSE (server side events).

Parameters:

  • content (AsyncGenerator) – An async generator to stream the events

  • stream (AsyncGenerator[Any, None]) – 

from asgi_tools import ResponseSSE
from asgi_tools.utils import aio_sleep  # for compatability with different async libs

async def stream_response():
    for number in range(10):
        await aio_sleep(1)
        # The response support messages as text
        yield "data: message text"

        # And as dictionaties as weel
        yield {
            "event": "ping",
            "data": time.time(),
        }

async def app(scope, receive, send):
    generator = stream_response()
    response = ResponseSSE(generator)
    await response(scope, receive, send)

ResponseFile#

class asgi_tools.ResponseFile(filepath, *, chunk_size=65536, filename=None, headers_only=False, **kwargs)[source]#

A helper to stream files as a response body.

Parameters:

  • filepath (str | Path) – The filepath to the file

  • chunk_size (int) – Default chunk size (32768)

  • filename (str) – If set, Content-Disposition header will be generated

  • headers_only (bool) – Return only file headers

from asgi_tools import ResponseFile

async def app(scope, receive, send):

    # Return file
    if scope['path'] == '/selfie':
        response = ResponseFile('/storage/my_best_selfie.jpeg')

    # Download file
    else:
        response = ResponseFile('/storage/video-2020-01-01.mp4', filename='movie.mp4')

    await response(scope, receive, send)

ResponseWebSocket#

class asgi_tools.ResponseWebSocket(scope, receive=None, send=None)[source]#

A helper to work with websockets.

Parameters:

  • scope (dict) – Request info (ASGI Scope | ASGI-Tools Request)

  • receive (Optional[TASGIReceive]) – ASGI receive function

  • send (Optional[TASGISend]) – ASGI send function

from asgi_tools import ResponseWebsocket

async def app(scope, receive, send):
    async with ResponseWebSocket(scope, receive, send) as ws:
        msg = await ws.receive()
        assert msg == 'ping'
        await ws.send('pong')
async accept(**params)[source]#

Accept a websocket connection.

Return type:

None

async close(code=1000)[source]#

Sent by the application to tell the server to close the connection.

Parameters:

code (int) –

Return type:

None

async send(msg, msg_type='websocket.send')[source]#

Send the given message to a client.

Parameters:

msg (Dict | str | bytes) –

Return type:

None

async send_json(data)[source]#

Serialize the given data to JSON and send to a client.

Return type:

None

async receive(*, raw=False)[source]#

Receive messages from a client.

Parameters:

raw (bool) – Receive messages as is.

Return type:

Union[TASGIMessage, str]

Middlewares#

RequestMiddleware#

class asgi_tools.RequestMiddleware(app=None)[source]#

Automatically create asgi_tools.Request from the scope and pass it to ASGI apps.

from asgi_tools import RequestMiddleware, Response

async def app(request, receive, send):
    content = f"{ request.method } { request.url.path }"
    response = Response(content)
    await response(scope, receive, send)

app = RequestMiddleware(app)
Parameters:

app (Optional[TASGIApp]) –

ResponseMiddleware#

class asgi_tools.ResponseMiddleware(app=None)[source]#

Automatically convert ASGI apps results into responses Response and send them to server as ASGI messages.

from asgi_tools import ResponseMiddleware, ResponseText, ResponseRedirect

async def app(scope, receive, send):
    # ResponseMiddleware catches ResponseError, ResponseRedirect and convert the exceptions
    # into HTTP response
    if scope['path'] == '/user':
        raise ResponseRedirect('/login')

    # Return ResponseHTML
    if scope['method'] == 'GET':
        return '<b>HTML is here</b>'

    # Return ResponseJSON
    if scope['method'] == 'POST':
        return {'json': 'here'}

    # Return any response explicitly
    if scope['method'] == 'PUT':
        return ResponseText('response is here')

    # Short form to responses: (status_code, body) or (status_code, body, headers)
    return 405, 'Unknown method'

app = ResponseMiddleware(app)

The conversion rules:

  • Response objects will be directly returned from the view

  • dict, list, int, bool, None results will be converted into ResponseJSON

  • str, bytes results will be converted into ResponseHTML

  • tuple[int, Any, dict] will be converted into a Response with int status code, dict will be used as headers, Any will be used to define the response’s type

from asgi_tools import ResponseMiddleware

# The result will be converted into HTML 404 response with the 'Not Found' body
async def app(request, receive, send):
    return 404, 'Not Found'

app = ResponseMiddleware(app)

You are able to raise ResponseError from yours ASGI apps and it will be catched and returned as a response

Parameters:

app (Optional[TASGIApp]) –

LifespanMiddleware#

class asgi_tools.LifespanMiddleware(app=None, *, logger=<Logger asgi-tools (WARNING)>, ignore_errors=False, on_startup=None, on_shutdown=None)[source]#

Manage ASGI Lifespan events.

Parameters:

  • ignore_errors (bool) – Ignore errors from startup/shutdown handlers

  • on_startup (Union[Callable, List[Callable], None]) – the list of callables to run when the app is starting

  • on_shutdown (Union[Callable, List[Callable], None]) – the list of callables to run when the app is finishing

  • app (Optional[TASGIApp]) – 

from asgi_tools import LifespanMiddleware, Response

async def app(scope, receive, send):
    response = Response('OK')
    await response(scope, receive, send)

app = lifespan = LifespanMiddleware(app)

@lifespan.on_startup
async def start():
    print('The app is starting')

@lifespan.on_shutdown
async def start():
    print('The app is finishing')

Lifespan middleware may be used as an async context manager for testing purposes

on_startup(fn)[source]#

Add a function to startup.

Parameters:

fn (Callable) –

Return type:

None

on_shutdown(fn)[source]#

Add a function to shutdown.

Parameters:

fn (Callable) –

Return type:

None

RouterMiddleware#

class asgi_tools.RouterMiddleware(app=None, router=None)[source]#

Manage routing.

from asgi_tools import RouterMiddleware, ResponseHTML, ResponseError

async def default_app(scope, receive, send):
    response = ResponseError.NOT_FOUND()
    await response(scope, receive, send)

app = router = RouterMiddleware(default_app)

@router.route('/status', '/stat')
async def status(scope, receive, send):
    response = ResponseHTML('STATUS OK')
    await response(scope, receive, send)

# Bind methods
# ------------
@router.route('/only-post', methods=['POST'])
async def only_post(scope, receive, send):
    response = ResponseHTML('POST OK')
    await response(scope, receive, send)

# Regexp paths
# ------------
import re

@router.route(re.compile(r'/\d+/?'))
async def num(scope, receive, send):
    num = int(scope['path'].strip('/'))
    response = ResponseHTML(f'Number { num }')
    await response(scope, receive, send)

# Dynamic paths
# -------------

@router.route('/hello/{name}')
async def hello(scope, receive, send):
    name = scope['path_params']['name']
    response = ResponseHTML(f'Hello { name.title() }')
    await response(scope, receive, send)

# Set regexp for params
@router.route(r'/multiply/{first:\d+}/{second:\d+}')
async def multiply(scope, receive, send):
    first, second = map(int, scope['path_params'].values())
    response = ResponseHTML(str(first * second))
    await response(scope, receive, send)

Path parameters are made available in the request/scope, as the path_params dictionary.

Parameters:

  • app (Optional[TASGIApp]) –

  • router (Optional[Router]) –

StaticFilesMiddleware#

class asgi_tools.StaticFilesMiddleware(app=None, url_prefix='/static', folders=None)[source]#

Serve static files.

Parameters:

  • url_prefix (str, "/static") – an URL prefix for static files

  • folders (list[str]) – Paths to folders with static files

  • app (Optional[TASGIApp]) – 

from asgi_tools import StaticFilesMiddleware, ResponseHTML

async def app(scope, receive, send):
    response = ResponseHTML('OK)
    await response(scope, receive, send)

# Files from static folder will be served from /static
app = StaticFilesMiddleware(app, folders=['static'])

BackgroundMiddleware#

class asgi_tools.BackgroundMiddleware(app=None)[source]#

Run background tasks.

from asgi_tools import BackgroundMiddleware, ResponseText

async def app(scope, receive, send):
    response = ResponseText('OK)

    # Schedule any awaitable for later execution
    BackgroundMiddleware.set_task(asyncio.sleep(1))

    # Return response immediately
    await response(scope, receive, send)

    # The task will be executed after the response is sent

app = BackgroundMiddleware(app)
Parameters:

app (Optional[TASGIApp]) –

Application#

class asgi_tools.App(*, debug=False, logger=<Logger asgi-tools (WARNING)>, static_url_prefix='/static', static_folders=None, trim_last_slash=False)[source]#

A helper to build ASGI Applications.

Features:

  • Routing

  • ASGI-Tools Request, Response

  • Exception management

  • Static files

  • Lifespan events

  • Simplest middlewares

Parameters:

  • debug (bool, False) – Enable debug mode (more logging, raise unhandled exceptions)

  • logger (logging.Logger) – Custom logger for the application

  • static_url_prefix (str, "/static") – A prefix for static files

  • static_folders (list[str]) – A list of folders to look static files

  • trim_last_slash (bool, False) – Consider “/path” and “/path/” as the same

route(*paths, methods=None, **opts)[source]#

Register a route.

Parameters:

  • paths (TPath) –

  • methods (Optional[TMethods]) –

on_startup(fn)[source]#

Register a startup handler.

Parameters:

fn (Callable) –

Return type:

None

on_shutdown(fn)[source]#

Register a shutdown handler.

Parameters:

fn (Callable) –

Return type:

None

on_error(etype)[source]#

Register an exception handler.

@app.on_error(TimeoutError)
async def timeout(request, error):
    return 'Something bad happens'

@app.on_error(ResponseError)
async def process_http_errors(request, response_error):
    if response_error.status_code == 404:
        return render_template('page_not_found.html'), 404
    return response_error
Parameters:

etype (Type[BaseException]) –

middleware(md, *, insert_first=False)[source]#

Register a middleware.

The App.middleware() supports two types of middlewares, see below:

from asgi_tools import App, Request, ResponseError

app = App()

# Classic (any ASGI middleware)
app = classic_middleware(app)

# As an alternative, register a "classic" middleware
@app.middleware
def classic_middleware(app):
    async def handler(scope, receive, send):
        if not Request(scope, receive, send).headers['authorization']:
            response = ResponseError.UNAUTHORIZED()
            await response(scope, receive, send)
        else:
            await app(scope, receive, send)

    return handler

# Register an internal middleware (the middleware function is async)
# The middlewares is guaranted to get a response from app and have to return a response
@app.middleware
async def simple_middleware(app, request, receive, send):
    response = await app(request, receive, send)
    response.headers['x-agent'] = 'SimpleX'
    return response

Middleware Exceptions

Any exception raised from an middleware wouldn’t be catched by the app

Parameters:

  • md (TVCallable) –

  • insert_first (bool) –

Return type:

TVCallable

Class Based Views#

class asgi_tools.HTTPView(request, **opts)[source]#

Class-based view pattern for handling HTTP method dispatching.

@app.route('/custom')
class CustomEndpoint(HTTPView):

    async def get(self, request):
        return 'Hello from GET'

    async def post(self, request):
        return 'Hello from POST'

# ...
async def test_my_endpoint(client):
    response = await client.get('/custom')
    assert await response.text() == 'Hello from GET'

    response = await client.post('/custom')
    assert await response.text() == 'Hello from POST'

    response = await client.put('/custom')
    assert response.status_code == 405
Parameters:

request (Request) –

TestClient#

class asgi_tools.tests.ASGITestClient(app, base_url='http://localhost')[source]#

The test client allows you to make requests against an ASGI application.

Features:

  • cookies

  • multipart/form-data

  • follow redirects

  • request streams

  • response streams

  • websocket support

  • lifespan management

Parameters:

  • app (TASGIApp) –

  • base_url (str) –

async request(path, method='GET', *, query='', headers=None, cookies=None, data=b'', json=None, follow_redirect=True, timeout=10.0)[source]#

Make a HTTP requests.

from asgi_tools import App
from asgi_tools.tests import ASGITestClient

app = Application()

@app.route('/')
async def index(request):
    return 'OK'

async def test_app():
    client = ASGITestClient(app)
    response = await client.get('/')
    assert response.status_code == 200
    assert await response.text() == 'OK'

Stream Request

async def test_app():
    client = ASGITestClient(app)
    async def stream():
        for n in range(10):
            yield b'chunk%s' % bytes(n)
            await aio_sleep(1)

    response = await client.get('/', data=stream)
    assert response.status_code == 200

Stream Response

@app.route('/')
async def index(request):
    async def stream():
        for n in range(10):
            yield b'chunk%s' % bytes(n)
            await aio_sleep(1)

    return ResponseStream(stream)


async def test_app():
    client = ASGITestClient(app)
    response = await client.get('/')
    assert response.status_code == 200
    async for chunk in response.stream():
        assert chunk.startswith('chunk')
Parameters:

  • path (str) –

  • method (str) –

  • query (Union[str, Dict]) –

  • headers (Optional[Dict[str, str]]) –

  • cookies (Optional[Dict[str, str]]) –

  • data (Union[bytes, str, Dict, AsyncGenerator]) –

  • json (TJSON) –

  • follow_redirect (bool) –

  • timeout (float) –

Return type:

TestResponse

websocket(path, query=None, headers=None, cookies=None)[source]#

Connect to a websocket.

from asgi_tools import App, ResponseWebSocket
from asgi_tools.tests import ASGITestClient

app = Application()

@app.route('/websocket')
async def websocket(request):
    async with ResponseWebSocket(request) as ws:
        msg = await ws.receive()
        assert msg == 'ping'
        await ws.send('pong')

async def test_app():
    client = ASGITestClient(app)
    async with client.websocket('/websocket') as ws:
        await ws.send('ping')
        msg = await ws.receive()
        assert msg == 'pong'
Parameters:

  • path (str) –

  • query (str | Dict | None) –

  • headers (Dict | None) –

  • cookies (Dict | None) –

lifespan(timeout=0.03)[source]#

Manage Lifespan protocol.

from asgi_tools import ResponseHTML
from asgi_tools.tests import ASGITestClient

SIDE_EFFECTS = {'started': False, 'finished': False}

async def app(scope, receive, send):
    # Process lifespan events
    if scope['type'] == 'lifespan':
        while True:
            msg = await receive()
            if msg['type'] == 'lifespan.startup':
                SIDE_EFFECTS['started'] = True
                await send({'type': 'lifespan.startup.complete'})

            elif msg['type'] == 'lifespan.shutdown':
                SIDE_EFFECTS['finished'] = True
                await send({'type': 'lifespan.shutdown.complete'})
                return

    # Otherwise return HTML response
    await ResponseHTML('OK')(scope, receive, send)

client = Client(app)

async with client.lifespan():
    assert SIDE_EFFECTS['started']
    assert not SIDE_EFFECTS['finished']
    res = await client.get('/')
    assert res.status_code == 200

assert SIDE_EFFECTS['started']
assert SIDE_EFFECTS['finished']
Parameters:

timeout (float) –