API#

This part of the documentation covers the interfaces of Muffin

Application#

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

The Muffin Application.

Parameters:: cfg_mods (Union[str, ModuleType])

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

Parameters:

paths (TPath)
methods (Optional[TMethods])

on_startup(fn)#

Parameters:: fn (Callable)
Return type:: None

on_shutdown(fn)#

Parameters:: fn (Callable)
Return type:: None

on_error(etype)#

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

from muffin import Application
from sentry_asgi import SentryMiddleware

app = Application()

app.middleware(SentryMiddleware)

# as an alternative method
app = SentryMiddleware(app)

from muffin import Application


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

Parameters:

md (TVCallable)
insert_first (bool)

Return type:

TVCallable

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)

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)

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 "GET: Hello f{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)#

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 Request object contains all the information about an incoming HTTP request.

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

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

request = Request(scope)
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']

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

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

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

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.

@app.route('/repeater')
async def repeater(request):
    body = b''
    async for chunk in request.stream():
        body += chunk

    return body

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

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

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

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 muffin.Response(content, *, status_code=None, content_type=None, headers=None, cookies=None)#

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 muffin import Response

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

headers: MultiDict#

Multidict of response’s headers

from muffin import Response

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

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 muffin import Response

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

ResponseText (`Response`)#

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

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

from muffin import ResponseText

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

Parameters:

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

ResponseHTML (`Response`)#

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

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

from muffin import ResponseHTML

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

Note

If your view function returns a string/byte-string the result will be converted into a HTML Response

Parameters:

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

ResponseJSON (`Response`)#

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

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 muffin import ResponseJSON

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

Note

If your view function returns a dictionary/list/boolean the result will be converted into a JSON Response

Parameters:

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

ResponseRedirect (`Response`)#

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

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 muffin import ResponseRedirect

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

You are able to raise the ResponseRedirect as an exception.

from muffin import ResponseRedirect

@app.route('/example')
async def example(request):
    if not request.headers.get('authorization'):
        # The client will be redirected to "/login"
        raise ResponseRedirect('/login')

    return 'OK'

ResponseError (`Response`)#

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

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 muffin import ResponseError

@app.route('/example')
async def example(request):
    return ResponseError('Timeout', 502)

You are able to raise the ResponseError as an exception.

from muffin import ResponseError

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

    return 'OK'

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 (`Response`)#

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

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 muffin import ResponseStream

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

@app.route('/example')
async def example(request):
    generator = stream_response()
    return ResponseStream(generator, content_type='plain/text')

ResponseSSE (`Response`)#

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

A helper to stream SSE (server side events).

Parameters:

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

from muffin import ResponseSSE

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(),
        }

@app.route('/example')
async def example(request):
    generator = stream_response()
    return ResponseSSE(generator, content_type='plain/text')

ResponseFile (`Response`)#

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

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 muffin import ResponseFile

@app.route('/selfie')
async def view_my_selfie(request):
    return ResponseFile('/storage/my_best_selfie.jpeg')

@app.route('/download')
async def download_movie(request):
    return ResponseFile('/storage/video.mp4', filename='movie-2020-01-01.mp4')

ResponseWebSocket (`Response`)#

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

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 muffin import ResponseWebsocket

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

async accept(**params)#

Accept a websocket connection.

Return type:: None

async close(code=1000)#

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

Send the given message to a client.

Parameters:: msg (dict | str | bytes)
Return type:: None

async send_json(data)#

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

Return type:: None

async receive(*, raw=False)#

Receive messages from a client.

Parameters:: raw (bool) – Receive messages as is.
Return type:: Union[TASGIMessage, str]

Test Client#

muffin.TestClient#: alias of ASGITestClient

API#

Application#

Class Based Handlers#

Request#

Responses#

ResponseText (Response)#

ResponseHTML (Response)#

ResponseJSON (Response)#

ResponseRedirect (Response)#

ResponseError (Response)#

ResponseStream (Response)#

ResponseSSE (Response)#

ResponseFile (Response)#

ResponseWebSocket (Response)#

Test Client#

ResponseText (`Response`)#

ResponseHTML (`Response`)#

ResponseJSON (`Response`)#

ResponseRedirect (`Response`)#

ResponseError (`Response`)#

ResponseStream (`Response`)#

ResponseSSE (`Response`)#

ResponseFile (`Response`)#

ResponseWebSocket (`Response`)#