API#

This section documents all public interfaces of Muffin.

Application#

class muffin.Application(*cfg_mods, **options)#

The Muffin Application.

Parameters:: cfg_mods (str | ModuleType)

import_submodules(*submodules, silent=False, **kwargs)#

Automatically import submodules.

# Somewhere in package "__init__.py" file

# import all submodules
app.import_submodules()

# import only specific submodules (in specified order)
app.import_submodules('submodule1', 'submodule2')

# ignore import errors
app.import_submodules(silent=True)

Parameters:

submodules (str)
silent (bool)

run_after_response(*tasks)#

Schedule awaitables to run after the response is sent.

The response is sent first, then the tasks run to completion before the request cycle ends. A task that raises is logged and does not affect the already-sent response.

Call it only from a request handler. Scheduling from a background task has no effect: “after the response” has already passed by then, so the task is never awaited and its coroutine raises a RuntimeWarning when it is garbage-collected.

from muffin import Application

app = Application()

async def send_email(email, message):
    # send email here
    ...

@app.route('/send')
async def send(request):
    app.run_after_response(send_email('user@email.com', 'Hello!'))
    return 'OK'

Parameters:: tasks (Awaitable)

route(*paths, methods=None, hosts=None, **opts)#

Parameters:

paths (TPath) – One or more URL paths to match
methods (Optional[TMethods]) – HTTP methods to match (GET, POST, etc.)
hosts (Optional[THostsArg]) – Restrict the route to the given host(s), compared case-insensitively and ignoring the port. Requires use_hosts=True on the application
opts (Any) – Additional options for the route

Returns:

Decorator function

Return type:

Callable[[TRouteHandler], TRouteHandler]

on_startup(fn)#

Parameters:: fn (Callable) – The function to call on startup
Return type:: None

on_shutdown(fn)#

Parameters:: fn (Callable) – The function to call on shutdown
Return type:: None

on_error(etype)#

Parameters:: etype (type[BaseException]) – The exception type to handle
Returns:: A decorator to register the handler
Return type:: Callable

Example:

>>> @app.on_error(TimeoutError)
... async def timeout_handler(request, error):
...     return 'Timeout occurred'

middleware(md, *, insert_first=False)#

Middleware can be a coroutine (request/response) or a regular callable (lifespan).

Parameters:

md (TCallable) – The middleware function or class
insert_first (bool) – If True, insert as the first middleware

Returns:

The registered middleware

Return type:

TCallable

Example:

>>> @app.middleware
... async def log_requests(handler, request):
...     print(f"Request: {request.method} {request.url}")
...     return await handler(request)

Middleware#

A coroutine middleware receives the app callable and the raw ASGI arguments:

from muffin import Application, ResponseHTML

app = Application()

@app.middleware
async def simple_md(app, request, receive, send):
    try:
        response = await app(request, receive, send)
        response.headers["x-simple-md"] = "passed"
        return response
    except RuntimeError:
        return ResponseHTML("Middleware Exception")

Anything else passed to app.middleware() is treated as ASGI middleware and bound to the lifespan, which is what most third-party ASGI packages expect:

from sentry_asgi import SentryMiddleware

app.middleware(SentryMiddleware)

See Usage for the request/response details.

Plugins#

class muffin.plugins.BasePlugin(app=None, **options)#

Base class for Muffin plugins.

Parameters:: app (Application | None)

setup(app, *, name=None, **options)#

Bind app and update the plugin’s configuration.

Precedence (lowest to highest): plugin defaults, application config, options, environment variables.

Parameters:

app (Application)
name (str | None)

Return type:

bool

async startup()#: Startup the plugin.

async shutdown()#: Shutdown the plugin.

async restart()#: Restart the plugin.

conftest()#: Return a context manager for pytest conftest.py.

The plugin contract is described in Plugins.

Class-Based Handlers#

class muffin.Handler(request, **opts)#

Class-based view pattern for handling HTTP method dispatching.

Methods must be set explicitly (methods) or inherited from a parent class. There is no automatic detection: defining async def get does not register GET on its own. A handler without methods reaches every request, and methods it does not implement end in a logged 500 instead of 405, so declare them.

@app.route('/hello', '/hello/{name}')
class HelloHandler(Handler):
    methods = ['get', 'post']

    async def get(self, request):
        name = request.path_params.get('name') or 'all'
        return f"GET: Hello {name}"

    async def post(self, request):
        name = request.path_params.get('name') or 'all'
        return f"POST: Hello {name}"

    @Handler.route('/hello/custom')
    async def custom(self, request):
        return 'Custom HELLO'

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

    response = await client.get('/hello/john')
    assert await response.text() == 'GET: Hello john'

    response = await client.post('/hello')
    assert await response.text() == 'POST: Hello all'

    response = await client.get('/hello/custom')
    assert await response.text() == 'Custom HELLO'

    response = await client.delete('/hello')
    assert response.status_code == 405

Parameters:: request (Request)

Request#

class muffin.Request(scope, receive, send)#

Provides a convenient, high-level interface for incoming HTTP requests.

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

scope#

receive#

send#

property url: 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.

property headers: CIMultiDict#

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.

property cookies: dict[str, str]#

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

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

property media: Mapping[str, str]#: A lazy property that parses the request’s Content-Type header into a mapping of media options, with the media type under the content_type key (e.g. {'content_type': 'text/html', 'charset': 'utf-8'}).

property charset: str#: Get an encoding charset for the current scope.

property query: MultiDictProxy[str]#: A lazy property that parse the current query string and returns it as a multidict.MultiDict.

property content_type: str#: Get a content type for the current scope.

async stream()#

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.

Return type:: AsyncGenerator[bytes, None]

async body()#

Read and return the request’s body as bytes.

body = await request.body()

Return type:: bytes

async text()#

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

text = await request.text()

Return type:: str

async json()#

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 form(max_size=10485760, upload_to=None)#

Read and return the request’s form data.

Parameters:

max_size (int) – Maximum size of the form data in bytes
upload_to (Optional[UploadHandler]) – Callable to handle file uploads

Returns:

Form data as MultiDict

Return type:

MultiDict

Raises:

ASGIRequestEntityTooLargeError – If the form data exceeds max_size. App turns this into an HTTP 413 response.

formdata = await request.form()

async data(*, raise_errors=False)#

Read and return the request’s data based on content type.

Parameters:: raise_errors (bool) – Raise an error if the given data is invalid.
Returns:: Request data in appropriate format
Return type:: Union[str, bytes, MultiDict, TJSON]
Raises:: ASGIDecodeError – If data cannot be decoded and raise_errors is True

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.

A request carries the parsed URL, headers, cookies and path parameters:

@app.route('/hello/{name}')
async def hello(request):
    return f"{request.method} {request.url.path} {request.path_params['name']}"

The body is read once, with one of the awaitable parsers (stream, body, text, json, form, data); Usage covers which to use.

Responses#

class muffin.Response(content, *, status_code=None, content_type=None, headers=None, cookies=None)#

Base class for creating HTTP responses.

Parameters:

content (str | bytes | dict | list | None) – 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

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’)

Basic response example:

from muffin import Response

@app.route('/hello')
async def hello(request):
    return Response('Hello, world!', content_type='text/plain')

Setting headers:

@app.route('/example')
async def example(request):
    response = Response('OK')
    response.headers["X-Version"] = "42"
    return response

Setting cookies:

@app.route('/example')
async def example(request):
    response = Response('OK')
    response.cookies["session"] = "xyz"
    response.cookies["session"]["path"] = "/"
    return response

ResponseText#

class muffin.ResponseText(content, *, status_code=None, content_type=None, headers=None, cookies=None)#

Returns plain text responses (text/plain).

Parameters:

content (TContent)
status_code (int)
content_type (str | None)
headers (MultiDict)
cookies (SimpleCookie)

@app.route('/example')
async def example(request):
    return ResponseText('Hello, world!')

ResponseHTML#

class muffin.ResponseHTML(content, *, status_code=None, content_type=None, headers=None, cookies=None)#

Returns HTML responses (text/html).

Parameters:

content (TContent)
status_code (int)
content_type (str | None)
headers (MultiDict)
cookies (SimpleCookie)

@app.route('/example')
async def example(request):
    return ResponseHTML('<h1>Hello, world!</h1>')

Note

Returning a string/bytes will automatically produce an HTML response.

ResponseJSON#

class muffin.ResponseJSON(content, *, status_code=None, content_type=None, headers=None, cookies=None)#

Returns JSON responses (application/json).

The class optionally supports orjson for faster JSON handling. Install it to use instead of the standard library.

Parameters:

content (TContent)
status_code (int)
content_type (str | None)
headers (MultiDict)
cookies (SimpleCookie)

@app.route('/example')
async def example(request):
    return ResponseJSON({'hello': 'world'})

Note

Returning dicts, lists, or booleans produces a JSON response.

ResponseRedirect#

class muffin.ResponseRedirect(url, status_code=None, **kwargs)#

Creates HTTP redirects. Uses a 307 status code by default.

Parameters:

url (str) – A string with the new location
status_code (int)

@app.route('/example')
async def example(request):
    return ResponseRedirect('/login')

Alternatively, raise as an exception:

@app.route('/example')
async def example(request):
    if not request.cookies.get('session'):
        raise ResponseRedirect('/login')
    return 'OK'

ResponseError#

class muffin.ResponseError(message=None, status_code=None, **kwargs)#

Helper for returning HTTP errors. Uses a 500 status code by default.

Parameters:

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

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

Raise HTTP errors:

@app.route('/example')
async def example(request):
    data = await request.data()
    if not data:
        raise ResponseError('Invalid request data', 400)
    return 'OK'

Supports http.HTTPStatus shortcuts:

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

ResponseStream#

class muffin.ResponseStream(stream, **kwargs)#

Streams response body as chunks.

Parameters:: stream (AsyncGenerator) – An async generator to stream the response’s body

from muffin import ResponseStream

async def stream_response():
    for i in range(10):
        await sleep(1)
        yield f"chunk {i}"

@app.route('/stream')
async def stream(request):
    return ResponseStream(stream_response())

ResponseSSE#

class muffin.ResponseSSE(stream, **kwargs)#

Streams Server-Sent Events (SSE).

Parameters:: stream (AsyncGenerator) – An async generator to stream the events

from muffin import ResponseSSE

async def sse_response():
    while True:
        await sleep(1)
        yield {"event": "ping", "data": "pong"}

@app.route('/sse')
async def sse(request):
    return ResponseSSE(sse_response())

ResponseFile#

class muffin.ResponseFile(filepath, *, chunk_size=65536, filename=None, headers_only=False, **kwargs)#

Serves files as HTTP responses.

Parameters:

filepath (str | Path) – The filepath to the file
chunk_size (int) – Default chunk size (65536)
filename (str) – If set, Content-Disposition header will be generated
headers_only (bool) – Return only file headers

Note

Under asyncio, serving files requires the optional aiofile package (pip install aiofile). Trio and Curio use their native async file I/O and need nothing extra.

@app.route('/download')
async def download(request):
    return ResponseFile('/path/to/file.txt', filename='file.txt')

ResponseWebSocket#

class muffin.ResponseWebSocket(scope, receive=None, send=None)#

Provides a WebSocket handler interface.

Parameters:

scope (TASGIScope)
receive (TASGIReceive | None)
send (TASGISend | None)

@app.route('/ws')
async def websocket(request):
    async with ResponseWebSocket(request) as ws:
        msg = await ws.receive()
        await ws.send(f"Echo: {msg}")

Test Client#

muffin.TestClient#: alias of ASGITestClient

Use for testing your Muffin applications efficiently.