return self.router.on_event(event_type)
def mount(self, path: str, app: ASGIApp, name: str = None) -> None:
+ """
+ We no longer document this API, and its usage is discouraged.
+ Instead you should use the following approach:
+
+ routes = [
+ Mount(path, ...),
+ ...
+ ]
+
+ app = Starlette(routes=routes)
+ """
+
self.router.mount(path, app=app, name=name)
- def host(self, host: str, app: ASGIApp, name: str = None) -> None:
+ def host(
+ self, host: str, app: ASGIApp, name: str = None
+ ) -> None: # pragma: no cover
+ """
+ We no longer document this API, and its usage is discouraged.
+ Instead you should use the following approach:
+
+ routes = [
+ Host(path, ...),
+ ...
+ ]
+
+ app = Starlette(routes=routes)
+ """
+
self.router.host(host, app=app, name=name)
def add_middleware(self, middleware_class: type, **options: typing.Any) -> None:
self,
exc_class_or_status_code: typing.Union[int, typing.Type[Exception]],
handler: typing.Callable,
- ) -> None:
+ ) -> None: # pragma: no cover
self.exception_handlers[exc_class_or_status_code] = handler
self.middleware_stack = self.build_middleware_stack()
- def add_event_handler(self, event_type: str, func: typing.Callable) -> None:
+ def add_event_handler(
+ self, event_type: str, func: typing.Callable
+ ) -> None: # pragma: no cover
self.router.add_event_handler(event_type, func)
def add_route(
methods: typing.List[str] = None,
name: str = None,
include_in_schema: bool = True,
- ) -> None:
+ ) -> None: # pragma: no cover
self.router.add_route(
path, route, methods=methods, name=name, include_in_schema=include_in_schema
)
def add_websocket_route(
self, path: str, route: typing.Callable, name: str = None
- ) -> None:
+ ) -> None: # pragma: no cover
self.router.add_websocket_route(path, route, name=name)
def exception_handler(
# The following usages are now discouraged in favour of configuration
# during Router.__init__(...)
def mount(self, path: str, app: ASGIApp, name: str = None) -> None:
+ """
+ We no longer document this API, and its usage is discouraged.
+ Instead you should use the following approach:
+
+ routes = [
+ Mount(path, ...),
+ ...
+ ]
+
+ app = Starlette(routes=routes)
+ """
+
route = Mount(path, app=app, name=name)
self.routes.append(route)
- def host(self, host: str, app: ASGIApp, name: str = None) -> None:
+ def host(
+ self, host: str, app: ASGIApp, name: str = None
+ ) -> None: # pragma: no cover
+ """
+ We no longer document this API, and its usage is discouraged.
+ Instead you should use the following approach:
+
+ routes = [
+ Host(path, ...),
+ ...
+ ]
+
+ app = Starlette(routes=routes)
+ """
+
route = Host(host, app=app, name=name)
self.routes.append(route)
return decorator
- def add_event_handler(self, event_type: str, func: typing.Callable) -> None:
+ def add_event_handler(
+ self, event_type: str, func: typing.Callable
+ ) -> None: # pragma: no cover
assert event_type in ("startup", "shutdown")
if event_type == "startup":
from starlette.applications import Starlette
from starlette.endpoints import HTTPEndpoint
from starlette.exceptions import HTTPException
+from starlette.middleware import Middleware
from starlette.middleware.trustedhost import TrustedHostMiddleware
from starlette.responses import JSONResponse, PlainTextResponse
from starlette.routing import Host, Mount, Route, Router, WebSocketRoute
else:
from contextlib2 import asynccontextmanager # pragma: no cover
-app = Starlette()
-
-app.add_middleware(TrustedHostMiddleware, allowed_hosts=["testserver", "*.example.org"])
-
-
-@app.exception_handler(500)
async def error_500(request, exc):
return JSONResponse({"detail": "Server Error"}, status_code=500)
-@app.exception_handler(405)
async def method_not_allowed(request, exc):
return JSONResponse({"detail": "Custom message"}, status_code=405)
-@app.exception_handler(HTTPException)
async def http_exception(request, exc):
return JSONResponse({"detail": exc.detail}, status_code=exc.status_code)
-@app.route("/func")
def func_homepage(request):
return PlainTextResponse("Hello, world!")
-@app.route("/async")
async def async_homepage(request):
return PlainTextResponse("Hello, world!")
-@app.route("/class")
class Homepage(HTTPEndpoint):
def get(self, request):
return PlainTextResponse("Hello, world!")
-users = Router()
-
-
-@users.route("/")
def all_users_page(request):
return PlainTextResponse("Hello, everyone!")
-@users.route("/{username}")
def user_page(request):
username = request.path_params["username"]
return PlainTextResponse(f"Hello, {username}!")
-app.mount("/users", users)
-
-
-subdomain = Router()
-
-
-@subdomain.route("/")
def custom_subdomain(request):
return PlainTextResponse("Subdomain: " + request.path_params["subdomain"])
-app.host("{subdomain}.example.org", subdomain)
-
-
-@app.route("/500")
def runtime_error(request):
raise RuntimeError()
-@app.websocket_route("/ws")
async def websocket_endpoint(session):
await session.accept()
await session.send_text("Hello, world!")
await session.close()
+users = Router(
+ routes=[
+ Route("/", endpoint=all_users_page),
+ Route("/{username}", endpoint=user_page),
+ ]
+)
+
+subdomain = Router(
+ routes=[
+ Route("/", custom_subdomain),
+ ]
+)
+
+exception_handlers = {
+ 500: error_500,
+ 405: method_not_allowed,
+ HTTPException: http_exception,
+}
+
+middleware = [
+ Middleware(TrustedHostMiddleware, allowed_hosts=["testserver", "*.example.org"])
+]
+
+app = Starlette(
+ routes=[
+ Route("/func", endpoint=func_homepage),
+ Route("/async", endpoint=async_homepage),
+ Route("/class", endpoint=Homepage),
+ Route("/500", endpoint=runtime_error),
+ WebSocketRoute("/ws", endpoint=websocket_endpoint),
+ Mount("/users", app=users),
+ Host("{subdomain}.example.org", app=subdomain),
+ ],
+ exception_handlers=exception_handlers,
+ middleware=middleware,
+)
+
+
@pytest.fixture
def client(test_client_factory):
with test_client_factory(app) as client:
Route("/func", endpoint=func_homepage, methods=["GET"]),
Route("/async", endpoint=async_homepage, methods=["GET"]),
Route("/class", endpoint=Homepage),
+ Route("/500", endpoint=runtime_error, methods=["GET"]),
+ WebSocketRoute("/ws", endpoint=websocket_endpoint),
Mount(
"/users",
app=Router(
"{subdomain}.example.org",
app=Router(routes=[Route("/", endpoint=custom_subdomain)]),
),
- Route("/500", endpoint=runtime_error, methods=["GET"]),
- WebSocketRoute("/ws", endpoint=websocket_endpoint),
]
with open(path, "w") as file:
file.write("<file content>")
- app = Starlette()
- app.mount("/static", StaticFiles(directory=tmpdir))
+ app = Starlette(
+ routes=[
+ Mount("/static", StaticFiles(directory=tmpdir)),
+ ]
+ )
client = test_client_factory(app)
def test_app_debug(test_client_factory):
- app = Starlette()
- app.debug = True
-
- @app.route("/")
async def homepage(request):
raise RuntimeError()
+ app = Starlette(
+ routes=[
+ Route("/", homepage),
+ ],
+ )
+ app.debug = True
+
client = test_client_factory(app, raise_server_exceptions=False)
response = client.get("/")
assert response.status_code == 500
def test_app_add_route(test_client_factory):
- app = Starlette()
-
async def homepage(request):
return PlainTextResponse("Hello, World!")
- app.add_route("/", homepage)
+ app = Starlette(
+ routes=[
+ Route("/", endpoint=homepage),
+ ]
+ )
+
client = test_client_factory(app)
response = client.get("/")
assert response.status_code == 200
def test_app_add_websocket_route(test_client_factory):
- app = Starlette()
-
async def websocket_endpoint(session):
await session.accept()
await session.send_text("Hello, world!")
await session.close()
- app.add_websocket_route("/ws", websocket_endpoint)
+ app = Starlette(
+ routes=[
+ WebSocketRoute("/ws", endpoint=websocket_endpoint),
+ ]
+ )
client = test_client_factory(app)
with client.websocket_connect("/ws") as session:
def test_app_add_event_handler(test_client_factory):
startup_complete = False
cleanup_complete = False
- app = Starlette()
def run_startup():
nonlocal startup_complete
nonlocal cleanup_complete
cleanup_complete = True
- app.add_event_handler("startup", run_startup)
- app.add_event_handler("shutdown", run_cleanup)
+ app = Starlette(
+ on_startup=[run_startup],
+ on_shutdown=[run_cleanup],
+ )
assert not startup_complete
assert not cleanup_complete
projects / thirdparty / starlette.git / commitdiff
