The Starlite App#
Application object#
At the root of every Starlite application is an instance of the Starlite
class. Typically, this code will be placed in a file called main.py at the project’s root directory.
Creating an app is straightforward – the only required arg is a list
of Controllers, Routers
or Route handlers:
from typing import Dict
from starlite import Starlite, get
@get("/")
def hello_world() -> Dict[str, str]:
"""Handler function that returns a greeting dictionary."""
return {"hello": "world"}
app = Starlite(route_handlers=[hello_world])
# run: /
from starlite import Starlite, get
@get("/")
def hello_world() -> dict[str, str]:
"""Handler function that returns a greeting dictionary."""
return {"hello": "world"}
app = Starlite(route_handlers=[hello_world])
# run: /
The app instance is the root level of the app - it has the base path of / and all root level Controllers, Routers
and Route Handlers should be registered on it.
See also
To learn more about registering routes, check out this chapter in the documentation: Registering Routes
Startup and Shutdown#
You can pass a list of callables - either sync or async functions, methods or class instances - to the on_startup
/ on_shutdown kwargs of the Starlite instance. Those will be called in
order, once the ASGI server (uvicorn, daphne etc.) emits the respective event.
A classic use case for this is database connectivity. Often, we want to establish a database connection on application startup, and then close it gracefully upon shutdown.
For example, lets create a database connection using the async engine from SQLAlchemy. We create two functions, one to get or establish the connection, and another to close it, and then pass them to the Starlite constructor:
from typing import cast
from pydantic import BaseSettings
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
from starlite import Starlite, State
class AppSettings(BaseSettings):
DATABASE_URI: str = "postgresql+asyncpg://postgres:mysecretpassword@pg.db:5432/db"
settings = AppSettings()
def get_db_connection(state: State) -> AsyncEngine:
"""Returns the db engine.
If it doesn't exist, creates it and saves it in on the application state object
"""
if not getattr(state, "engine", None):
state.engine = create_async_engine(settings.DATABASE_URI)
return cast("AsyncEngine", state.engine)
async def close_db_connection(state: State) -> None:
"""Closes the db connection stored in the application State object."""
if getattr(state, "engine", None):
await cast("AsyncEngine", state.engine).dispose()
app = Starlite(route_handlers=[], on_startup=[get_db_connection], on_shutdown=[close_db_connection])
Using Application State#
As seen in the examples for the on_startup / on_shutdown,
callables passed to these hooks can receive an optional kwarg called state, which is the application’s state object.
The advantage of using application state, is that it can be accessed during multiple stages of the connection, and
it can be injected into dependencies and route handlers.
The Application State is an instance of the State datastructure, and it is accessible
via the
app.state attribute. As such it can be accessed wherever the app instance is accessible.
It’s important to understand in this context that the application instance is injected into the ASGI scope mapping for
each connection (i.e. request or websocket connection) as scope["app"].state. This makes the application accessible
wherever the scope mapping is available, e.g. in middleware, on Request and
Websocket instances (accessible as request.app / socket.app) and many other
places.
Therefore, state offers an easy way to share contextual data between disparate parts of the application, as seen below:
import logging
from typing import TYPE_CHECKING, Any
from starlite import Provide, Request, Starlite, State, get
if TYPE_CHECKING:
from starlite.types import ASGIApp, Receive, Scope, Send
logger = logging.getLogger(__name__)
def set_state_on_startup(state: State) -> None:
"""Startup and shutdown hooks can receive `State` as a keyword arg."""
state.value = "abc123"
def middleware_factory(*, app: "ASGIApp") -> "ASGIApp":
"""A middleware can access application state via `scope`."""
async def my_middleware(scope: "Scope", receive: "Receive", send: "Send") -> None:
state = scope["app"].state
logger.info("state value in middleware: %s", state.value)
await app(scope, receive, send)
return my_middleware
def my_dependency(state: State) -> Any:
"""Dependencies can receive state via injection."""
logger.info("state value in dependency: %s", state.value)
@get("/", dependencies={"dep": Provide(my_dependency)}, middleware=[middleware_factory])
def get_handler(state: State, request: Request, dep: Any) -> None: # pylint: disable=unused-argument
"""Handlers can receive state via injection."""
logger.info("state value in handler from `State`: %s", state.value)
logger.info("state value in handler from `Request`: %s", request.app.state.value)
app = Starlite(route_handlers=[get_handler], on_startup=[set_state_on_startup], debug=True)
Initializing Application State#
You can pass an object from which the application state will be instantiated using the initial_state kwarg of the
Starlite constructor:
from starlite import Starlite, State, get
@get("/")
def handler(state: State) -> dict:
return state.dict()
app = Starlite(route_handlers=[handler], initial_state={"count": 100})
Note
The initial_state can be a dictionary, an instance of ImmutableState
or State, or a list of tuples containing key/value pairs.
Attention
Any value passed to initial_state will be deep copied - to prevent mutation from outside the application context.
Injecting Application State into Route Handlers and Dependencies#
As seen in the above example, Starlite offers an easy way to inject state into route handlers and dependencies - simply
by specifying state as a kwarg to the handler function. I.e., you can simply do this in handler function or dependency
to access the application state:
from starlite import get, State
@get("/")
def handler(state: State) -> None:
...
When using this pattern you can specify the class to use for the state object. This type is not merely for type checkers, rather Starlite will instantiate a new state instance based on the type you set there. This allows users to use custom classes for State, e.g.:
While this is very powerful, it might encourage users to follow anti-patterns: it’s important to emphasize that using
state can lead to code that’s hard to reason about and bugs that are difficult to understand, due to changes in
different ASGI contexts. As such, this pattern should be used only when it is the best choice and in a limited fashion.
To discourage its use, Starlite also offers a builtin ImmutableState class. You can use this class to type state and
ensure that no mutation of state is allowed:
from starlite import ImmutableState, Starlite, get
@get("/")
def handler(state: ImmutableState) -> dict:
setattr(state, "count", 1) # raises AttributeError
return state.dict()
app = Starlite(route_handlers=[handler])
Static Files#
Static files are served by the app from predefined locations. To configure static file serving, either pass an
instance of StaticFilesConfig or a list
thereof to the Starlite constructor using the static_files_config kwarg.
For example, lets say our Starlite app is going to serve regular files from the my_app/static folder and html
documents from the my_app/html folder, and we would like to serve the static files on the /files path,
and the html files on the /html path:
from starlite import Starlite, StaticFilesConfig
app = Starlite(
route_handlers=[...],
static_files_config=[
StaticFilesConfig(directories=["static"], path="/files"),
StaticFilesConfig(directories=["html"], path="/html", html_mode=True),
],
)
Matching is done based on filename, for example, assume we have a request that is trying to retrieve the path
/files/file.txt, the directory for the base path /files will be searched for the file file.txt. If it is
found, the file will be sent, otherwise a 404 response will be sent.
If html_mode is enabled and no specific file is requested, the application will fall back to serving index.html. If
no file is found the application will look for a 404.html file in order to render a response, otherwise a 404
NotFoundException will be returned.
You can provide a name parameter to StaticFilesConfig to identify the given config and generate links to files in
folders belonging to that config. name should be a unique string across all static configs and
route handlers.
from starlite import Starlite, StaticFilesConfig
app = Starlite(
route_handlers=[...],
static_files_config=[
StaticFilesConfig(
directories=["static"], path="/some_folder/static/path", name="static"
),
],
)
url_path = app.url_for_static_asset("static", "file.pdf")
# /some_folder/static/path/file.pdf
Sending files as attachments#
By default, files are sent “inline”, meaning they will have a Content-Disposition: inline header.
To send them as attachments, use the send_as_attachment=True flag, which will add a
Content-Disposition: attachment header:
from starlite import Starlite, StaticFilesConfig
app = Starlite(
route_handlers=[...],
static_files_config=[
StaticFilesConfig(
directories=["static"],
path="/some_folder/static/path",
name="static",
send_as_attachment=True,
),
],
)
File System support and Cloud Files#
The StaticFilesConfig class accepts a value called file_system,
which can be any class adhering to the Starlite FileSystemProtocol.
This protocol is similar to the file systems defined by fsspec,
which cover all major cloud providers and a wide range of other use cases (e.g. HTTP based file service, ftp etc.).
In order to use any file system, simply use fsspec or one of
the libraries based upon it, or provide a custom implementation adhering to the
FileSystemProtocol.
Logging#
Starlite has builtin pydantic based logging configuration that allows users to easily define logging:
from starlite import Starlite, LoggingConfig, Request, get
@get("/")
def my_router_handler(request: Request) -> None:
request.logger.info("inside a request")
return None
logging_config = LoggingConfig(
loggers={
"my_app": {
"level": "INFO",
"handlers": ["queue_listener"],
}
}
)
app = Starlite(route_handlers=[my_router_handler], logging_config=logging_config)
Attention
Starlite configures a non-blocking QueueListenerHandler which is keyed as queue_listener in the logging configuration. The above example is using this handler, which is optimal for async applications. Make sure to use it in your own loggers as in the above example.
Using Picologging#
Picologging is a high performance logging library that is developed by
Microsoft. Starlite will default to using this library automatically if its installed - requiring zero configuration on
the part of the user. That is, if picologging is present the previous example will work with it automatically.
Using StructLog#
StructLog is a powerful structured-logging library. Starlite ships with a dedicated logging config for using it:
from starlite import Starlite, StructLoggingConfig, Request, get
@get("/")
def my_router_handler(request: Request) -> None:
request.logger.info("inside a request")
return None
logging_config = StructLoggingConfig()
app = Starlite(route_handlers=[my_router_handler], logging_config=logging_config)
Subclass Logging Configs#
You can easily create you own LoggingConfig class by subclassing
BaseLoggingConfig and
implementing the configure method.
Application Hooks#
Starlite includes several application level hooks that allow users to run their own sync or async callables. While you are free to use these hooks as you see fit, the design intention behind them is to allow for easy instrumentation for observability (monitoring, tracing, logging etc.).
Note
All application hook kwargs detailed below receive either a single callable or a list of callables. If a list is provided, it is called in the order it is given.
Before / After Startup#
The before_startup and after_startup hooks take a sync or async callable that
receives the Starlite application as an argument and run during the ASGI startup event. The callable is invoked
respectively before or after the list of callables defined in the on_startup list of callables.
import logging
from asyncio import sleep
from datetime import datetime
from starlite import Starlite
logger = logging.getLogger()
async def startup_callable() -> None:
"""Function called during 'on_startup'."""
await sleep(0.5)
def before_startup_handler(app_instance: Starlite) -> None:
"""Function called before 'on_startup'."""
start_time = datetime.now()
app_instance.state.start_time = start_time.timestamp()
logger.info("startup sequence begin at %s", start_time.isoformat())
def after_startup_handler(app_instance: Starlite) -> None:
"""Function called after 'on_startup'."""
logger.info(
"startup sequence ended at: %s, time elapsed: %d",
datetime.now().isoformat(),
datetime.now().timestamp() - app_instance.state.start_time,
)
app = Starlite(
[],
on_startup=[startup_callable],
before_startup=before_startup_handler,
after_startup=after_startup_handler,
)
Before / After Shutdown#
The before_shutdown and after_shutdown are basically identical, with the difference being that the
callable they receive in callable is invoked respectively before or after the
list of callables defined in the on_shutdown list of callables.
import logging
from asyncio import sleep
from datetime import datetime
from starlite import Starlite
logger = logging.getLogger()
async def shutdown_callable() -> None:
"""Function called during 'on_shutdown'."""
await sleep(0.5)
def before_shutdown_handler(app_instance: Starlite) -> None:
"""Function called before 'on_shutdown'."""
start_time = datetime.now()
app_instance.state.start_time = start_time.timestamp()
logger.info("shutdown sequence begin at %s", start_time.isoformat())
def after_shutdown_handler(app_instance: Starlite) -> None:
"""Function called after 'on_shutdown'."""
logger.info(
"shutdown sequence ended at: %s, time elapsed: %d",
datetime.now().isoformat(),
datetime.now().timestamp() - app_instance.state.start_time,
)
app = Starlite(
[],
on_shutdown=[shutdown_callable],
before_shutdown=before_shutdown_handler,
after_shutdown=after_shutdown_handler,
)
After Exception#
The after_exception hook takes a sync or async callable that is called with
three arguments: the exception that occurred, the ASGI scope of the request or websocket connection and the
application state.
import logging
from typing import TYPE_CHECKING
from starlite import HTTPException, Starlite, get
from starlite.status_codes import HTTP_400_BAD_REQUEST
logger = logging.getLogger()
if TYPE_CHECKING:
from starlite.datastructures import State
from starlite.types import Scope
@get("/some-path")
def my_handler() -> None:
"""Route handler that raises an exception."""
raise HTTPException(detail="bad request", status_code=HTTP_400_BAD_REQUEST)
async def after_exception_handler(exc: Exception, scope: "Scope", state: "State") -> None:
"""Hook function that will be invoked after each exception."""
if not hasattr(state, "error_count"):
state.error_count = 1
else:
state.error_count += 1
logger.info(
"an exception of type %s has occurred for requested path %s and the application error count is %d.",
type(exc).__name__,
scope["path"],
state.error_count,
)
app = Starlite([my_handler], after_exception=after_exception_handler)
Attention
This hook is not meant to handle exceptions - it just receives them to allow for side effects. To handle exceptions you should define exception handlers.
Before Send#
The before_send hook takes a sync or async callable that is called when
an ASGI message is sent. The hook receives the message instance and the application state.
from typing import TYPE_CHECKING, Dict
from starlite import Starlite, get
from starlite.datastructures import MutableScopeHeaders
if TYPE_CHECKING:
from starlite.datastructures import State
from starlite.types import Message
@get("/test")
def handler() -> Dict[str, str]:
"""Example Handler function."""
return {"key": "value"}
async def before_send_hook_handler(message: "Message", state: "State") -> None:
"""The function will be called on each ASGI message.
We therefore ensure it runs only on the message start event.
"""
if message["type"] == "http.response.start":
headers = MutableScopeHeaders.from_message(message=message)
headers["My Header"] = state.message
def on_startup(state: "State") -> None:
"""A function that will populate the app state before any requests are received."""
state.message = "value injected during send"
app = Starlite(route_handlers=[handler], on_startup=[on_startup], before_send=before_send_hook_handler)
from typing import TYPE_CHECKING
from starlite import Starlite, get
from starlite.datastructures import MutableScopeHeaders
if TYPE_CHECKING:
from starlite.datastructures import State
from starlite.types import Message
@get("/test")
def handler() -> dict[str, str]:
"""Example Handler function."""
return {"key": "value"}
async def before_send_hook_handler(message: "Message", state: "State") -> None:
"""The function will be called on each ASGI message.
We therefore ensure it runs only on the message start event.
"""
if message["type"] == "http.response.start":
headers = MutableScopeHeaders.from_message(message=message)
headers["My Header"] = state.message
def on_startup(state: "State") -> None:
"""A function that will populate the app state before any requests are received."""
state.message = "value injected during send"
app = Starlite(route_handlers=[handler], on_startup=[on_startup], before_send=before_send_hook_handler)
Application Init#
Starlite includes a hook for intercepting the arguments passed to the Starlite constructor,
before they are used to instantiate the application.
Handlers can be passed to the on_app_init parameter on construction of the application, and in turn, each will receive
an instance of AppConfig and must return an instance of same.
This hook is useful for applying common configuration between applications, and for use by developers who may wish to develop third-party application configuration systems.
Note
on_app_init handlers cannot be async def functions, as they are called within Starlite.__init__(), outside of an async context.
from typing import TYPE_CHECKING
from starlite import Starlite
if TYPE_CHECKING:
from starlite.config import AppConfig
async def close_db_connection() -> None:
"""Closes the database connection on application shutdown."""
def receive_app_config(app_config: "AppConfig") -> "AppConfig":
"""Receives parameters from the application.
In reality, this would be a library of boilerplate that is carried from one application to another, or a third-party
developed application configuration tool.
"""
app_config.on_shutdown.append(close_db_connection)
return app_config
app = Starlite([], on_app_init=[receive_app_config])
Layered architecture#
Starlite has a layered architecture compromising of (generally speaking) 4 layers:
The application object
Routers
Controllers
Handlers
There are many parameters that can be defined on every layer, in which case the parameter defined on the layer closest to the handler takes precedence. This allows for maximum flexibility and simplicity when configuring complex applications and enables transparent overriding of parameters.
Parameters that support layering are:
securitytagstype_encoders