Request¶

Lilya brings the Request class object. This object is a nice interface between the incoming request and the ASGI scope.

This means, you don't need to access directly the scope and extract all the information required for a request type of object.

from lilya.requests import Request

The `Request` class¶

A Request instance receives a scope, a receive and a send parameter.

from lilya.enums import ScopeType
from lilya.requests import Request
from lilya.responses import PlainText
from lilya.types import Receive, Scope, Send


async def app(scope: Scope, receive: Receive, send: Send):
    assert scope["type"] == ScopeType.HTTP

    request = Request(scope=scope, receive=receive, send=send)
    data = f"{request.method} {request.url.path}"

    response = PlainText(content=data)
    await response(scope, receive, send)

The requests, as mentioned before, present an interface to the scope, which means if you use requests['app'] or requests['headers'] or requests['path'] it will retrieve the same information as it was retrieving from the scope.

Note¶

If there is not a need to access the request body, you can instantiate a request without providing a receive argument.

Example

from lilya.requests import Request

request = Request(scope)

Attributes¶

There are many available attributes that you can access within the request.

Method¶

The request method that is used to access.

from lilya.requests import Request

request = Request(scope)

request.method

URL¶

from lilya.requests import Request

request = Request(scope)

request.url

This property exposes all the components that can be parsed out of the URL.

Example

from lilya.requests import Request

request = Request(scope)

request.url.port
request.url.path
request.url.scheme
request.url.netloc
request.url.query

Lilya uses the multidict for its headers and adds some extra flavours on the top of it.

from lilya.requests import Request

request = Request(scope)

request.headers['content-type']

Query Params¶

Lilya uses the multidict for its query parameters and adds some extra flavours on the top of it.

from lilya.requests import Request

request = Request(scope)

request.query_params['search']

Path Params¶

Extracted directly from the scope as a dictionary like python object

from lilya.requests import Request

request = Request(scope)

request.path_params['username']

Client Address¶

The client's remote address is exposed as a dataclass request.client.

from lilya.requests import Request

request = Request(scope)

request.client.host
request.client.port

Cookies¶

Extracted directly from the headers and parsed as a dictionary like python object.

from lilya.requests import Request

request = Request(scope)

request.cookies.get('a-cookie')

Body¶

Now here it is different. To extract and use the body, a receive must be passed into the request instance and it can be extracted in different ways.

As bytes¶

from lilya.requests import Request

request = Request(scope, receive)

await request.body()

As JSON¶

from lilya.requests import Request

request = Request(scope, receive)

await request.json()

As text¶

from lilya.requests import Request

request = Request(scope, receive)

await request.text()

As form data or multipart form¶

from lilya.requests import Request

request = Request(scope, receive)

async with request.form() as form:
    ...

As data¶

from lilya.requests import Request

request = Request(scope, receive)

await request.data()

As a stream¶

from lilya.enums import ScopeType
from lilya.requests import Request
from lilya.responses import PlainText
from lilya.types import Receive, Scope, Send


async def app(scope: Scope, receive: Receive, send: Send):
    assert scope["type"] == ScopeType.HTTP

    request = Request(scope=scope, receive=receive, send=send)
    data = b""

    async for chunk in request.stream():
        data += chunk

    response = PlainText(content=data)
    await response(scope, receive, send)

When employing .stream(), byte chunks are furnished without the necessity of storing the entire body in memory. Subsequent calls to .body(), .form(), or .json() will result in an error.

In specific situations, such as long-polling or streaming responses, it becomes crucial to determine whether the client has disconnected.

This status can be ascertained using the following: disconnected = await request.is_disconnected().

Request files¶

Typically, files are transmitted as multipart form data (multipart/form-data).

from lilya.requests import Request

request = Request(scope, receive)

request.form(max_files=1000, max_fields=1000)

You have the flexibility to set the maximum number of fields or files using the max_files and max_fields parameters:

async with request.form(max_files=1000, max_fields=1000):
    ...

Warning

These limitations serve security purposes. Allowing an unlimited number of fields or files could pose a risk of a denial-of-service attack, consuming excessive CPU and memory resources by parsing numerous empty fields.

When invoking async with request.form() as form, you obtain a lilya.datastructures.FormData, which is an immutable multidict containing both file uploads and text input.

File upload items are represented as instances of lilya.datastructures.DataUpload.

DataUpload¶

DataUpload has the following attributes:

filename: A str with the original file name that was uploaded or None if its not available (e.g. profile.png).
file: A SpooledTemporaryFile (a file-like object). This is the actual Python file that you can pass directly to other functions or libraries that expect a "file-like" object.
headers: A Header object. Often this will only be the Content-Type header, but if additional headers were included in the multipart field they will be included here. Note that these headers have no relationship with the headers in Request.headers.
size: An int with uploaded file's size in bytes. This value is calculated from request's contents, making it better choice to find uploaded file's size than Content-Length header. None if not set.

The DataUpload class provides several asynchronous methods that invoke the corresponding file operations using the internal SpooledTemporaryFile:

async write(data): Writes the specified data (in bytes) to the file.
async read(size): Reads the specified number of bytes (as an integer) from the file.
async seek(offset): Positions the file cursor at the byte offset specified (as an integer). For example, using await profile.seek(0) would move the cursor to the beginning of the file.
async close(): Closes the file.

Since all these methods are asynchronous, the await keyword is necessary when invoking them.

Example

async with request.form() as form:
    filename = form["upload_file"].filename
    contents = await form["upload_file"].read()

Application¶

The Lilya application.

from lilya.requests import Request

request = Request(scope)

request.app

State¶

If you wish to include supplementary information with the request, you can achieve this by using the request.state.

from lilya.requests import Request

request = Request(scope)

request.state.admin = "example@lilya.dev"

Request Lifecycle¶

Lilya supports the concept of request lifecycle. What does this actually mean?

Means that you can add behaviour before the response and after the response.

This can be very useful if for example you want to add some logging, telemety or anything else really.

This also means that you can add behaviour also on every layer of your application, this means, Lilya, Include, Host, Path and Router .

There are two cycles, the before_request and the after_request. All of these available, you guessed, on Lilya, Include, Host, Path and Router objects.

How to use it¶

Like everything in Lilya, this behaves similarly to an ASGI app except you don't need to declare the app parameter like you do in the middleware and permissions.

In fact, you need to declare only a function, sync or async with scope, receive and send as paramters and you don't need to return anything.

Using the function approach¶

There are two ways of making it happening: sync and async.

AsyncSync

from lilya.apps import Lilya
from lilya.routing import Path
from lilya.types import Scope, Receive, Send

async def before_path_request(scope: Scope, receive: Receive, send: Send):
    # Add logging on entering the request
    ...

async def after_path_request(scope: Scope, receive: Receive, send: Send):
    # Add logging on exiting the request
    ...


async def home() -> str:
    return "Hello, World!"


app = Lilya(
    routes=[
        Path("/", home,
            before_request=[before_path_request],
            after_request=[after_path_request],
        ),
    ]
)

from lilya.apps import Lilya
from lilya.routing import Path
from lilya.types import Scope, Receive, Send

def before_path_request(scope: Scope, receive: Receive, send: Send):
    # Add logging on entering the request
    ...

def after_path_request(scope: Scope, receive: Receive, send: Send):
    # Add logging on exiting the request
    ...


async def home() -> str:
    return "Hello, World!"


app = Lilya(
    routes=[
        Path("/", home,
            before_request=[before_path_request],
            after_request=[after_path_request],
        ),
    ]
)

Within levels¶

You can mix with different levels as well, for instance with an Include.

AsyncSync

from lilya.apps import Lilya
from lilya.routing import Path, Include
from lilya.types import Scope, Receive, Send

async def before_path_request(scope: Scope, receive: Receive, send: Send):
    # Add logging on entering the request
    ...

async def after_path_request(scope: Scope, receive: Receive, send: Send):
    # Add logging on exiting the request
    ...


async def before_include_request(scope: Scope, receive: Receive, send: Send):
    # Add logging on entering the include
    ...

async def after_include_request(scope: Scope, receive: Receive, send: Send):
    # Add logging on exiting the include
    ...


async def home() -> str:
    return "Hello, World!"


app = Lilya(
    routes=[
        Include("/",
            before_request=[before_include_request],
            after_request=[after_include_request],
            routes=[
                Path("/", home,
                    before_request=[before_path_request],
                    after_request=[after_path_request],
                ),
            ]
        ),
    ]
)

from lilya.apps import Lilya
from lilya.routing import Path, Include
from lilya.types import Scope, Receive, Send

def before_path_request(scope: Scope, receive: Receive, send: Send):
    # Add logging on entering the request
    ...

def after_path_request(scope: Scope, receive: Receive, send: Send):
    # Add logging on exiting the request
    ...


def before_include_request(scope: Scope, receive: Receive, send: Send):
    # Add logging on entering the include
    ...

def after_include_request(scope: Scope, receive: Receive, send: Send):
    # Add logging on exiting the include
    ...


async def home() -> str:
    return "Hello, World!"


app = Lilya(
    routes=[
        Include("/",
            before_request=[before_include_request],
            after_request=[after_include_request],
            routes=[
                Path("/", home,
                    before_request=[before_path_request],
                    after_request=[after_path_request],
                ),
            ]
        ),
    ]
)

You get the point, don't you? It is this simple and versatile.

Using the class approach¶

In the same way you do for functions, you can apply the same principle using classes and the only thing you need to declare is the __call__(scope, receive, send)__.

Remember: No app is declared since Lilya automatically passes the scope, receive and send to the handler.

AsyncSync

from lilya.apps import Lilya
from lilya.routing import Path
from lilya.types import Scope, Receive, Send


class BeforePathRequest:
    async def __call__(self, scope: Scope, receive: Receive, send: Send):
        # Add logging on entering the request
        ...

class AfterPathRequest:
    async def __call__(self, scope: Scope, receive: Receive, send: Send):
        # Add logging on exiting the request
        ...


async def home() -> str:
    return "Hello, World!"


app = Lilya(
    routes=[
        Path("/", home,
            before_request=[BeforePathRequest],
            after_request=[AfterPathRequest],
        ),
    ]
)

from lilya.apps import Lilya
from lilya.routing import Path
from lilya.types import Scope, Receive, Send


class BeforePathRequest:
    def __call__(self, scope: Scope, receive: Receive, send: Send):
        # Add logging on entering the request
        ...

class AfterPathRequest:
    def __call__(self, scope: Scope, receive: Receive, send: Send):
        # Add logging on exiting the request
        ...


async def home() -> str:
    return "Hello, World!"


app = Lilya(
    routes=[
        Path("/", home,
            before_request=[BeforePathRequest],
            after_request=[AfterPathRequest],
        ),
    ]
)

Within levels¶

You can mix with different levels as well, for instance with an Include.

AsyncSync

from lilya.apps import Lilya
from lilya.routing import Path, Include
from lilya.types import Scope, Receive, Send


class BeforePathRequest:
    async def __call__(self, scope: Scope, receive: Receive, send: Send):
        # Add logging on entering the request
        ...

class AfterPathRequest:
    async def __call__(self, scope: Scope, receive: Receive, send: Send):
        # Add logging on exiting the request
        ...

class BeforeIncludeRequest:
    async def __call__(self, scope: Scope, receive: Receive, send: Send):
        # Add logging on entering the include
        ...

class AfterIncludeRequest:
    async def __call__(self, scope: Scope, receive: Receive, send: Send):
        # Add logging on exiting the include
        ...


async def home() -> str:
    return "Hello, World!"


app = Lilya(
    routes=[
        Include("/",
            before_request=[BeforeIncludeRequest],
            after_request=[AfterIncludeRequest],
            routes=[
                Path("/", home,
                    before_request=[BeforePathRequest],
                    after_request=[AfterPathRequest],
                ),
            ]
        ),
    ]
)

from lilya.apps import Lilya
from lilya.routing import Path, Include
from lilya.types import Scope, Receive, Send


class BeforePathRequest:
    def __call__(self, scope: Scope, receive: Receive, send: Send):
        # Add logging on entering the request
        ...

class AfterPathRequest:
    def __call__(self, scope: Scope, receive: Receive, send: Send):
        # Add logging on exiting the request
        ...

class BeforeIncludeRequest:
    def __call__(self, scope: Scope, receive: Receive, send: Send):
        # Add logging on entering the include
        ...

class AfterIncludeRequest:
    def __call__(self, scope: Scope, receive: Receive, send: Send):
        # Add logging on exiting the include
        ...


async def home() -> str:
    return "Hello, World!"


app = Lilya(
    routes=[
        Include("/",
            before_request=[BeforeIncludeRequest],
            after_request=[AfterIncludeRequest],
            routes=[
                Path("/", home,
                    before_request=[BeforePathRequest],
                    after_request=[AfterPathRequest],
                ),
            ]
        ),
    ]
)

You get the point, don't you? This is also this simple.

Notes¶

The before_request and after_request cycles are lists of callables, which means that you can have multiple callables within the same level and it will be called by the same order given in the list.

Example: [CallableOne, CallableTwo, CallableThree] will execute from left to right.

Request¶

The Request class¶

Note¶

Attributes¶

Method¶

URL¶

Header¶

Query Params¶

Path Params¶

Client Address¶

Cookies¶

Body¶

As bytes¶

As JSON¶

As text¶

As form data or multipart form¶

As data¶

As a stream¶

Request files¶

DataUpload¶

Application¶

State¶

Request Lifecycle¶

How to use it¶

Using the function approach¶

Within levels¶

Using the class approach¶

Within levels¶

Notes¶

The `Request` class¶