Skip to content

Instantly share code, notes, and snippets.

@jph00

jph00/starlette.md

Created June 26, 2024 05:23
Show Gist options
  • Save jph00/07913f47c17be29794fa38ef203b52a9 to your computer and use it in GitHub Desktop.
Save jph00/07913f47c17be29794fa38ef203b52a9 to your computer and use it in GitHub Desktop.
Download ZIP
Starlette docs
Raw
starlette.md

index.md

starlette

✨ The little ASGI framework that shines. ✨

Build Status Package version

Introduction

Starlette is a lightweight ASGI framework/toolkit, which is ideal for building async web services in Python.

It is production-ready, and gives you the following:

  • A lightweight, low-complexity HTTP web framework.
  • WebSocket support.
  • In-process background tasks.
  • Startup and shutdown events.
  • Test client built on httpx.
  • CORS, GZip, Static Files, Streaming responses.
  • Session and Cookie support.
  • 100% test coverage.
  • 100% type annotated codebase.
  • Few hard dependencies.
  • Compatible with asyncio and trio backends.
  • Great overall performance against independent benchmarks.

Requirements

Python 3.8+

Installation

$ pip3 install starlette

You'll also want to install an ASGI server, such as uvicorn, daphne, or hypercorn.

$ pip3 install uvicorn

Example

example.py:

from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route


async def homepage(request):
    return JSONResponse({'hello': 'world'})


app = Starlette(debug=True, routes=[
    Route('/', homepage),
])

Then run the application...

$ uvicorn example:app

For a more complete example, see here.

Dependencies

Starlette only requires anyio, and the following dependencies are optional:

  • httpx - Required if you want to use the TestClient.
  • jinja2 - Required if you want to use Jinja2Templates.
  • python-multipart - Required if you want to support form parsing, with request.form().
  • itsdangerous - Required for SessionMiddleware support.
  • pyyaml - Required for SchemaGenerator support.

You can install all of these with pip3 install starlette[full].

Framework or Toolkit

Starlette is designed to be used either as a complete framework, or as an ASGI toolkit. You can use any of its components independently.

from starlette.responses import PlainTextResponse


async def app(scope, receive, send):
    assert scope['type'] == 'http'
    response = PlainTextResponse('Hello, world!')
    await response(scope, receive, send)

Run the app application in example.py:

$ uvicorn example:app
INFO: Started server process [11509]
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

Run uvicorn with --reload to enable auto-reloading on code changes.

Modularity

The modularity that Starlette is designed on promotes building re-usable components that can be shared between any ASGI framework. This should enable an ecosystem of shared middleware and mountable applications.

The clean API separation also means it's easier to understand each component in isolation.

Starlette is BSD licensed code.
Designed & crafted with care.
— ⭐️ —

applications.md

Starlette includes an application class Starlette that nicely ties together all of its other functionality.

from starlette.applications import Starlette
from starlette.responses import PlainTextResponse
from starlette.routing import Route, Mount, WebSocketRoute
from starlette.staticfiles import StaticFiles


def homepage(request):
    return PlainTextResponse('Hello, world!')

def user_me(request):
    username = "John Doe"
    return PlainTextResponse('Hello, %s!' % username)

def user(request):
    username = request.path_params['username']
    return PlainTextResponse('Hello, %s!' % username)

async def websocket_endpoint(websocket):
    await websocket.accept()
    await websocket.send_text('Hello, websocket!')
    await websocket.close()

def startup():
    print('Ready to go')


routes = [
    Route('/', homepage),
    Route('/user/me', user_me),
    Route('/user/{username}', user),
    WebSocketRoute('/ws', websocket_endpoint),
    Mount('/static', StaticFiles(directory="static")),
]

app = Starlette(debug=True, routes=routes, on_startup=[startup])

Instantiating the application

::: starlette.applications.Starlette :docstring:

Storing state on the app instance

You can store arbitrary extra state on the application instance, using the generic app.state attribute.

For example:

app.state.ADMIN_EMAIL = 'admin@example.org'

Accessing the app instance

Where a request is available (i.e. endpoints and middleware), the app is available on request.app.

requests.md

Starlette includes a Request class that gives you a nicer interface onto the incoming request, rather than accessing the ASGI scope and receive channel directly.

Request

Signature: Request(scope, receive=None)

from starlette.requests import Request
from starlette.responses import Response


async def app(scope, receive, send):
    assert scope['type'] == 'http'
    request = Request(scope, receive)
    content = '%s %s' % (request.method, request.url.path)
    response = Response(content, media_type='text/plain')
    await response(scope, receive, send)

Requests present a mapping interface, so you can use them in the same way as a scope.

For instance: request['path'] will return the ASGI path.

If you don't need to access the request body you can instantiate a request without providing an argument to receive.

Method

The request method is accessed as request.method.

URL

The request URL is accessed as request.url.

The property is a string-like object that exposes all the components that can be parsed out of the URL.

For example: request.url.path, request.url.port, request.url.scheme.

Headers

Headers are exposed as an immutable, case-insensitive, multi-dict.

For example: request.headers['content-type']

Query Parameters

Query parameters are exposed as an immutable multi-dict.

For example: request.query_params['search']

Path Parameters

Router path parameters are exposed as a dictionary interface.

For example: request.path_params['username']

Client Address

The client's remote address is exposed as a named two-tuple request.client (or None).

The hostname or IP address: request.client.host

The port number from which the client is connecting: request.client.port

Cookies

Cookies are exposed as a regular dictionary interface.

For example: request.cookies.get('mycookie')

Cookies are ignored in case of an invalid cookie. (RFC2109)

Body

There are a few different interfaces for returning the body of the request:

The request body as bytes: await request.body()

The request body, parsed as form data or multipart: async with request.form() as form:

The request body, parsed as JSON: await request.json()

You can also access the request body as a stream, using the async for syntax:

from starlette.requests import Request
from starlette.responses import Response

    
async def app(scope, receive, send):
    assert scope['type'] == 'http'
    request = Request(scope, receive)
    body = b''
    async for chunk in request.stream():
        body += chunk
    response = Response(body, media_type='text/plain')
    await response(scope, receive, send)

If you access .stream() then the byte chunks are provided without storing the entire body to memory. Any subsequent calls to .body(), .form(), or .json() will raise an error.

In some cases such as long-polling, or streaming responses you might need to determine if the client has dropped the connection. You can determine this state with disconnected = await request.is_disconnected().

Request Files

Request files are normally sent as multipart form data (multipart/form-data).

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

You can configure the number of maximum fields or files with the parameters max_files and max_fields:

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

!!! info These limits are for security reasons, allowing an unlimited number of fields or files could lead to a denial of service attack by consuming a lot of CPU and memory parsing too many empty fields.

When you call async with request.form() as form you receive a starlette.datastructures.FormData which is an immutable multidict, containing both file uploads and text input. File upload items are represented as instances of starlette.datastructures.UploadFile.

UploadFile has the following attributes:

  • filename: An str with the original file name that was uploaded or None if its not available (e.g. myimage.jpg).
  • content_type: An str with the content type (MIME type / media type) or None if it's not available (e.g. image/jpeg).
  • 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 Headers 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.

UploadFile has the following async methods. They all call the corresponding file methods underneath (using the internal SpooledTemporaryFile).

  • async write(data): Writes data (bytes) to the file.
  • async read(size): Reads size (int) bytes of the file.
  • async seek(offset): Goes to the byte position offset (int) in the file.
    • E.g., await myfile.seek(0) would go to the start of the file.
  • async close(): Closes the file.

As all these methods are async methods, you need to "await" them.

For example, you can get the file name and the contents with:

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

!!! info As settled in RFC-7578: 4.2, form-data content part that contains file assumed to have name and filename fields in Content-Disposition header: Content-Disposition: form-data; name="user"; filename="somefile". Though filename field is optional according to RFC-7578, it helps Starlette to differentiate which data should be treated as file. If filename field was supplied, UploadFile object will be created to access underlying file, otherwise form-data part will be parsed and available as a raw string.

Application

The originating Starlette application can be accessed via request.app.

Other state

If you want to store additional information on the request you can do so using request.state.

For example:

request.state.time_started = time.time()

responses.md

Starlette includes a few response classes that handle sending back the appropriate ASGI messages on the send channel.

Response

Signature: Response(content, status_code=200, headers=None, media_type=None)

  • content - A string or bytestring.
  • status_code - An integer HTTP status code.
  • headers - A dictionary of strings.
  • media_type - A string giving the media type. eg. "text/html"

Starlette will automatically include a Content-Length header. It will also include a Content-Type header, based on the media_type and appending a charset for text types, unless a charset has already been specified in the media_type.

Once you've instantiated a response, you can send it by calling it as an ASGI application instance.

from starlette.responses import Response


async def app(scope, receive, send):
    assert scope['type'] == 'http'
    response = Response('Hello, world!', media_type='text/plain')
    await response(scope, receive, send)

Set Cookie

Starlette provides a set_cookie method to allow you to set cookies on the response object.

Signature: Response.set_cookie(key, value, max_age=None, expires=None, path="/", domain=None, secure=False, httponly=False, samesite="lax")

  • key - A string that will be the cookie's key.
  • value - A string that will be the cookie's value.
  • max_age - An integer that defines the lifetime of the cookie in seconds. A negative integer or a value of 0 will discard the cookie immediately. Optional
  • expires - Either an integer that defines the number of seconds until the cookie expires, or a datetime. Optional
  • path - A string that specifies the subset of routes to which the cookie will apply. Optional
  • domain - A string that specifies the domain for which the cookie is valid. Optional
  • secure - A bool indicating that the cookie will only be sent to the server if request is made using SSL and the HTTPS protocol. Optional
  • httponly - A bool indicating that the cookie cannot be accessed via JavaScript through Document.cookie property, the XMLHttpRequest or Request APIs. Optional
  • samesite - A string that specifies the samesite strategy for the cookie. Valid values are 'lax', 'strict' and 'none'. Defaults to 'lax'. Optional

Delete Cookie

Conversely, Starlette also provides a delete_cookie method to manually expire a set cookie.

Signature: Response.delete_cookie(key, path='/', domain=None)

HTMLResponse

Takes some text or bytes and returns an HTML response.

from starlette.responses import HTMLResponse


async def app(scope, receive, send):
    assert scope['type'] == 'http'
    response = HTMLResponse('<html><body><h1>Hello, world!</h1></body></html>')
    await response(scope, receive, send)

PlainTextResponse

Takes some text or bytes and returns a plain text response.

from starlette.responses import PlainTextResponse


async def app(scope, receive, send):
    assert scope['type'] == 'http'
    response = PlainTextResponse('Hello, world!')
    await response(scope, receive, send)

JSONResponse

Takes some data and returns an application/json encoded response.

from starlette.responses import JSONResponse


async def app(scope, receive, send):
    assert scope['type'] == 'http'
    response = JSONResponse({'hello': 'world'})
    await response(scope, receive, send)

Custom JSON serialization

If you need fine-grained control over JSON serialization, you can subclass JSONResponse and override the render method.

For example, if you wanted to use a third-party JSON library such as orjson:

from typing import Any

import orjson
from starlette.responses import JSONResponse


class OrjsonResponse(JSONResponse):
    def render(self, content: Any) -> bytes:
        return orjson.dumps(content)

In general you probably want to stick with JSONResponse by default unless you are micro-optimising a particular endpoint or need to serialize non-standard object types.

RedirectResponse

Returns an HTTP redirect. Uses a 307 status code by default.

from starlette.responses import PlainTextResponse, RedirectResponse


async def app(scope, receive, send):
    assert scope['type'] == 'http'
    if scope['path'] != '/':
        response = RedirectResponse(url='/')
    else:
        response = PlainTextResponse('Hello, world!')
    await response(scope, receive, send)

StreamingResponse

Takes an async generator or a normal generator/iterator and streams the response body.

from starlette.responses import StreamingResponse
import asyncio


async def slow_numbers(minimum, maximum):
    yield '<html><body><ul>'
    for number in range(minimum, maximum + 1):
        yield '<li>%d</li>' % number
        await asyncio.sleep(0.5)
    yield '</ul></body></html>'


async def app(scope, receive, send):
    assert scope['type'] == 'http'
    generator = slow_numbers(1, 10)
    response = StreamingResponse(generator, media_type='text/html')
    await response(scope, receive, send)

Have in mind that file-like objects (like those created by open()) are normal iterators. So, you can return them directly in a StreamingResponse.

FileResponse

Asynchronously streams a file as the response.

Takes a different set of arguments to instantiate than the other response types:

  • path - The filepath to the file to stream.
  • headers - Any custom headers to include, as a dictionary.
  • media_type - A string giving the media type. If unset, the filename or path will be used to infer a media type.
  • filename - If set, this will be included in the response Content-Disposition.
  • content_disposition_type - will be included in the response Content-Disposition. Can be set to "attachment" (default) or "inline".

File responses will include appropriate Content-Length, Last-Modified and ETag headers.

from starlette.responses import FileResponse


async def app(scope, receive, send):
    assert scope['type'] == 'http'
    response = FileResponse('statics/favicon.ico')
    await response(scope, receive, send)

Third party responses

EventSourceResponse

A response class that implements Server-Sent Events. It enables event streaming from the server to the client without the complexity of websockets.

baize.asgi.FileResponse

As a smooth replacement for Starlette FileResponse, it will automatically handle Head method and Range requests.

websockets.md

Starlette includes a WebSocket class that fulfils a similar role to the HTTP request, but that allows sending and receiving data on a websocket.

WebSocket

Signature: WebSocket(scope, receive=None, send=None)

from starlette.websockets import WebSocket


async def app(scope, receive, send):
    websocket = WebSocket(scope=scope, receive=receive, send=send)
    await websocket.accept()
    await websocket.send_text('Hello, world!')
    await websocket.close()

WebSockets present a mapping interface, so you can use them in the same way as a scope.

For instance: websocket['path'] will return the ASGI path.

URL

The websocket URL is accessed as websocket.url.

The property is actually a subclass of str, and also exposes all the components that can be parsed out of the URL.

For example: websocket.url.path, websocket.url.port, websocket.url.scheme.

Headers

Headers are exposed as an immutable, case-insensitive, multi-dict.

For example: websocket.headers['sec-websocket-version']

Query Parameters

Query parameters are exposed as an immutable multi-dict.

For example: websocket.query_params['search']

Path Parameters

Router path parameters are exposed as a dictionary interface.

For example: websocket.path_params['username']

Accepting the connection

  • await websocket.accept(subprotocol=None, headers=None)

Sending data

  • await websocket.send_text(data)
  • await websocket.send_bytes(data)
  • await websocket.send_json(data)

JSON messages default to being sent over text data frames, from version 0.10.0 onwards. Use websocket.send_json(data, mode="binary") to send JSON over binary data frames.

Receiving data

  • await websocket.receive_text()
  • await websocket.receive_bytes()
  • await websocket.receive_json()

May raise starlette.websockets.WebSocketDisconnect().

JSON messages default to being received over text data frames, from version 0.10.0 onwards. Use websocket.receive_json(data, mode="binary") to receive JSON over binary data frames.

Iterating data

  • websocket.iter_text()
  • websocket.iter_bytes()
  • websocket.iter_json()

Similar to receive_text, receive_bytes, and receive_json but returns an async iterator.

from starlette.websockets import WebSocket


async def app(scope, receive, send):
    websocket = WebSocket(scope=scope, receive=receive, send=send)
    await websocket.accept()
    async for message in websocket.iter_text():
        await websocket.send_text(f"Message text was: {message}")
    await websocket.close()

When starlette.websockets.WebSocketDisconnect is raised, the iterator will exit.

Closing the connection

  • await websocket.close(code=1000, reason=None)

Sending and receiving messages

If you need to send or receive raw ASGI messages then you should use websocket.send() and websocket.receive() rather than using the raw send and receive callables. This will ensure that the websocket's state is kept correctly updated.

  • await websocket.send(message)
  • await websocket.receive()

Send Denial Response

If you call websocket.close() before calling websocket.accept() then the server will automatically send a HTTP 403 error to the client.

If you want to send a different error response, you can use the websocket.send_denial_response() method. This will send the response and then close the connection.

  • await websocket.send_denial_response(response)

This requires the ASGI server to support the WebSocket Denial Response extension. If it is not supported a RuntimeError will be raised.

routing.md

HTTP Routing

Starlette has a simple but capable request routing system. A routing table is defined as a list of routes, and passed when instantiating the application.

from starlette.applications import Starlette
from starlette.responses import PlainTextResponse
from starlette.routing import Route


async def homepage(request):
    return PlainTextResponse("Homepage")

async def about(request):
    return PlainTextResponse("About")


routes = [
    Route("/", endpoint=homepage),
    Route("/about", endpoint=about),
]

app = Starlette(routes=routes)

The endpoint argument can be one of:

  • A regular function or async function, which accepts a single request argument and which should return a response.
  • A class that implements the ASGI interface, such as Starlette's HTTPEndpoint.

Path Parameters

Paths can use URI templating style to capture path components.

Route('/users/{username}', user)

By default this will capture characters up to the end of the path or the next /.

You can use convertors to modify what is captured. The available convertors are:

  • str returns a string, and is the default.
  • int returns a Python integer.
  • float returns a Python float.
  • uuid return a Python uuid.UUID instance.
  • path returns the rest of the path, including any additional / characters.

Convertors are used by prefixing them with a colon, like so:

Route('/users/{user_id:int}', user)
Route('/floating-point/{number:float}', floating_point)
Route('/uploaded/{rest_of_path:path}', uploaded)

If you need a different converter that is not defined, you can create your own. See below an example on how to create a datetime convertor, and how to register it:

from datetime import datetime

from starlette.convertors import Convertor, register_url_convertor


class DateTimeConvertor(Convertor):
    regex = "[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}(.[0-9]+)?"

    def convert(self, value: str) -> datetime:
        return datetime.strptime(value, "%Y-%m-%dT%H:%M:%S")

    def to_string(self, value: datetime) -> str:
        return value.strftime("%Y-%m-%dT%H:%M:%S")

register_url_convertor("datetime", DateTimeConvertor())

After registering it, you'll be able to use it as:

Route('/history/{date:datetime}', history)

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

async def user(request):
    user_id = request.path_params['user_id']
    ...

Handling HTTP methods

Routes can also specify which HTTP methods are handled by an endpoint:

Route('/users/{user_id:int}', user, methods=["GET", "POST"])

By default function endpoints will only accept GET requests, unless specified.

Submounting routes

In large applications you might find that you want to break out parts of the routing table, based on a common path prefix.

routes = [
    Route('/', homepage),
    Mount('/users', routes=[
        Route('/', users, methods=['GET', 'POST']),
        Route('/{username}', user),
    ])
]

This style allows you to define different subsets of the routing table in different parts of your project.

from myproject import users, auth

routes = [
    Route('/', homepage),
    Mount('/users', routes=users.routes),
    Mount('/auth', routes=auth.routes),
]

You can also use mounting to include sub-applications within your Starlette application. For example...

# This is a standalone static files server:
app = StaticFiles(directory="static")

# This is a static files server mounted within a Starlette application,
# underneath the "/static" path.
routes = [
    ...
    Mount("/static", app=StaticFiles(directory="static"), name="static")
]

app = Starlette(routes=routes)

Reverse URL lookups

You'll often want to be able to generate the URL for a particular route, such as in cases where you need to return a redirect response.

  • Signature: url_for(name, **path_params) -> URL
routes = [
    Route("/", homepage, name="homepage")
]

# We can use the following to return a URL...
url = request.url_for("homepage")

URL lookups can include path parameters...

routes = [
    Route("/users/{username}", user, name="user_detail")
]

# We can use the following to return a URL...
url = request.url_for("user_detail", username=...)

If a Mount includes a name, then submounts should use a {prefix}:{name} style for reverse URL lookups.

routes = [
    Mount("/users", name="users", routes=[
        Route("/", user, name="user_list"),
        Route("/{username}", user, name="user_detail")
    ])
]

# We can use the following to return URLs...
url = request.url_for("users:user_list")
url = request.url_for("users:user_detail", username=...)

Mounted applications may include a path=... parameter.

routes = [
    ...
    Mount("/static", app=StaticFiles(directory="static"), name="static")
]

# We can use the following to return URLs...
url = request.url_for("static", path="/css/base.css")

For cases where there is no request instance, you can make reverse lookups against the application, although these will only return the URL path.

url = app.url_path_for("user_detail", username=...)

Host-based routing

If you want to use different routes for the same path based on the Host header.

Note that port is removed from the Host header when matching. For example, Host (host='example.org:3600', ...) will be processed even if the Host header contains or does not contain a port other than 3600 (example.org:5600, example.org). Therefore, you can specify the port if you need it for use in url_for.

There are several ways to connect host-based routes to your application

site = Router()  # Use eg. `@site.route()` to configure this.
api = Router()  # Use eg. `@api.route()` to configure this.
news = Router()  # Use eg. `@news.route()` to configure this.

routes = [
    Host('api.example.org', api, name="site_api")
]

app = Starlette(routes=routes)

app.host('www.example.org', site, name="main_site")

news_host = Host('news.example.org', news)
app.router.routes.append(news_host)

URL lookups can include host parameters just like path parameters

routes = [
    Host("{subdomain}.example.org", name="sub", app=Router(routes=[
        Mount("/users", name="users", routes=[
            Route("/", user, name="user_list"),
            Route("/{username}", user, name="user_detail")
        ])
    ]))
]
...
url = request.url_for("sub:users:user_detail", username=..., subdomain=...)
url = request.url_for("sub:users:user_list", subdomain=...)

Route priority

Incoming paths are matched against each Route in order.

In cases where more that one route could match an incoming path, you should take care to ensure that more specific routes are listed before general cases.

For example:

# Don't do this: `/users/me` will never match incoming requests.
routes = [
    Route('/users/{username}', user),
    Route('/users/me', current_user),
]

# Do this: `/users/me` is tested first.
routes = [
    Route('/users/me', current_user),
    Route('/users/{username}', user),
]

Working with Router instances

If you're working at a low-level you might want to use a plain Router instance, rather that creating a Starlette application. This gives you a lightweight ASGI application that just provides the application routing, without wrapping it up in any middleware.

app = Router(routes=[
    Route('/', homepage),
    Mount('/users', routes=[
        Route('/', users, methods=['GET', 'POST']),
        Route('/{username}', user),
    ])
])

WebSocket Routing

When working with WebSocket endpoints, you should use WebSocketRoute instead of the usual Route.

Path parameters, and reverse URL lookups for WebSocketRoute work the the same as HTTP Route, which can be found in the HTTP Route section above.

from starlette.applications import Starlette
from starlette.routing import WebSocketRoute


async def websocket_index(websocket):
    await websocket.accept()
    await websocket.send_text("Hello, websocket!")
    await websocket.close()


async def websocket_user(websocket):
    name = websocket.path_params["name"]
    await websocket.accept()
    await websocket.send_text(f"Hello, {name}")
    await websocket.close()


routes = [
    WebSocketRoute("/", endpoint=websocket_index),
    WebSocketRoute("/{name}", endpoint=websocket_user),
]

app = Starlette(routes=routes)

The endpoint argument can be one of:

  • An async function, which accepts a single websocket argument.
  • A class that implements the ASGI interface, such as Starlette's WebSocketEndpoint.

endpoints.md

Starlette includes the classes HTTPEndpoint and WebSocketEndpoint that provide a class-based view pattern for handling HTTP method dispatching and WebSocket sessions.

HTTPEndpoint

The HTTPEndpoint class can be used as an ASGI application:

from starlette.responses import PlainTextResponse
from starlette.endpoints import HTTPEndpoint


class App(HTTPEndpoint):
    async def get(self, request):
        return PlainTextResponse(f"Hello, world!")

If you're using a Starlette application instance to handle routing, you can dispatch to an HTTPEndpoint class. Make sure to dispatch to the class itself, rather than to an instance of the class:

from starlette.applications import Starlette
from starlette.responses import PlainTextResponse
from starlette.endpoints import HTTPEndpoint
from starlette.routing import Route


class Homepage(HTTPEndpoint):
    async def get(self, request):
        return PlainTextResponse(f"Hello, world!")


class User(HTTPEndpoint):
    async def get(self, request):
        username = request.path_params['username']
        return PlainTextResponse(f"Hello, {username}")

routes = [
    Route("/", Homepage),
    Route("/{username}", User)
]

app = Starlette(routes=routes)

HTTP endpoint classes will respond with "405 Method not allowed" responses for any request methods which do not map to a corresponding handler.

WebSocketEndpoint

The WebSocketEndpoint class is an ASGI application that presents a wrapper around the functionality of a WebSocket instance.

The ASGI connection scope is accessible on the endpoint instance via .scope and has an attribute encoding which may optionally be set, in order to validate the expected websocket data in the on_receive method.

The encoding types are:

  • 'json'
  • 'bytes'
  • 'text'

There are three overridable methods for handling specific ASGI websocket message types:

  • async def on_connect(websocket, **kwargs)
  • async def on_receive(websocket, data)
  • async def on_disconnect(websocket, close_code)
from starlette.endpoints import WebSocketEndpoint


class App(WebSocketEndpoint):
    encoding = 'bytes'

    async def on_connect(self, websocket):
        await websocket.accept()

    async def on_receive(self, websocket, data):
        await websocket.send_bytes(b"Message: " + data)

    async def on_disconnect(self, websocket, close_code):
        pass

The WebSocketEndpoint can also be used with the Starlette application class:

import uvicorn
from starlette.applications import Starlette
from starlette.endpoints import WebSocketEndpoint, HTTPEndpoint
from starlette.responses import HTMLResponse
from starlette.routing import Route, WebSocketRoute


html = """
<!DOCTYPE html>
<html>
    <head>
        <title>Chat</title>
    </head>
    <body>
        <h1>WebSocket Chat</h1>
        <form action="" onsubmit="sendMessage(event)">
            <input type="text" id="messageText" autocomplete="off"/>
            <button>Send</button>
        </form>
        <ul id='messages'>
        </ul>
        <script>
            var ws = new WebSocket("ws://localhost:8000/ws");
            ws.onmessage = function(event) {
                var messages = document.getElementById('messages')
                var message = document.createElement('li')
                var content = document.createTextNode(event.data)
                message.appendChild(content)
                messages.appendChild(message)
            };
            function sendMessage(event) {
                var input = document.getElementById("messageText")
                ws.send(input.value)
                input.value = ''
                event.preventDefault()
            }
        </script>
    </body>
</html>
"""

class Homepage(HTTPEndpoint):
    async def get(self, request):
        return HTMLResponse(html)

class Echo(WebSocketEndpoint):
    encoding = "text"

    async def on_receive(self, websocket, data):
        await websocket.send_text(f"Message text was: {data}")

routes = [
    Route("/", Homepage),
    WebSocketRoute("/ws", Echo)
]

app = Starlette(routes=routes)

middleware.md

Starlette includes several middleware classes for adding behavior that is applied across your entire application. These are all implemented as standard ASGI middleware classes, and can be applied either to Starlette or to any other ASGI application.

Using middleware

The Starlette application class allows you to include the ASGI middleware in a way that ensures that it remains wrapped by the exception handler.

from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.httpsredirect import HTTPSRedirectMiddleware
from starlette.middleware.trustedhost import TrustedHostMiddleware

routes = ...

# Ensure that all requests include an 'example.com' or
# '*.example.com' host header, and strictly enforce https-only access.
middleware = [
    Middleware(
        TrustedHostMiddleware,
        allowed_hosts=['example.com', '*.example.com'],
    ),
    Middleware(HTTPSRedirectMiddleware)
]

app = Starlette(routes=routes, middleware=middleware)

Every Starlette application automatically includes two pieces of middleware by default:

  • ServerErrorMiddleware - Ensures that application exceptions may return a custom 500 page, or display an application traceback in DEBUG mode. This is always the outermost middleware layer.
  • ExceptionMiddleware - Adds exception handlers, so that particular types of expected exception cases can be associated with handler functions. For example raising HTTPException(status_code=404) within an endpoint will end up rendering a custom 404 page.

Middleware is evaluated from top-to-bottom, so the flow of execution in our example application would look like this:

  • Middleware
    • ServerErrorMiddleware
    • TrustedHostMiddleware
    • HTTPSRedirectMiddleware
    • ExceptionMiddleware
  • Routing
  • Endpoint

The following middleware implementations are available in the Starlette package:

CORSMiddleware

Adds appropriate CORS headers to outgoing responses in order to allow cross-origin requests from browsers.

The default parameters used by the CORSMiddleware implementation are restrictive by default, so you'll need to explicitly enable particular origins, methods, or headers, in order for browsers to be permitted to use them in a Cross-Domain context.

from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware

routes = ...

middleware = [
    Middleware(CORSMiddleware, allow_origins=['*'])
]

app = Starlette(routes=routes, middleware=middleware)

The following arguments are supported:

  • allow_origins - A list of origins that should be permitted to make cross-origin requests. eg. ['https://example.org', 'https://www.example.org']. You can use ['*'] to allow any origin.
  • allow_origin_regex - A regex string to match against origins that should be permitted to make cross-origin requests. eg. 'https://.*\.example\.org'.
  • allow_methods - A list of HTTP methods that should be allowed for cross-origin requests. Defaults to ['GET']. You can use ['*'] to allow all standard methods.
  • allow_headers - A list of HTTP request headers that should be supported for cross-origin requests. Defaults to []. You can use ['*'] to allow all headers. The Accept, Accept-Language, Content-Language and Content-Type headers are always allowed for CORS requests.
  • allow_credentials - Indicate that cookies should be supported for cross-origin requests. Defaults to False. Also, allow_origins, allow_methods and allow_headers cannot be set to ['*'] for credentials to be allowed, all of them must be explicitly specified.
  • expose_headers - Indicate any response headers that should be made accessible to the browser. Defaults to [].
  • max_age - Sets a maximum time in seconds for browsers to cache CORS responses. Defaults to 600.

The middleware responds to two particular types of HTTP request...

CORS preflight requests

These are any OPTIONS request with Origin and Access-Control-Request-Method headers. In this case the middleware will intercept the incoming request and respond with appropriate CORS headers, and either a 200 or 400 response for informational purposes.

Simple requests

Any request with an Origin header. In this case the middleware will pass the request through as normal, but will include appropriate CORS headers on the response.

SessionMiddleware

Adds signed cookie-based HTTP sessions. Session information is readable but not modifiable.

Access or modify the session data using the request.session dictionary interface.

The following arguments are supported:

  • secret_key - Should be a random string.
  • session_cookie - Defaults to "session".
  • max_age - Session expiry time in seconds. Defaults to 2 weeks. If set to None then the cookie will last as long as the browser session.
  • same_site - SameSite flag prevents the browser from sending session cookie along with cross-site requests. Defaults to 'lax'.
  • path - The path set for the session cookie. Defaults to '/'.
  • https_only - Indicate that Secure flag should be set (can be used with HTTPS only). Defaults to False.
  • domain - Domain of the cookie used to share cookie between subdomains or cross-domains. The browser defaults the domain to the same host that set the cookie, excluding subdomains (reference).
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.sessions import SessionMiddleware

routes = ...

middleware = [
    Middleware(SessionMiddleware, secret_key=..., https_only=True)
]

app = Starlette(routes=routes, middleware=middleware)

HTTPSRedirectMiddleware

Enforces that all incoming requests must either be https or wss. Any incoming requests to http or ws will be redirected to the secure scheme instead.

from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.httpsredirect import HTTPSRedirectMiddleware

routes = ...

middleware = [
    Middleware(HTTPSRedirectMiddleware)
]

app = Starlette(routes=routes, middleware=middleware)

There are no configuration options for this middleware class.

TrustedHostMiddleware

Enforces that all incoming requests have a correctly set Host header, in order to guard against HTTP Host Header attacks.

from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.trustedhost import TrustedHostMiddleware

routes = ...

middleware = [
    Middleware(TrustedHostMiddleware, allowed_hosts=['example.com', '*.example.com'])
]

app = Starlette(routes=routes, middleware=middleware)

The following arguments are supported:

  • allowed_hosts - A list of domain names that should be allowed as hostnames. Wildcard domains such as *.example.com are supported for matching subdomains. To allow any hostname either use allowed_hosts=["*"] or omit the middleware.
  • www_redirect - If set to True, requests to non-www versions of the allowed hosts will be redirected to their www counterparts. Defaults to True.

If an incoming request does not validate correctly then a 400 response will be sent.

GZipMiddleware

Handles GZip responses for any request that includes "gzip" in the Accept-Encoding header.

The middleware will handle both standard and streaming responses.

from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.gzip import GZipMiddleware


routes = ...

middleware = [
    Middleware(GZipMiddleware, minimum_size=1000, compresslevel=9)
]

app = Starlette(routes=routes, middleware=middleware)

The following arguments are supported:

  • minimum_size - Do not GZip responses that are smaller than this minimum size in bytes. Defaults to 500.
  • compresslevel - Used during GZip compression. It is an integer ranging from 1 to 9. Defaults to 9. Lower value results in faster compression but larger file sizes, while higher value results in slower compression but smaller file sizes.

The middleware won't GZip responses that already have a Content-Encoding set, to prevent them from being encoded twice.

BaseHTTPMiddleware

An abstract class that allows you to write ASGI middleware against a request/response interface.

Usage

To implement a middleware class using BaseHTTPMiddleware, you must override the async def dispatch(request, call_next) method.

from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware


class CustomHeaderMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        response = await call_next(request)
        response.headers['Custom'] = 'Example'
        return response

routes = ...

middleware = [
    Middleware(CustomHeaderMiddleware)
]

app = Starlette(routes=routes, middleware=middleware)

If you want to provide configuration options to the middleware class you should override the __init__ method, ensuring that the first argument is app, and any remaining arguments are optional keyword arguments. Make sure to set the app attribute on the instance if you do this.

class CustomHeaderMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, header_value='Example'):
        super().__init__(app)
        self.header_value = header_value

    async def dispatch(self, request, call_next):
        response = await call_next(request)
        response.headers['Custom'] = self.header_value
        return response


middleware = [
    Middleware(CustomHeaderMiddleware, header_value='Customized')
]

app = Starlette(routes=routes, middleware=middleware)

Middleware classes should not modify their state outside of the __init__ method. Instead you should keep any state local to the dispatch method, or pass it around explicitly, rather than mutating the middleware instance.

Limitations

Currently, the BaseHTTPMiddleware has some known limitations:

  • Using BaseHTTPMiddleware will prevent changes to contextlib.ContextVars from propagating upwards. That is, if you set a value for a ContextVar in your endpoint and try to read it from a middleware you will find that the value is not the same value you set in your endpoint (see this test for an example of this behavior).

To overcome these limitations, use pure ASGI middleware, as shown below.

staticfiles.md

Starlette also includes a StaticFiles class for serving files in a given directory:

StaticFiles

Signature: StaticFiles(directory=None, packages=None, html=False, check_dir=True, follow_symlink=False)

  • directory - A string or os.PathLike denoting a directory path.
  • packages - A list of strings or list of tuples of strings of python packages.
  • html - Run in HTML mode. Automatically loads index.html for directories if such file exist.
  • check_dir - Ensure that the directory exists upon instantiation. Defaults to True.
  • follow_symlink - A boolean indicating if symbolic links for files and directories should be followed. Defaults to False.

You can combine this ASGI application with Starlette's routing to provide comprehensive static file serving.

from starlette.applications import Starlette
from starlette.routing import Mount
from starlette.staticfiles import StaticFiles


routes = [
    ...
    Mount('/static', app=StaticFiles(directory='static'), name="static"),
]

app = Starlette(routes=routes)

Static files will respond with "404 Not found" or "405 Method not allowed" responses for requests which do not match. In HTML mode if 404.html file exists it will be shown as 404 response.

The packages option can be used to include "static" directories contained within a python package. The Python "bootstrap4" package is an example of this.

from starlette.applications import Starlette
from starlette.routing import Mount
from starlette.staticfiles import StaticFiles


routes=[
    ...
    Mount('/static', app=StaticFiles(directory='static', packages=['bootstrap4']), name="static"),
]

app = Starlette(routes=routes)

By default StaticFiles will look for statics directory in each package, you can change the default directory by specifying a tuple of strings.

routes=[
    ...
    Mount('/static', app=StaticFiles(packages=[('bootstrap4', 'static')]), name="static"),
]

You may prefer to include static files directly inside the "static" directory rather than using Python packaging to include static files, but it can be useful for bundling up reusable components.

templates.md

Starlette is not strictly coupled to any particular templating engine, but Jinja2 provides an excellent choice.

Jinja2Templates

Signature: Jinja2Templates(directory, context_processors=None, **env_options)

  • directory - A string, os.Pathlike or a list of strings or os.Pathlike denoting a directory path.
  • context_processors - A list of functions that return a dictionary to add to the template context.
  • **env_options - Additional keyword arguments to pass to the Jinja2 environment.

Starlette provides a simple way to get jinja2 configured. This is probably what you want to use by default.

from starlette.applications import Starlette
from starlette.routing import Route, Mount
from starlette.templating import Jinja2Templates
from starlette.staticfiles import StaticFiles


templates = Jinja2Templates(directory='templates')

async def homepage(request):
    return templates.TemplateResponse(request, 'index.html')

routes = [
    Route('/', endpoint=homepage),
    Mount('/static', StaticFiles(directory='static'), name='static')
]

app = Starlette(debug=True, routes=routes)

Note that the incoming request instance must be included as part of the template context.

The Jinja2 template context will automatically include a url_for function, so we can correctly hyperlink to other pages within the application.

For example, we can link to static files from within our HTML templates:

<link href="{{ url_for('static', path='/css/bootstrap.min.css') }}" rel="stylesheet" />

If you want to use custom filters, you will need to update the env property of Jinja2Templates:

from commonmark import commonmark
from starlette.templating import Jinja2Templates

def marked_filter(text):
    return commonmark(text)

templates = Jinja2Templates(directory='templates')
templates.env.filters['marked'] = marked_filter

Using custom jinja2.Environment instance

Starlette also accepts a preconfigured jinja2.Environment instance.

import jinja2
from starlette.templating import Jinja2Templates

env = jinja2.Environment(...)
templates = Jinja2Templates(env=env)

Context processors

A context processor is a function that returns a dictionary to be merged into a template context. Every function takes only one argument request and must return a dictionary to add to the context.

A common use case of template processors is to extend the template context with shared variables.

import typing
from starlette.requests import Request

def app_context(request: Request) -> typing.Dict[str, typing.Any]:
    return {'app': request.app}

Registering context templates

Pass context processors to context_processors argument of the Jinja2Templates class.

import typing

from starlette.requests import Request
from starlette.templating import Jinja2Templates

def app_context(request: Request) -> typing.Dict[str, typing.Any]:
    return {'app': request.app}

templates = Jinja2Templates(
    directory='templates', context_processors=[app_context]
)

!!! info Asynchronous functions as context processors are not supported.

Testing template responses

When using the test client, template responses include .template and .context attributes.

from starlette.testclient import TestClient


def test_homepage():
    client = TestClient(app)
    response = client.get("/")
    assert response.status_code == 200
    assert response.template.name == 'index.html'
    assert "request" in response.context

Customizing Jinja2 Environment

Jinja2Templates accepts all options supported by Jinja2 Environment. This will allow more control over the Environment instance created by Starlette.

For the list of options available to Environment you can check Jinja2 documentation here

from starlette.templating import Jinja2Templates


templates = Jinja2Templates(directory='templates', autoescape=False, auto_reload=True)

Asynchronous template rendering

Jinja2 supports async template rendering, however as a general rule we'd recommend that you keep your templates free from logic that invokes database lookups, or other I/O operations.

Instead we'd recommend that you ensure that your endpoints perform all I/O, for example, strictly evaluate any database queries within the view and include the final results in the context.

graphql.md

GraphQL support in Starlette was deprecated in version 0.15.0, and removed in version 0.17.0.

Although GraphQL support is no longer built in to Starlette, you can still use GraphQL with Starlette via 3rd party libraries. These libraries all have Starlette-specific guides to help you do just that:

authentication.md

Starlette offers a simple but powerful interface for handling authentication and permissions. Once you've installed AuthenticationMiddleware with an appropriate authentication backend the request.user and request.auth interfaces will be available in your endpoints.

from starlette.applications import Starlette
from starlette.authentication import (
    AuthCredentials, AuthenticationBackend, AuthenticationError, SimpleUser
)
from starlette.middleware import Middleware
from starlette.middleware.authentication import AuthenticationMiddleware
from starlette.responses import PlainTextResponse
from starlette.routing import Route
import base64
import binascii


class BasicAuthBackend(AuthenticationBackend):
    async def authenticate(self, conn):
        if "Authorization" not in conn.headers:
            return

        auth = conn.headers["Authorization"]
        try:
            scheme, credentials = auth.split()
            if scheme.lower() != 'basic':
                return
            decoded = base64.b64decode(credentials).decode("ascii")
        except (ValueError, UnicodeDecodeError, binascii.Error) as exc:
            raise AuthenticationError('Invalid basic auth credentials')

        username, _, password = decoded.partition(":")
        # TODO: You'd want to verify the username and password here.
        return AuthCredentials(["authenticated"]), SimpleUser(username)


async def homepage(request):
    if request.user.is_authenticated:
        return PlainTextResponse('Hello, ' + request.user.display_name)
    return PlainTextResponse('Hello, you')

routes = [
    Route("/", endpoint=homepage)
]

middleware = [
    Middleware(AuthenticationMiddleware, backend=BasicAuthBackend())
]

app = Starlette(routes=routes, middleware=middleware)

Users

Once AuthenticationMiddleware is installed the request.user interface will be available to endpoints or other middleware.

This interface should subclass BaseUser, which provides two properties, as well as whatever other information your user model includes.

  • .is_authenticated
  • .display_name

Starlette provides two built-in user implementations: UnauthenticatedUser(), and SimpleUser(username).

AuthCredentials

It is important that authentication credentials are treated as separate concept from users. An authentication scheme should be able to restrict or grant particular privileges independently of the user identity.

The AuthCredentials class provides the basic interface that request.auth exposes:

  • .scopes

Permissions

Permissions are implemented as an endpoint decorator, that enforces that the incoming request includes the required authentication scopes.

from starlette.authentication import requires


@requires('authenticated')
async def dashboard(request):
    ...

You can include either one or multiple required scopes:

from starlette.authentication import requires


@requires(['authenticated', 'admin'])
async def dashboard(request):
    ...

By default 403 responses will be returned when permissions are not granted. In some cases you might want to customize this, for example to hide information about the URL layout from unauthenticated users.

from starlette.authentication import requires


@requires(['authenticated', 'admin'], status_code=404)
async def dashboard(request):
    ...

!!! note The status_code parameter is not supported with WebSockets. The 403 (Forbidden) status code will always be used for those.

Alternatively you might want to redirect unauthenticated users to a different page.

from starlette.authentication import requires


async def homepage(request):
    ...


@requires('authenticated', redirect='homepage')
async def dashboard(request):
    ...

When redirecting users, the page you redirect them to will include URL they originally requested at the next query param:

from starlette.authentication import requires
from starlette.responses import RedirectResponse


@requires('authenticated', redirect='login')
async def admin(request):
    ...


async def login(request):
    if request.method == "POST":
        # Now that the user is authenticated,
        # we can send them to their original request destination
        if request.user.is_authenticated:
            next_url = request.query_params.get("next")
            if next_url:
                return RedirectResponse(next_url)
            return RedirectResponse("/")

For class-based endpoints, you should wrap the decorator around a method on the class.

from starlette.authentication import requires
from starlette.endpoints import HTTPEndpoint


class Dashboard(HTTPEndpoint):
    @requires("authenticated")
    async def get(self, request):
        ...

Custom authentication error responses

You can customise the error response sent when a AuthenticationError is raised by an auth backend:

from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.authentication import AuthenticationMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse


def on_auth_error(request: Request, exc: Exception):
    return JSONResponse({"error": str(exc)}, status_code=401)

app = Starlette(
    middleware=[
        Middleware(AuthenticationMiddleware, backend=BasicAuthBackend(), on_error=on_auth_error),
    ],
)

schemas.md

Starlette supports generating API schemas, such as the widely used OpenAPI specification. (Formerly known as "Swagger".)

Schema generation works by inspecting the routes on the application through app.routes, and using the docstrings or other attributes on the endpoints in order to determine a complete API schema.

Starlette is not tied to any particular schema generation or validation tooling, but includes a simple implementation that generates OpenAPI schemas based on the docstrings.

from starlette.applications import Starlette
from starlette.routing import Route
from starlette.schemas import SchemaGenerator


schemas = SchemaGenerator(
    {"openapi": "3.0.0", "info": {"title": "Example API", "version": "1.0"}}
)

def list_users(request):
    """
    responses:
      200:
        description: A list of users.
        examples:
          [{"username": "tom"}, {"username": "lucy"}]
    """
    raise NotImplementedError()


def create_user(request):
    """
    responses:
      200:
        description: A user.
        examples:
          {"username": "tom"}
    """
    raise NotImplementedError()


def openapi_schema(request):
    return schemas.OpenAPIResponse(request=request)


routes = [
    Route("/users", endpoint=list_users, methods=["GET"]),
    Route("/users", endpoint=create_user, methods=["POST"]),
    Route("/schema", endpoint=openapi_schema, include_in_schema=False)
]

app = Starlette(routes=routes)

We can now access an OpenAPI schema at the "/schema" endpoint.

You can generate the API Schema directly with .get_schema(routes):

schema = schemas.get_schema(routes=app.routes)
assert schema == {
    "openapi": "3.0.0",
    "info": {"title": "Example API", "version": "1.0"},
    "paths": {
        "/users": {
            "get": {
                "responses": {
                    200: {
                        "description": "A list of users.",
                        "examples": [{"username": "tom"}, {"username": "lucy"}],
                    }
                }
            },
            "post": {
                "responses": {
                    200: {"description": "A user.", "examples": {"username": "tom"}}
                }
            },
        },
    },
}

You might also want to be able to print out the API schema, so that you can use tooling such as generating API documentation.

if __name__ == '__main__':
    assert sys.argv[-1] in ("run", "schema"), "Usage: example.py [run|schema]"

    if sys.argv[-1] == "run":
        uvicorn.run("example:app", host='0.0.0.0', port=8000)
    elif sys.argv[-1] == "schema":
        schema = schemas.get_schema(routes=app.routes)
        print(yaml.dump(schema, default_flow_style=False))

Third party packages

starlette-apispec

Easy APISpec integration for Starlette, which supports some object serialization libraries.

lifespan.md

Starlette applications can register a lifespan handler for dealing with code that needs to run before the application starts up, or when the application is shutting down.

import contextlib

from starlette.applications import Starlette


@contextlib.asynccontextmanager
async def lifespan(app):
    async with some_async_resource():
        print("Run at startup!")
        yield
        print("Run on shutdown!")


routes = [
    ...
]

app = Starlette(routes=routes, lifespan=lifespan)

Starlette will not start serving any incoming requests until the lifespan has been run.

The lifespan teardown will run once all connections have been closed, and any in-process background tasks have completed.

Consider using anyio.create_task_group() for managing asynchronous tasks.

Lifespan State

The lifespan has the concept of state, which is a dictionary that can be used to share the objects between the lifespan, and the requests.

import contextlib
from typing import AsyncIterator, TypedDict

import httpx
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import PlainTextResponse
from starlette.routing import Route


class State(TypedDict):
    http_client: httpx.AsyncClient


@contextlib.asynccontextmanager
async def lifespan(app: Starlette) -> AsyncIterator[State]:
    async with httpx.AsyncClient() as client:
        yield {"http_client": client}


async def homepage(request: Request) -> PlainTextResponse:
    client = request.state.http_client
    response = await client.get("https://www.example.com")
    return PlainTextResponse(response.text)


app = Starlette(
    lifespan=lifespan,
    routes=[Route("/", homepage)]
)

The state received on the requests is a shallow copy of the state received on the lifespan handler.

Running lifespan in tests

You should use TestClient as a context manager, to ensure that the lifespan is called.

from example import app
from starlette.testclient import TestClient


def test_homepage():
    with TestClient(app) as client:
        # Application's lifespan is called on entering the block.
        response = client.get("/")
        assert response.status_code == 200

    # And the lifespan's teardown is run when exiting the block.

background.md

Starlette includes a BackgroundTask class for in-process background tasks.

A background task should be attached to a response, and will run only once the response has been sent.

Background Task

Used to add a single background task to a response.

Signature: BackgroundTask(func, *args, **kwargs)

from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route
from starlette.background import BackgroundTask


...

async def signup(request):
    data = await request.json()
    username = data['username']
    email = data['email']
    task = BackgroundTask(send_welcome_email, to_address=email)
    message = {'status': 'Signup successful'}
    return JSONResponse(message, background=task)

async def send_welcome_email(to_address):
    ...


routes = [
    ...
    Route('/user/signup', endpoint=signup, methods=['POST'])
]

app = Starlette(routes=routes)

BackgroundTasks

Used to add multiple background tasks to a response.

Signature: BackgroundTasks(tasks=[])

from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.background import BackgroundTasks

async def signup(request):
    data = await request.json()
    username = data['username']
    email = data['email']
    tasks = BackgroundTasks()
    tasks.add_task(send_welcome_email, to_address=email)
    tasks.add_task(send_admin_notification, username=username)
    message = {'status': 'Signup successful'}
    return JSONResponse(message, background=tasks)

async def send_welcome_email(to_address):
    ...

async def send_admin_notification(username):
    ...

routes = [
    Route('/user/signup', endpoint=signup, methods=['POST'])
]

app = Starlette(routes=routes)

!!! important The tasks are executed in order. In case one of the tasks raises an exception, the following tasks will not get the opportunity to be executed.

server-push.md

Starlette includes support for HTTP/2 and HTTP/3 server push, making it possible to push resources to the client to speed up page load times.

Request.send_push_promise

Used to initiate a server push for a resource. If server push is not available this method does nothing.

Signature: send_push_promise(path)

  • path - A string denoting the path of the resource.
from starlette.applications import Starlette
from starlette.responses import HTMLResponse
from starlette.routing import Route, Mount
from starlette.staticfiles import StaticFiles


async def homepage(request):
    """
    Homepage which uses server push to deliver the stylesheet.
    """
    await request.send_push_promise("/static/style.css")
    return HTMLResponse(
        '<html><head><link rel="stylesheet" href="/static/style.css"/></head></html>'
    )

routes = [
    Route("/", endpoint=homepage),
    Mount("/static", StaticFiles(directory="static"), name="static")
]

app = Starlette(routes=routes)

exceptions.md

Starlette allows you to install custom exception handlers to deal with how you return responses when errors or handled exceptions occur.

from starlette.applications import Starlette
from starlette.exceptions import HTTPException
from starlette.requests import Request
from starlette.responses import HTMLResponse


HTML_404_PAGE = ...
HTML_500_PAGE = ...


async def not_found(request: Request, exc: HTTPException):
    return HTMLResponse(content=HTML_404_PAGE, status_code=exc.status_code)

async def server_error(request: Request, exc: HTTPException):
    return HTMLResponse(content=HTML_500_PAGE, status_code=exc.status_code)


exception_handlers = {
    404: not_found,
    500: server_error
}

app = Starlette(routes=routes, exception_handlers=exception_handlers)

If debug is enabled and an error occurs, then instead of using the installed 500 handler, Starlette will respond with a traceback response.

app = Starlette(debug=True, routes=routes, exception_handlers=exception_handlers)

As well as registering handlers for specific status codes, you can also register handlers for classes of exceptions.

In particular you might want to override how the built-in HTTPException class is handled. For example, to use JSON style responses:

async def http_exception(request: Request, exc: HTTPException):
    return JSONResponse({"detail": exc.detail}, status_code=exc.status_code)

exception_handlers = {
    HTTPException: http_exception
}

The HTTPException is also equipped with the headers argument. Which allows the propagation of the headers to the response class:

async def http_exception(request: Request, exc: HTTPException):
    return JSONResponse(
        {"detail": exc.detail},
        status_code=exc.status_code,
        headers=exc.headers
    )

You might also want to override how WebSocketException is handled:

async def websocket_exception(websocket: WebSocket, exc: WebSocketException):
    await websocket.close(code=1008)

exception_handlers = {
    WebSocketException: websocket_exception
}

Errors and handled exceptions

It is important to differentiate between handled exceptions and errors.

Handled exceptions do not represent error cases. They are coerced into appropriate HTTP responses, which are then sent through the standard middleware stack. By default the HTTPException class is used to manage any handled exceptions.

Errors are any other exception that occurs within the application. These cases should bubble through the entire middleware stack as exceptions. Any error logging middleware should ensure that it re-raises the exception all the way up to the server.

In practical terms, the error handled used is exception_handler[500] or exception_handler[Exception]. Both keys 500 and Exception can be used. See below:

async def handle_error(request: Request, exc: HTTPException):
    # Perform some logic
    return JSONResponse({"detail": exc.detail}, status_code=exc.status_code)

exception_handlers = {
    Exception: handle_error  # or "500: handle_error"
}

It's important to notice that in case a BackgroundTask raises an exception, it will be handled by the handle_error function, but at that point, the response was already sent. In other words, the response created by handle_error will be discarded. In case the error happens before the response was sent, then it will use the response object - in the above example, the returned JSONResponse.

In order to deal with this behaviour correctly, the middleware stack of a Starlette application is configured like this:

  • ServerErrorMiddleware - Returns 500 responses when server errors occur.
  • Installed middleware
  • ExceptionMiddleware - Deals with handled exceptions, and returns responses.
  • Router
  • Endpoints

HTTPException

The HTTPException class provides a base class that you can use for any handled exceptions. The ExceptionMiddleware implementation defaults to returning plain-text HTTP responses for any HTTPException.

  • HTTPException(status_code, detail=None, headers=None)

You should only raise HTTPException inside routing or endpoints. Middleware classes should instead just return appropriate responses directly.

WebSocketException

You can use the WebSocketException class to raise errors inside of WebSocket endpoints.

  • WebSocketException(code=1008, reason=None)

You can set any code valid as defined in the specification.

config.md

Starlette encourages a strict separation of configuration from code, following the twelve-factor pattern.

Configuration should be stored in environment variables, or in a .env file that is not committed to source control.

from sqlalchemy import create_engine
from starlette.applications import Starlette
from starlette.config import Config
from starlette.datastructures import CommaSeparatedStrings, Secret

# Config will be read from environment variables and/or ".env" files.
config = Config(".env")

DEBUG = config('DEBUG', cast=bool, default=False)
DATABASE_URL = config('DATABASE_URL')
SECRET_KEY = config('SECRET_KEY', cast=Secret)
ALLOWED_HOSTS = config('ALLOWED_HOSTS', cast=CommaSeparatedStrings)

app = Starlette(debug=DEBUG)
engine = create_engine(DATABASE_URL)
...
# Don't commit this to source control.
# Eg. Include ".env" in your `.gitignore` file.
DEBUG=True
DATABASE_URL=postgresql://user:password@localhost:5432/database
SECRET_KEY=43n080musdfjt54t-09sdgr
ALLOWED_HOSTS=127.0.0.1, localhost

Configuration precedence

The order in which configuration values are read is:

  • From an environment variable.
  • From the .env file.
  • The default value given in config.

If none of those match, then config(...) will raise an error.

Secrets

For sensitive keys, the Secret class is useful, since it helps minimize occasions where the value it holds could leak out into tracebacks or other code introspection.

To get the value of a Secret instance, you must explicitly cast it to a string. You should only do this at the point at which the value is used.

>>> from myproject import settings
>>> settings.SECRET_KEY
Secret('**********')
>>> str(settings.SECRET_KEY)
'98n349$%8b8-7yjn0n8y93T$23r'

!!! tip

You can use `DatabaseURL` from `databases`
package [here](https://github.com/encode/databases/blob/ab5eb718a78a27afe18775754e9c0fa2ad9cd211/databases/core.py#L420)
to store database URLs and avoid leaking them in the logs.

CommaSeparatedStrings

For holding multiple inside a single config key, the CommaSeparatedStrings type is useful.

>>> from myproject import settings
>>> print(settings.ALLOWED_HOSTS)
CommaSeparatedStrings(['127.0.0.1', 'localhost'])
>>> print(list(settings.ALLOWED_HOSTS))
['127.0.0.1', 'localhost']
>>> print(len(settings.ALLOWED_HOSTS))
2
>>> print(settings.ALLOWED_HOSTS[0])
'127.0.0.1'

Reading or modifying the environment

In some cases you might want to read or modify the environment variables programmatically. This is particularly useful in testing, where you may want to override particular keys in the environment.

Rather than reading or writing from os.environ, you should use Starlette's environ instance. This instance is a mapping onto the standard os.environ that additionally protects you by raising an error if any environment variable is set after the point that it has already been read by the configuration.

If you're using pytest, then you can setup any initial environment in tests/conftest.py.

from starlette.config import environ

environ['DEBUG'] = 'TRUE'

Reading prefixed environment variables

You can namespace the environment variables by setting env_prefix argument.

import os

from starlette.config import Config

os.environ['APP_DEBUG'] = 'yes'
os.environ['ENVIRONMENT'] = 'dev'

config = Config(env_prefix='APP_')

DEBUG = config('DEBUG') # lookups APP_DEBUG, returns "yes"
ENVIRONMENT = config('ENVIRONMENT') # lookups APP_ENVIRONMENT, raises KeyError as variable is not defined

A full example

Structuring large applications can be complex. You need proper separation of configuration and code, database isolation during tests, separate test and production databases, etc...

Here we'll take a look at a complete example, that demonstrates how we can start to structure an application.

First, let's keep our settings, our database table definitions, and our application logic separated:

from starlette.config import Config
from starlette.datastructures import Secret

config = Config(".env")

DEBUG = config('DEBUG', cast=bool, default=False)
SECRET_KEY = config('SECRET_KEY', cast=Secret)

DATABASE_URL = config('DATABASE_URL')
import sqlalchemy

# Database table definitions.
metadata = sqlalchemy.MetaData()

organisations = sqlalchemy.Table(
    ...
)
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.sessions import SessionMiddleware
from starlette.routing import Route

from myproject import settings


async def homepage(request):
    ...

routes = [
    Route("/", endpoint=homepage)
]

middleware = [
    Middleware(
        SessionMiddleware,
        secret_key=settings.SECRET_KEY,
    )
]

app = Starlette(debug=settings.DEBUG, routes=routes, middleware=middleware)

Now let's deal with our test configuration. We'd like to create a new test database every time the test suite runs, and drop it once the tests complete. We'd also like to ensure

from starlette.config import environ
from starlette.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy_utils import create_database, database_exists, drop_database

# This line would raise an error if we use it after 'settings' has been imported.
environ['DEBUG'] = 'TRUE'

from myproject import settings
from myproject.app import app
from myproject.tables import metadata


@pytest.fixture(autouse=True, scope="session")
def setup_test_database():
    """
    Create a clean test database every time the tests are run.
    """
    url = settings.DATABASE_URL
    engine = create_engine(url)
    assert not database_exists(url), 'Test database already exists. Aborting tests.'
    create_database(url)             # Create the test database.
    metadata.create_all(engine)      # Create the tables.
    yield                            # Run the tests.
    drop_database(url)               # Drop the test database.


@pytest.fixture()
def client():
    """
    Make a 'client' fixture available to test cases.
    """
    # Our fixture is created within a context manager. This ensures that
    # application lifespan runs for every test case.
    with TestClient(app) as test_client:
        yield test_client

testclient.md

The test client allows you to make requests against your ASGI application, using the httpx library.

from starlette.responses import HTMLResponse
from starlette.testclient import TestClient


async def app(scope, receive, send):
    assert scope['type'] == 'http'
    response = HTMLResponse('<html><body>Hello, world!</body></html>')
    await response(scope, receive, send)


def test_app():
    client = TestClient(app)
    response = client.get('/')
    assert response.status_code == 200

The test client exposes the same interface as any other httpx session. In particular, note that the calls to make a request are just standard function calls, not awaitables.

You can use any of httpx standard API, such as authentication, session cookies handling, or file uploads.

For example, to set headers on the TestClient you can do:

client = TestClient(app)

# Set headers on the client for future requests
client.headers = {"Authorization": "..."}
response = client.get("/")

# Set headers for each request separately
response = client.get("/", headers={"Authorization": "..."})

And for example to send files with the TestClient:

client = TestClient(app)

# Send a single file
with open("example.txt", "rb") as f:
    response = client.post("/form", files={"file": f})

# Send multiple files
with open("example.txt", "rb") as f1:
    with open("example.png", "rb") as f2:
        files = {"file1": f1, "file2": ("filename", f2, "image/png")}
        response = client.post("/form", files=files)

For more information you can check the httpx documentation.

By default the TestClient will raise any exceptions that occur in the application. Occasionally you might want to test the content of 500 error responses, rather than allowing client to raise the server exception. In this case you should use client = TestClient(app, raise_server_exceptions=False).

!!! note

If you want the `TestClient` to run the `lifespan` handler,
you will need to use the `TestClient` as a context manager. It will
not be triggered when the `TestClient` is instantiated. You can learn more about it
[here](lifespan.md#running-lifespan-in-tests).

Selecting the Async backend

TestClient takes arguments backend (a string) and backend_options (a dictionary). These options are passed to anyio.start_blocking_portal(). See the anyio documentation for more information about the accepted backend options. By default, asyncio is used with default options.

To run Trio, pass backend="trio". For example:

def test_app()
    with TestClient(app, backend="trio") as client:
       ...

To run asyncio with uvloop, pass backend_options={"use_uvloop": True}. For example:

def test_app()
    with TestClient(app, backend_options={"use_uvloop": True}) as client:
       ...

Testing WebSocket sessions

You can also test websocket sessions with the test client.

The httpx library will be used to build the initial handshake, meaning you can use the same authentication options and other headers between both http and websocket testing.

from starlette.testclient import TestClient
from starlette.websockets import WebSocket


async def app(scope, receive, send):
    assert scope['type'] == 'websocket'
    websocket = WebSocket(scope, receive=receive, send=send)
    await websocket.accept()
    await websocket.send_text('Hello, world!')
    await websocket.close()


def test_app():
    client = TestClient(app)
    with client.websocket_connect('/') as websocket:
        data = websocket.receive_text()
        assert data == 'Hello, world!'

The operations on session are standard function calls, not awaitables.

It's important to use the session within a context-managed with block. This ensure that the background thread on which the ASGI application is properly terminated, and that any exceptions that occur within the application are always raised by the test client.

Establishing a test session

  • .websocket_connect(url, subprotocols=None, **options) - Takes the same set of arguments as httpx.get().

May raise starlette.websockets.WebSocketDisconnect if the application does not accept the websocket connection.

websocket_connect() must be used as a context manager (in a with block).

!!! note The params argument is not supported by websocket_connect. If you need to pass query arguments, hard code it directly in the URL.

```python
with client.websocket_connect('/path?foo=bar') as websocket:
    ...
```

Sending data

  • .send_text(data) - Send the given text to the application.
  • .send_bytes(data) - Send the given bytes to the application.
  • .send_json(data, mode="text") - Send the given data to the application. Use mode="binary" to send JSON over binary data frames.

Receiving data

  • .receive_text() - Wait for incoming text sent by the application and return it.
  • .receive_bytes() - Wait for incoming bytestring sent by the application and return it.
  • .receive_json(mode="text") - Wait for incoming json data sent by the application and return it. Use mode="binary" to receive JSON over binary data frames.

May raise starlette.websockets.WebSocketDisconnect.

Closing the connection

  • .close(code=1000) - Perform a client-side close of the websocket connection.

Asynchronous tests

Sometimes you will want to do async things outside of your application. For example, you might want to check the state of your database after calling your app using your existing async database client / infrastructure.

For these situations, using TestClient is difficult because it creates it's own event loop and async resources (like a database connection) often cannot be shared across event loops. The simplest way to work around this is to just make your entire test async and use an async client, like httpx.AsyncClient.

Here is an example of such a test:

from httpx import AsyncClient
from starlette.applications import Starlette
from starlette.routing import Route
from starlette.requests import Request
from starlette.responses import PlainTextResponse


def hello(request: Request) -> PlainTextResponse:
    return PlainTextResponse("Hello World!")


app = Starlette(routes=[Route("/", hello)])


# if you're using pytest, you'll need to to add an async marker like:
# @pytest.mark.anyio  # using https://github.com/agronholm/anyio
# or install and configure pytest-asyncio (https://github.com/pytest-dev/pytest-asyncio)
async def test_app() -> None:
    # note: you _must_ set `base_url` for relative urls like "/" to work
    async with AsyncClient(app=app, base_url="http://testserver") as client:
        r = await client.get("/")
        assert r.status_code == 200
        assert r.text == "Hello World!"
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment