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

Await the given awaitable after the response is completed.

from muffin import Application

app = Application()

@app.task
def send_email(email, message):
    # send email here
    pass

@app.route('/send')
async def send(request):

# Schedule any awaitable for later execution
app.run_after_response(send_email('user@email.com', 'Hello from Muffin!'))

# Return response to a client immediately
# The task will be executed after the response is sent
return "OK"

Parameters:: tasks (Awaitable)

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

Parameters:

paths (TPath) – One or more URL paths to match
methods (Optional[TMethods]) – HTTP methods to match (GET, POST, etc.)
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#

from muffin import Application
from sentry_asgi import SentryMiddleware

app = Application()

app.middleware(SentryMiddleware)

# or wrap directly
app = SentryMiddleware(app)

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

Class-Based Handlers#

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

Class-based view pattern for handling HTTP method dispatching.

@app.route('/hello', '/hello/{name}')
class HelloHandler(Handler):

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

    async def post(self, request):
        name = request.patch_params.get('name') or 'all'
        return "POST: Hello f{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]#: Prepare a media data for the request.

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

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=0, upload_to=None, file_memory_limit=1048576)#

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
file_memory_limit (int) – Maximum size of file to keep in memory

Returns:

Form data as MultiDict

Return type:

MultiDict

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.

Example usage:

@app.route('/')
async def home(request):
    return f"{request.method} {request.url.path}"

Properties:

query – Query parameters (MultiDict)
headers – HTTP headers (CIMultiDict)
cookies – Cookies dictionary
charset – Request encoding
content_type – Content-Type header
url – Full URL object

Methods:

stream() – Stream body in chunks
body() – Return body as bytes
text() – Return body decoded as text
form() – Parse form data (multipart/form-data or application/x-www-form-urlencoded)
json() – Parse body as JSON
data() – Smart parser: returns JSON, form or text depending on Content-Type

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 ujson and orjson JSON libraries. Install one of them to use instead 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#

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:

content (AsyncGenerator) – An async generator to stream the response’s body
stream (AsyncGenerator[Any, None])

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:

content (AsyncGenerator) – An async generator to stream the events
stream (AsyncGenerator[Any, None])

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 (32768)
filename (str) – If set, Content-Disposition header will be generated
headers_only (bool) – Return only file headers

@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.

API#

Application#

Middleware#

Class-Based Handlers#

Request#

Responses#

ResponseText#

ResponseHTML#

ResponseJSON#

ResponseRedirect#

ResponseError#

ResponseStream#

ResponseSSE#

ResponseFile#

ResponseWebSocket#

Test Client#

This Page