## Example
+**example.py**:
+
```python
from starlette.applications import Starlette
from starlette.responses import JSONResponse
-import uvicorn
-
-app = Starlette(debug=True)
+from starlette.routing import Route
-@app.route('/')
async def homepage(request):
return JSONResponse({'hello': 'world'})
-if __name__ == '__main__':
- uvicorn.run(app, host='0.0.0.0', port=8000)
+routes = [
+ Route("/", endpoint=homepage)
+]
+
+app = Starlette(debug=True, routes=route)
+```
+
+Then run the application using Uvicorn:
+
+```shell
+$ uvicorn example:app
```
For a more complete example, see [encode/starlette-example](https://github.com/encode/starlette-example).
```python
from starlette.applications import Starlette
from starlette.responses import PlainTextResponse
+from starlette.routing import Route, Mount, WebSocketRoute
from starlette.staticfiles import StaticFiles
-app = Starlette()
-app.debug = True
-app.mount('/static', StaticFiles(directory="static"))
-
-
-@app.route('/')
def homepage(request):
return PlainTextResponse('Hello, world!')
-@app.route('/user/me')
def user_me(request):
username = "John Doe"
return PlainTextResponse('Hello, %s!' % username)
-@app.route('/user/{username}')
def user(request):
username = request.path_params['username']
return PlainTextResponse('Hello, %s!' % username)
-
-@app.websocket_route('/ws')
async def websocket_endpoint(websocket):
await websocket.accept()
await websocket.send_text('Hello, websocket!')
await websocket.close()
-
-@app.on_event('startup')
def startup():
print('Ready to go')
-```
-
-### Instantiating the application
-
-* `Starlette(debug=False)` - Create a new Starlette application.
-
-### Adding routes to the application
-You can use any of the following to add handled routes to the application:
-* `app.add_route(path, func, methods=["GET"])` - Add an HTTP route. The function may be either a coroutine or a regular function, with a signature like `func(request, **kwargs) -> response`.
-* `app.add_websocket_route(path, func)` - Add a websocket session route. The function must be a coroutine, with a signature like `func(session, **kwargs)`.
-* `@app.route(path)` - Add an HTTP route, decorator style.
-* `@app.websocket_route(path)` - Add a WebSocket route, decorator style.
+routes = [
+ Route('/', homepage),
+ Route('/user/me', user_me),
+ Route('/user/{username}', user),
+ WebSocketRoute('/ws', websocket_endpoint),
+ Mount('/static', StaticFiles(directory="static")),
+]
-### Adding event handlers to the application
-
-There are two ways to add event handlers:
-
-* `@app.on_event(event_type)` - Add an event, decorator style
-* `app.add_event_handler(event_type, func)` - Add an event through a function call.
-
-`event_type` must be specified as either `'startup'` or `'shutdown'`.
-
-### Submounting other applications
-
-Submounting applications is a powerful way to include reusable ASGI applications.
-
-* `app.mount(prefix, app)` - Include an ASGI app, mounted under the given path prefix
-
-### Customizing exception handling
+app = Starlette(debug=True, routes=routes, on_startup=[startup])
+```
-You can use either of the following to catch and handle particular types of
-exceptions that occur within the application:
+### Instantiating the application
-* `app.add_exception_handler(exc_class_or_status_code, handler)` - Add an error handler. The handler function may be either a coroutine or a regular function, with a signature like `func(request, exc) -> response`.
-* `@app.exception_handler(exc_class_or_status_code)` - Add an error handler, decorator style.
-* `app.debug` - Enable or disable error tracebacks in the browser.
+::: starlette.applications.Starlette
+ :docstring:
### Storing state on the app instance
app.state.ADMIN_EMAIL = 'admin@example.org'
```
-### Acessing the app instance
+### Accessing the app instance
-Where a `request` is available (i.e. endpoints and middleware), the app is available on `request.app`. For other situations it can be imported from wherever it's instantiated.
+Where a `request` is available (i.e. endpoints and middleware), the app is available on `request.app`.
AuthenticationBackend, AuthenticationError, SimpleUser, UnauthenticatedUser,
AuthCredentials
)
+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
raise AuthenticationError('Invalid basic auth credentials')
username, _, password = decoded.partition(":")
- # TODO: You'd want to verify the username and password here,
- # possibly by installing `DatabaseMiddleware`
- # and retrieving user information from `request.database`.
+ # TODO: You'd want to verify the username and password here.
return AuthCredentials(["authenticated"]), SimpleUser(username)
-app = Starlette()
-app.add_middleware(AuthenticationMiddleware, backend=BasicAuthBackend())
-
-
-@app.route('/')
async def homepage(request):
if request.user.is_authenticated:
- return PlainTextResponse('hello, ' + request.user.display_name)
- return PlainTextResponse('hello, you')
+ 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
from starlette.authentication import requires
-@app.route('/dashboard')
@requires('authenticated')
async def dashboard(request):
...
from starlette.authentication import requires
-@app.route('/dashboard')
@requires(['authenticated', 'admin'])
async def dashboard(request):
...
from starlette.authentication import requires
-@app.route('/dashboard')
@requires(['authenticated', 'admin'], status_code=404)
async def dashboard(request):
...
from starlette.authentication import requires
-@app.route('/homepage')
async def homepage(request):
...
-@app.route('/dashboard')
@requires('authenticated', redirect='homepage')
async def dashboard(request):
...
around a method on the class.
```python
-@app.route("/dashboard")
class Dashboard(HTTPEndpoint):
@requires("authenticated")
async def get(self, request):
```python
from starlette.applications import Starlette
from starlette.responses import JSONResponse
+from starlette.routing import Route
from starlette.background import BackgroundTask
-app = Starlette()
-@app.route('/user/signup', methods=['POST'])
+...
+
async def signup(request):
data = await request.json()
username = data['username']
async def send_welcome_email(to_address):
...
+
+
+routes = [
+ ...
+ Route('/user/signup', endpoint=signup, methods=['POST'])
+]
+
+app = Starlette(routes=routes)
```
### BackgroundTasks
from starlette.responses import JSONResponse
from starlette.background import BackgroundTasks
-app = Starlette()
-
-@app.route('/user/signup', methods=['POST'])
async def signup(request):
data = await request.json()
username = data['username']
async def send_admin_notification(username):
...
+routes = [
+ Route('/user/signup', endpoint=signup, methods=['POST'])
+]
+
+app = Starlette(routes=routes)
```
SECRET_KEY = config('SECRET_KEY', cast=Secret)
ALLOWED_HOSTS = config('ALLOWED_HOSTS', cast=CommaSeparatedStrings)
-app = Starlette()
-app.debug = DEBUG
+app = Starlette(debug=DEBUG)
...
```
```python
from starlette.applications import Starlette
-from starlette.middleware.database import DatabaseMiddleware
+from starlette.middleware import Middleware
from starlette.middleware.session import SessionMiddleware
+from starlette.routing import Route
from myproject import settings
-app = Starlette()
+async def homepage(request):
+ ...
-app.debug = settings.DEBUG
+routes = [
+ Route("/", endpoint=homepage)
+]
-app.add_middleware(
- SessionMiddleware,
- secret_key=settings.SECRET_KEY,
-)
-app.add_middleware(
- DatabaseMiddleware,
- database_url=settings.DATABASE_URL,
- rollback_on_shutdown=settings.TESTING
-)
+middleware = [
+ Middleware(
+ SessionMiddleware,
+ secret_key=settings.SECRET_KEY,
+ )
+]
-@app.route('/', methods=['GET'])
-async def homepage(request):
- ...
+app = Starlette(debug=settings.DEBUG, routes=routes, middleware=middleware)
```
Now let's deal with our test configuration.
sqlalchemy.Column("completed", sqlalchemy.Boolean),
)
-# Main application code.
database = databases.Database(DATABASE_URL)
-app = Starlette()
-
-
-@app.on_event("startup")
-async def startup():
- await database.connect()
-
-
-@app.on_event("shutdown")
-async def shutdown():
- await database.disconnect()
-@app.route("/notes", methods=["GET"])
+# Main application code.
async def list_notes(request):
query = notes.select()
results = await database.fetch_all(query)
]
return JSONResponse(content)
-
-@app.route("/notes", methods=["POST"])
async def add_note(request):
data = await request.json()
query = notes.insert().values(
"text": data["text"],
"completed": data["completed"]
})
+
+routes = [
+ Route("/notes", endpoint=list_notes, methods=["GET"]),
+ Route("/notes", endpoint=add_note, methods=["POST"]),
+]
+
+app = Starlette(
+ routes=routes,
+ on_startup=[database.connect],
+ on_shutdown=[database.disconnect]
+)
```
## Queries
```
If you're using a Starlette application instance to handle routing, you can
-dispatch to an `HTTPEndpoint` class by using the `@app.route()` decorator, or the
-`app.add_route()` function. Make sure to dispatch to the class itself, rather
-than to an instance of the class:
+dispatch to an `HTTPEndpoint` class. Make sure to dispatch to the class itself,
+rather than to an instance of the class:
```python
from starlette.applications import Starlette
from starlette.responses import PlainTextResponse
from starlette.endpoints import HTTPEndpoint
+from starlette.routing import Route
-app = Starlette()
-
-
-@app.route("/")
class Homepage(HTTPEndpoint):
async def get(self, request):
return PlainTextResponse(f"Hello, world!")
-@app.route("/{username}")
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
from starlette.applications import Starlette
from starlette.endpoints import WebSocketEndpoint, HTTPEndpoint
from starlette.responses import HTMLResponse
+from starlette.routing import Route, WebSocketRoute
-app = Starlette()
html = """
<!DOCTYPE html>
</html>
"""
-
-@app.route("/")
class Homepage(HTTPEndpoint):
async def get(self, request):
return HTMLResponse(html)
-
-@app.websocket_route("/ws")
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", WebSocketEndpoint)
+]
-if __name__ == "__main__":
- uvicorn.run(app, host="0.0.0.0", port=8000)
+app = Starlette(routes=routes)
```
These event handlers can either be `async` coroutines, or regular syncronous
functions.
-The event handlers can be registered with a decorator syntax, like so:
+The event handlers should be included on the application like so:
```python
from starlette.applications import Starlette
-app = Starlette()
+async def some_startup_task():
+ pass
-@app.on_event('startup')
-async def open_database_connection_pool():
- ...
-
-@app.on_event('shutdown')
-async def close_database_connection_pool():
- ...
-```
-Or as a regular function call:
-
-```python
-from starlette.applications import Starlette
-
-
-app = Starlette()
+async def some_shutdown_task():
+ pass
-async def open_database_connection_pool():
+routes = [
...
+]
-async def close_database_connection_pool():
- ...
-
-app.add_event_handler('startup', open_database_connection_pool)
-app.add_event_handler('shutdown', close_database_connection_pool)
-
+app = Starlette(
+ routes=routes,
+ on_startup=[some_startup_task],
+ on_shutdown=[some_shutdown_task]
+)
```
Starlette will not start serving any incoming requests until all of the
def test_homepage():
with TestClient(app) as client:
- # Application 'startup' handlers are called on entering the block.
+ # Application 'on_startup' handlers are called on entering the block.
response = client.get("/")
assert response.status_code == 200
- # Application 'shutdown' handlers are called on exiting the block.
+ # Application 'on_shutdown' handlers are called on exiting the block.
```
HTML_500_PAGE = ...
-app = Starlette()
-
-
-@app.exception_handler(404)
async def not_found(request, exc):
return HTMLResponse(content=HTML_404_PAGE, status_code=exc.status_code)
-@app.exception_handler(500)
async def server_error(request, exc):
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.
```python
-app = Starlette(debug=True)
+app = Starlette(debug=True, routes=routes, exception_handlers=exception_handlers)
```
As well as registering handlers for specific status codes, you can also
is handled. For example, to use JSON style responses:
```python
-@app.exception_handler(HTTPException)
async def http_exception(request, exc):
return JSONResponse({"detail": exc.detail}, status_code=exc.status_code)
+
+exception_handlers = {
+ HTTPException: http_exception
+}
```
## Errors and handled exceptions
```python
from starlette.applications import Starlette
+from starlette.routing import Route
from starlette.graphql import GraphQLApp
import graphene
def resolve_hello(self, info, name):
return "Hello " + name
+routes = [
+ Route('/', GraphQLApp(schema=graphene.Schema(query=Query)))
+]
-app = Starlette()
-app.add_route('/', GraphQLApp(schema=graphene.Schema(query=Query)))
+app = Starlette(routes=route)
```
If you load up the page in a browser, you'll be served the GraphiQL tool,
from graphql.execution.executors.asyncio import AsyncioExecutor
from starlette.applications import Starlette
from starlette.graphql import GraphQLApp
+from starlette.routing import Route
import graphene
# We can make asynchronous network calls here.
return "Hello " + name
+routes = [
+ # We're using `executor_class=AsyncioExecutor` here.
+ Route('/', GraphQLApp(
+ schema=graphene.Schema(query=Query),
+ executor_class=AsyncioExecutor
+ ))
+]
-app = Starlette()
-
-# We're using `executor_class=AsyncioExecutor` here.
-app.add_route('/', GraphQLApp(schema=graphene.Schema(query=Query), executor_class=AsyncioExecutor))
+app = Starlette(routes=routes)
```
## Example
+**example.py**:
+
```python
from starlette.applications import Starlette
from starlette.responses import JSONResponse
-import uvicorn
+from starlette.routing import Route
-app = Starlette(debug=True)
-@app.route('/')
async def homepage(request):
return JSONResponse({'hello': 'world'})
-if __name__ == '__main__':
- uvicorn.run(app, host='0.0.0.0', port=8000)
+
+app = Starlette(debug=True, routes=[
+ Route('/', homepage),
+])
+```
+
+Then run the application...
+
+```shell
+$ uvicorn example:app
```
For a more complete example, [see here](https://github.com/encode/starlette-example).
-Starlette includes several middleware classes for adding behaviour that is applied across your entire application. These are all implemented as standard ASGI
+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.
The Starlette application class allows you to include the ASGI middleware
```python
from starlette.applications import Starlette
+from starlette.middleware import Middleware
from starlette.middleware.httpsredirect import HTTPSRedirectMiddleware
from starlette.middleware.trustedhost import TrustedHostMiddleware
-
-app = Starlette()
+routes = ...
# Ensure that all requests include an 'example.com' or '*.example.com' host header,
# and strictly enforce https-only access.
-app.add_middleware(TrustedHostMiddleware, allowed_hosts=['example.com', '*.example.com'])
-app.add_middleware(HTTPSRedirectMiddleware)
+middleware = [
+ Middleware(TrustedHostMiddleware, allowed_hosts=['example.com', '*.example.com']),
+ Middleware(HTTPSRedirectMiddleware)
+]
+
+app = Starlette(routes=routes, middleware=middleware)
```
-Starlette also allows you to add middleware functions, using a decorator syntax:
+Every Starlette application automatically includes two pieces of middleware
+by default:
-```python
-@app.middleware("http")
-async def add_custom_header(request, call_next):
- response = await call_next(request)
- response.headers['Custom'] = 'Example'
- return response
-```
+* `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:
```python
from starlette.applications import Starlette
+from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware
+routes = ...
+
+middleware = [
+ Middleware(CORSMiddleware, allow_origins=['*'])
+]
-app = Starlette()
-app.add_middleware(CORSMiddleware, allow_origins=['*'])
+app = Starlette(routes=routes, middleware=middleware)
```
The following arguments are supported:
from starlette.applications import Starlette
from starlette.middleware.httpsredirect import HTTPSRedirectMiddleware
+routes = ...
-app = Starlette()
-app.add_middleware(HTTPSRedirectMiddleware)
+middleware = [
+ Middleware(HTTPSRedirectMiddleware)
+]
+
+app = Starlette(routes=routes, middleware=middleware)
```
There are no configuration options for this middleware class.
```python
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()
-app.add_middleware(TrustedHostMiddleware, allowed_hosts=['example.com', '*.example.com'])
+app = Starlette(routes=routes, middleware=middleware)
```
The following arguments are supported:
```python
from starlette.applications import Starlette
+from starlette.middleware import Middleware
from starlette.middleware.gzip import GZipMiddleware
-app = Starlette()
-app.add_middleware(GZipMiddleware, minimum_size=1000)
+routes = ...
+
+middleware = [
+ Middleware(GZipMiddleware, minimum_size=1000)
+]
+
+app = Starlette(routes=routes, middleware=middleware)
```
The following arguments are supported:
response.headers['Custom'] = 'Example'
return response
+middleware = [
+ Middleware(CustomHeaderMiddleware)
+]
-app = Starlette()
-app.add_middleware(CustomHeaderMiddleware)
+app = Starlette(routes=routes, middleware=middleware)
```
If you want to provide configuration options to the middleware class you should
return response
-app = Starlette()
-app.add_middleware(CustomHeaderMiddleware, header_value='Customized')
+
+middleware = [
+ Middleware(CustomHeaderMiddleware, header_value='Customized')
+]
+
+app = Starlette(routes=routes, middleware=middleware)
```
Middleware classes should not modify their state outside of the `__init__` method.
```
You can do this with a Starlette application instance too, but it is preferable
-to use `.add_middleware`, as it'll ensure that you don't lose the reference
-to the application object, and that the exception handling always wraps around
-any other behaviour.
+to use the `middleware=<List of Middleware instances>` style, as it will:
+
+* Ensure that everything remains wrapped in a single outermost `ServerErrorMiddleware`.
+* Preserves the top-level `app` instance.
## Third party middleware
-
-Starlette includes a `Router` class which is an ASGI application that
-dispatches incoming requests to endpoints or submounted applications.
+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.
```python
-from starlette.routing import Mount, Route, Router
-from myproject import Homepage, SubMountedApp
+from starlette.applications import Starlette
+from starlette.responses import PlainTextResponse
+from starlette.routing import Route
-app = Router([
- Route('/', endpoint=Homepage, methods=['GET']),
- Mount('/mount', app=SubMountedApp)
-])
+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 [class based
+views](endpoints.md).
+
+## Path Parameters
+
Paths can use URI templating style to capture path components.
```python
-Route('/users/{username}', endpoint=User, methods=['GET'])
+Route('/users/{username}', user)
```
Convertors for `int`, `float`, and `path` are also available:
```python
-Route('/users/{user_id:int}', endpoint=User, methods=['GET'])
+Route('/users/{user_id:int}', user)
```
Path parameters are made available in the request, as the `request.path_params`
dictionary.
-Because the target of a `Mount` is an ASGI instance itself, routers
-allow for easy composition. For example:
+```python
+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:
```python
-app = Router([
- Route('/', endpoint=Homepage, methods=['GET']),
- Mount('/users', app=Router([
- Route('/', endpoint=Users, methods=['GET', 'POST']),
- Route('/{username}', endpoint=User, methods=['GET']),
- ]))
-])
+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.
+
+```python
+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.
+
+```python
+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...
+
+```python
+# 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", 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.
+
+```python
+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...
+
+```python
+routes = [
+ Route("/users/{username}", user, name="user_detail")
+]
+
+# We can use the following to return a URL...
+url = request.url_for("user_detail", username=...)
```
-The router will respond with "404 Not found" or "405 Method not allowed"
-responses for requests which do not match.
+If a `Mount` includes a `name`, then submounts should use a `{prefix}:{name}`
+style for reverse URL lookups.
+
+```python
+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.
+
+```python
+routes = [
+ ...
+ Mount("/static", 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.
+
+```python
+url = app.url_path_for("user_detail", username=...)
+```
+
+## Route priority
Incoming paths are matched against each `Route` in order.
-If you need to have a `Route` with a fixed path that would also match a
-`Route` with parameters you should add the `Route` with the fixed path first.
+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, with an additional `Route` like:
+For example:
```python
-Route('/users/me', endpoint=UserMe, methods=['GET'])
+# 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),
+]
```
-You should add that route for `/users/me` before the one for `/users/{username}`:
+## 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.
```python
-app = Router([
- Route('/users/me', endpoint=UserMe, methods=['GET']),
- Route('/users/{username}', endpoint=User, methods=['GET']),
+app = Router(routes=[
+ Route('/', homepage),
+ Mount('/users', routes=[
+ Route('/', users, methods=['GET', 'POST']),
+ Route('/{username}', user),
+ ])
])
```
```python
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"}}
)
-app = Starlette()
-
-@app.route("/users", methods=["GET"])
def list_users(request):
"""
responses:
raise NotImplementedError()
-@app.route("/users", methods=["POST"])
def create_user(request):
"""
responses:
raise NotImplementedError()
-@app.route("/schema", methods=["GET"], include_in_schema=False)
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()
```
We can now access an OpenAPI schema at the "/schema" endpoint.
assert sys.argv[-1] in ("run", "schema"), "Usage: example.py [run|schema]"
if sys.argv[-1] == "run":
- uvicorn.run(app, host='0.0.0.0', port=8000)
+ 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))
```python
from starlette.applications import Starlette
from starlette.responses import HTMLResponse
+from starlette.routing import Route, Mount
from starlette.staticfiles import StaticFiles
-app = Starlette()
-
-@app.route("/")
async def homepage(request):
"""
Homepage which uses server push to deliver the stylesheet.
'<html><head><link rel="stylesheet" href="/static/style.css"/></head></html>'
)
+routes = [
+ Route("/", endpoint=homepage),
+ Mount("/static", StaticFiles(directory="static"), name="static")
+]
-app.mount("/static", StaticFiles(directory="static"))
+app = Starlette(routes=routes)
```
comprehensive static file serving.
```python
-from starlette.routing import Router, Mount
+from starlette.applications import Starlette
+from starlette.routing import Mount
from starlette.staticfiles import StaticFiles
-app = Router(routes=[
+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"
a python package. The Python "bootstrap4" package is an example of this.
```python
-from starlette.routing import Router, Mount
+from starlette.applications import Starlette
+from starlette.routing import Mount
from starlette.staticfiles import StaticFiles
-app = Router(routes=[
+routes=[
+ ...
Mount('/static', app=StaticFiles(directory='static', packages=['bootstrap4']), name="static"),
-])
+]
+
+app = Starlette(routes=routes)
```
You may prefer to include static files directly inside the "static" directory
```python
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')
-app = Starlette(debug=True)
-app.mount('/static', StaticFiles(directory='static'), name='static')
-
-
-@app.route('/')
async def homepage(request):
return templates.TemplateResponse('index.html', {'request': request})
+
+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
markdown_extensions:
- markdown.extensions.codehilite:
guess_lang: false
+ - mkautodoc
# Documentation
mkdocs
mkdocs-material
+mkautodoc
class Starlette:
+ """
+ Creates an application instance.
+
+ **Parameters:**
+
+ * **debug** - Boolean indicating if debug tracebacks should be returned on errors.
+ * **routes** - A list of routes to serve incoming HTTP and WebSocket requests.
+ * **middleware** - A list of middleware to run for every request. A starlette
+ application will always automatically include two middleware classes.
+ `ServerErrorMiddleware` is added the very outermost middleware, to handle
+ any uncaught errors occuring anywhere in the entire stack.
+ `ExceptionMiddleware` is added as the very innermost middleware, to deal
+ with handled exception cases occuring in the routing or endpoints.
+ * **exception_handlers** - A dictionary mapping either integer status codes,
+ or exception class types onto callables which handle the exceptions.
+ Exception handler callables should be of the form `handler(request, exc) -> response`
+ and may be be either standard functions, or async functions.
+ * **on_startup** - A list of callables to run on application startup.
+ Startup handler callables do not take any arguments, and may be be either
+ standard functions, or async functions.
+ * **on_shutdown** - A list of callables to run on application shutdown.
+ Shutdown handler callables do not take any arguments, and may be be either
+ standard functions, or async functions.
+ """
+
def __init__(
self,
debug: bool = False,
routes: typing.List[BaseRoute] = None,
- middleware: typing.List[Middleware] = None,
+ middleware: typing.List[typing.Optional[Middleware]] = None,
exception_handlers: typing.Dict[
typing.Union[int, typing.Type[Exception]], typing.Callable
] = None,
self.exception_handlers = (
{} if exception_handlers is None else dict(exception_handlers)
)
- self.user_middleware = list(middleware or [])
+ self.user_middleware = [item for item in middleware or [] if item is not None]
self.middleware_stack = self.build_middleware_stack()
def build_middleware_stack(self) -> ASGIApp:
else:
exception_handlers[key] = value
- server_errors = Middleware(
- ServerErrorMiddleware, options={"handler": error_handler, "debug": debug},
- )
- exceptions = Middleware(
- ExceptionMiddleware,
- options={"handlers": exception_handlers, "debug": debug},
+ middleware = (
+ [Middleware(ServerErrorMiddleware, handler=error_handler, debug=debug,)]
+ + self.user_middleware
+ + [
+ Middleware(
+ ExceptionMiddleware, handlers=exception_handlers, debug=debug,
+ )
+ ]
)
- middleware = [server_errors] + self.user_middleware + [exceptions]
-
app = self.router
- for cls, options, enabled in reversed(middleware):
- if enabled:
- app = cls(app=app, **options)
+ for cls, options in reversed(middleware):
+ app = cls(app=app, **options)
return app
@property
self._debug = value
self.middleware_stack = self.build_middleware_stack()
+ def url_path_for(self, name: str, **path_params: str) -> URLPath:
+ return self.router.url_path_for(name, **path_params)
+
+ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+ scope["app"] = self
+ await self.middleware_stack(scope, receive, send)
+
+ # The following usages are now discouraged in favour of configuration
+ # during Starlette.__init__(...)
def on_event(self, event_type: str) -> typing.Callable:
- return self.router.lifespan.on_event(event_type)
+ return self.router.on_event(event_type)
def mount(self, path: str, app: ASGIApp, name: str = None) -> None:
self.router.mount(path, app=app, name=name)
def host(self, host: str, app: ASGIApp, name: str = None) -> None:
self.router.host(host, app=app, name=name)
- def add_middleware(self, middleware_class: type, **kwargs: typing.Any) -> None:
- self.user_middleware.insert(0, Middleware(middleware_class, options=kwargs))
+ def add_middleware(self, middleware_class: type, **options: typing.Any) -> None:
+ self.user_middleware.insert(0, Middleware(middleware_class, **options))
self.middleware_stack = self.build_middleware_stack()
def add_exception_handler(
self.middleware_stack = self.build_middleware_stack()
def add_event_handler(self, event_type: str, func: typing.Callable) -> None:
- self.router.lifespan.add_event_handler(event_type, func)
+ self.router.add_event_handler(event_type, func)
def add_route(
self,
return func
return decorator
-
- def url_path_for(self, name: str, **path_params: str) -> URLPath:
- return self.router.url_path_for(name, **path_params)
-
- async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
- scope["app"] = self
- await self.middleware_stack(scope, receive, send)
class Middleware:
- def __init__(
- self, cls: type, options: dict = None, enabled: typing.Any = True
- ) -> None:
+ def __init__(self, cls: type, **options: typing.Any) -> None:
self.cls = cls
- self.options = dict(options) if options else {}
- self.enabled = bool(enabled)
+ self.options = options
def __iter__(self) -> typing.Iterator:
- as_tuple = (self.cls, self.options, self.enabled)
+ as_tuple = (self.cls, self.options)
return iter(as_tuple)
def __repr__(self) -> str:
class_name = self.__class__.__name__
- options_repr = "" if not self.options else f", options={self.options!r}"
- enabled_repr = "" if self.enabled else ", enabled=False"
- return f"{class_name}({self.cls.__name__}{options_repr}{enabled_repr})"
+ option_strings = ["{key}={value!r}" for key, value in self.options.items()]
+ options_repr = ", ".join(option_strings)
+ return f"{class_name}({self.cls.__name__}{options_repr})"
def url_path_for(self, name: str, **path_params: str) -> URLPath:
raise NotImplementedError() # pragma: no cover
- async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+ async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
raise NotImplementedError() # pragma: no cover
+ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+ """
+ A route may be used in isolation as a stand-alone ASGI app.
+ This is a somewhat contrived case, as they'll almost always be used
+ within a Router, but could be useful for some tooling and minimal apps.
+ """
+ match, child_scope = self.matches(scope)
+ if match == Match.NONE:
+ if scope["type"] == "http":
+ response = PlainTextResponse("Not Found", status_code=404)
+ await response(scope, receive, send)
+ elif scope["type"] == "websocket":
+ websocket_close = WebSocketClose()
+ await websocket_close(scope, receive, send)
+ return
+
+ scope.update(child_scope)
+ await self.handle(scope, receive, send)
+
class Route(BaseRoute):
def __init__(
assert not remaining_params
return URLPath(path=path, protocol="http")
- async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+ async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
if self.methods and scope["method"] not in self.methods:
if "app" in scope:
raise HTTPException(status_code=405)
assert not remaining_params
return URLPath(path=path, protocol="websocket")
- async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+ async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
await self.app(scope, receive, send)
def __eq__(self, other: typing.Any) -> bool:
pass
raise NoMatchFound()
- async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+ async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
await self.app(scope, receive, send)
def __eq__(self, other: typing.Any) -> bool:
pass
raise NoMatchFound()
- async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+ async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
await self.app(scope, receive, send)
def __eq__(self, other: typing.Any) -> bool:
)
-class Lifespan(BaseRoute):
+class Router:
def __init__(
self,
- on_startup: typing.Union[typing.Callable, typing.List[typing.Callable]] = None,
- on_shutdown: typing.Union[typing.Callable, typing.List[typing.Callable]] = None,
- ):
- self.startup_handlers = self.to_list(on_startup)
- self.shutdown_handlers = self.to_list(on_shutdown)
-
- def to_list(
- self, item: typing.Union[typing.Callable, typing.List[typing.Callable]] = None
- ) -> typing.List[typing.Callable]:
- if item is None:
- return []
- return list(item) if isinstance(item, (list, tuple)) else [item]
-
- def matches(self, scope: Scope) -> typing.Tuple[Match, Scope]:
- if scope["type"] == "lifespan":
- return Match.FULL, {}
- return Match.NONE, {}
+ routes: typing.List[BaseRoute] = None,
+ redirect_slashes: bool = True,
+ default: ASGIApp = None,
+ on_startup: typing.List[typing.Callable] = None,
+ on_shutdown: typing.List[typing.Callable] = None,
+ ) -> None:
+ self.routes = [] if routes is None else list(routes)
+ self.redirect_slashes = redirect_slashes
+ self.default = self.not_found if default is None else default
+ self.on_startup = [] if on_startup is None else list(on_startup)
+ self.on_shutdown = [] if on_shutdown is None else list(on_shutdown)
- def add_event_handler(self, event_type: str, func: typing.Callable) -> None:
- assert event_type in ("startup", "shutdown")
+ async def not_found(self, scope: Scope, receive: Receive, send: Send) -> None:
+ if scope["type"] == "websocket":
+ websocket_close = WebSocketClose()
+ await websocket_close(scope, receive, send)
+ return
- if event_type == "startup":
- self.startup_handlers.append(func)
+ # If we're running inside a starlette application then raise an
+ # exception, so that the configurable exception handler can deal with
+ # returning the response. For plain ASGI apps, just return the response.
+ if "app" in scope:
+ raise HTTPException(status_code=404)
else:
- assert event_type == "shutdown"
- self.shutdown_handlers.append(func)
-
- def on_event(self, event_type: str) -> typing.Callable:
- def decorator(func: typing.Callable) -> typing.Callable:
- self.add_event_handler(event_type, func)
- return func
+ response = PlainTextResponse("Not Found", status_code=404)
+ await response(scope, receive, send)
- return decorator
+ def url_path_for(self, name: str, **path_params: str) -> URLPath:
+ for route in self.routes:
+ try:
+ return route.url_path_for(name, **path_params)
+ except NoMatchFound:
+ pass
+ raise NoMatchFound()
async def startup(self) -> None:
- for handler in self.startup_handlers:
+ """
+ Run any `.on_startup` event handlers.
+ """
+ for handler in self.on_startup:
if asyncio.iscoroutinefunction(handler):
await handler()
else:
handler()
async def shutdown(self) -> None:
- for handler in self.shutdown_handlers:
+ """
+ Run any `.on_shutdown` event handlers.
+ """
+ for handler in self.on_shutdown:
if asyncio.iscoroutinefunction(handler):
await handler()
else:
handler()
- async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+ async def lifespan(self, scope: Scope, receive: Receive, send: Send) -> None:
+ """
+ Handle ASGI lifespan messages, which allows us to manage application
+ startup and shutdown events.
+ """
message = await receive()
assert message["type"] == "lifespan.startup"
await self.shutdown()
await send({"type": "lifespan.shutdown.complete"})
+ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+ """
+ The main entry point to the Router class.
+ """
+ assert scope["type"] in ("http", "websocket", "lifespan")
-class Router:
- def __init__(
- self,
- routes: typing.List[BaseRoute] = None,
- redirect_slashes: bool = True,
- default: ASGIApp = None,
- on_startup: typing.List[typing.Callable] = None,
- on_shutdown: typing.List[typing.Callable] = None,
- ) -> None:
- self.routes = [] if routes is None else list(routes)
- self.redirect_slashes = redirect_slashes
- self.default = self.not_found if default is None else default
- self.lifespan = Lifespan(on_startup=on_startup, on_shutdown=on_shutdown)
+ if "router" not in scope:
+ scope["router"] = self
+ if scope["type"] == "lifespan":
+ await self.lifespan(scope, receive, send)
+ return
+
+ partial = None
+
+ for route in self.routes:
+ # Determine if any route matches the incoming scope,
+ # and hand over to the matching route if found.
+ match, child_scope = route.matches(scope)
+ if match == Match.FULL:
+ scope.update(child_scope)
+ await route.handle(scope, receive, send)
+ return
+ elif match == Match.PARTIAL and partial is None:
+ partial = route
+ partial_scope = child_scope
+
+ if partial is not None:
+ # Handle partial matches. These are cases where an endpoint is
+ # able to handle the request, but is not a preferred option.
+ # We use this in particular to deal with "406 Method Not Found".
+ scope.update(partial_scope)
+ await partial.handle(scope, receive, send)
+ return
+
+ if scope["type"] == "http" and self.redirect_slashes:
+ if not scope["path"].endswith("/"):
+ redirect_scope = dict(scope)
+ redirect_scope["path"] += "/"
+
+ for route in self.routes:
+ match, child_scope = route.matches(redirect_scope)
+ if match != Match.NONE:
+ redirect_url = URL(scope=redirect_scope)
+ response = RedirectResponse(url=str(redirect_url))
+ await response(scope, receive, send)
+ return
+
+ await self.default(scope, receive, send)
+
+ def __eq__(self, other: typing.Any) -> bool:
+ return isinstance(other, Router) and self.routes == other.routes
+
+ # The following usages are now discouraged in favour of configuration
+ # during Router.__init__(...)
def mount(self, path: str, app: ASGIApp, name: str = None) -> None:
route = Mount(path, app=app, name=name)
self.routes.append(route)
return decorator
- async def not_found(self, scope: Scope, receive: Receive, send: Send) -> None:
- if scope["type"] == "websocket":
- websocket_close = WebSocketClose()
- await websocket_close(receive, send)
- return
+ def add_event_handler(self, event_type: str, func: typing.Callable) -> None:
+ assert event_type in ("startup", "shutdown")
- # If we're running inside a starlette application then raise an
- # exception, so that the configurable exception handler can deal with
- # returning the response. For plain ASGI apps, just return the response.
- if "app" in scope:
- raise HTTPException(status_code=404)
+ if event_type == "startup":
+ self.on_startup.append(func)
else:
- response = PlainTextResponse("Not Found", status_code=404)
- await response(scope, receive, send)
-
- def url_path_for(self, name: str, **path_params: str) -> URLPath:
- for route in self.routes:
- try:
- return route.url_path_for(name, **path_params)
- except NoMatchFound:
- pass
- raise NoMatchFound()
-
- async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
- assert scope["type"] in ("http", "websocket", "lifespan")
+ self.on_shutdown.append(func)
- if "router" not in scope:
- scope["router"] = self
-
- partial = None
-
- for route in self.routes:
- match, child_scope = route.matches(scope)
- if match == Match.FULL:
- scope.update(child_scope)
- await route(scope, receive, send)
- return
- elif match == Match.PARTIAL and partial is None:
- partial = route
- partial_scope = child_scope
-
- if partial is not None:
- scope.update(partial_scope)
- await partial(scope, receive, send)
- return
-
- if scope["type"] == "http" and self.redirect_slashes:
- if scope["path"].endswith("/"):
- redirect_path = scope["path"].rstrip("/")
- else:
- redirect_path = scope["path"] + "/"
-
- if redirect_path: # Note that we skip the "/" -> "" case.
- redirect_scope = dict(scope)
- redirect_scope["path"] = redirect_path
-
- for route in self.routes:
- match, child_scope = route.matches(redirect_scope)
- if match != Match.NONE:
- redirect_url = URL(scope=redirect_scope)
- response = RedirectResponse(url=str(redirect_url))
- await response(scope, receive, send)
- return
-
- if scope["type"] == "lifespan":
- await self.lifespan(scope, receive, send)
- else:
- await self.default(scope, receive, send)
+ def on_event(self, event_type: str) -> typing.Callable:
+ def decorator(func: typing.Callable) -> typing.Callable:
+ self.add_event_handler(event_type, func)
+ return func
- def __eq__(self, other: typing.Any) -> bool:
- return isinstance(other, Router) and self.routes == other.routes
+ return decorator
def __init__(self, code: int = 1000) -> None:
self.code = code
- async def __call__(self, receive: Receive, send: Send) -> None:
+ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
await send({"type": "websocket.close", "code": self.code})
assert response.headers["Custom-Header"] == "Example"
-def test_app_disabled_middleware_argument():
- def homepage(request):
- return PlainTextResponse("Homepage")
-
- app = Starlette(
- routes=[Route("/", homepage)],
- middleware=[Middleware(CustomMiddleware, enabled=False)],
- )
-
- client = TestClient(app)
- response = client.get("/")
- assert "Custom-Header" not in response.headers
-
-
def test_middleware_repr():
middleware = Middleware(CustomMiddleware)
assert repr(middleware) == "Middleware(CustomMiddleware)"
from starlette.applications import Starlette
from starlette.responses import PlainTextResponse
-from starlette.routing import Lifespan, Route, Router
+from starlette.routing import Route, Router
from starlette.testclient import TestClient
shutdown_complete = True
app = Router(
- routes=[
- Lifespan(on_startup=run_startup, on_shutdown=run_shutdown),
- Route("/", hello_world),
- ]
+ on_startup=[run_startup],
+ on_shutdown=[run_shutdown],
+ routes=[Route("/", hello_world),],
)
assert not startup_complete
def run_startup():
raise RuntimeError()
- router = Router(routes=[Lifespan(on_startup=run_startup)])
+ router = Router(on_startup=[run_startup])
async def app(scope, receive, send):
async def _send(message):
def run_shutdown():
raise RuntimeError()
- app = Router(routes=[Lifespan(on_shutdown=run_shutdown)])
+ app = Router(on_shutdown=[run_shutdown])
with pytest.raises(RuntimeError):
with TestClient(app):
app = Starlette(routes=double_mount_routes)
url = app.url_path_for("mount:static", path="123")
assert url == "/mount/static/123"
+
+
+def test_standalone_route_matches():
+ app = Route("/", PlainTextResponse("Hello, World!"))
+ client = TestClient(app)
+ response = client.get("/")
+ assert response.status_code == 200
+ assert response.text == "Hello, World!"
+
+
+def test_standalone_route_does_not_match():
+ app = Route("/", PlainTextResponse("Hello, World!"))
+ client = TestClient(app)
+ response = client.get("/invalid")
+ assert response.status_code == 404
+ assert response.text == "Not Found"
+
+
+async def ws_helloworld(websocket):
+ await websocket.accept()
+ await websocket.send_text("Hello, world!")
+ await websocket.close()
+
+
+def test_standalone_ws_route_matches():
+ app = WebSocketRoute("/", ws_helloworld)
+ client = TestClient(app)
+ with client.websocket_connect("/") as websocket:
+ text = websocket.receive_text()
+ assert text == "Hello, world!"
+
+
+def test_standalone_ws_route_does_not_match():
+ app = WebSocketRoute("/", ws_helloworld)
+ client = TestClient(app)
+ with pytest.raises(WebSocketDisconnect):
+ client.websocket_connect("/invalid")
projects / thirdparty / starlette.git / commitdiff
