),
] = Default(generate_unique_id),
) -> None:
- # Handle on_startup/on_shutdown locally since Starlette removed support
- # Ref: https://github.com/Kludex/starlette/pull/3117
- # TODO: deprecate this once the lifespan (or alternative) interface is improved
- self.on_startup: list[Callable[[], Any]] = (
- [] if on_startup is None else list(on_startup)
- )
- self.on_shutdown: list[Callable[[], Any]] = (
- [] if on_shutdown is None else list(on_shutdown)
- )
-
# Determine the lifespan context to use
if lifespan is None:
# Use the default lifespan that runs on_startup/on_shutdown handlers
assert not prefix.endswith("/"), (
"A path prefix must not end with '/', as the routes will start with '/'"
)
+
+ # Handle on_startup/on_shutdown locally since Starlette removed support
+ # Ref: https://github.com/Kludex/starlette/pull/3117
+ # TODO: deprecate this once the lifespan (or alternative) interface is improved
+ self.on_startup: list[Callable[[], Any]] = (
+ [] if on_startup is None else list(on_startup)
+ )
+ self.on_shutdown: list[Callable[[], Any]] = (
+ [] if on_shutdown is None else list(on_shutdown)
+ )
+
self.prefix = prefix
self.tags: list[Union[str, Enum]] = tags or []
self.dependencies = list(dependencies or [])
assert response.json() == {"message": "Hello World"}
assert state.app_startup is True
assert state.app_shutdown is True
+
+
+def test_startup_shutdown_handlers_as_parameters(state: State) -> None:
+ """Test that startup/shutdown handlers passed as parameters to FastAPI are called correctly."""
+
+ def app_startup() -> None:
+ state.app_startup = True
+
+ def app_shutdown() -> None:
+ state.app_shutdown = True
+
+ app = FastAPI(on_startup=[app_startup], on_shutdown=[app_shutdown])
+
+ @app.get("/")
+ def main() -> dict[str, str]:
+ return {"message": "Hello World"}
+
+ def router_startup() -> None:
+ state.router_startup = True
+
+ def router_shutdown() -> None:
+ state.router_shutdown = True
+
+ router = APIRouter(on_startup=[router_startup], on_shutdown=[router_shutdown])
+
+ def sub_router_startup() -> None:
+ state.sub_router_startup = True
+
+ def sub_router_shutdown() -> None:
+ state.sub_router_shutdown = True
+
+ sub_router = APIRouter(
+ on_startup=[sub_router_startup], on_shutdown=[sub_router_shutdown]
+ )
+
+ router.include_router(sub_router)
+ app.include_router(router)
+
+ assert state.app_startup is False
+ assert state.router_startup is False
+ assert state.sub_router_startup is False
+ assert state.app_shutdown is False
+ assert state.router_shutdown is False
+ assert state.sub_router_shutdown is False
+ with TestClient(app) as client:
+ assert state.app_startup is True
+ assert state.router_startup is True
+ assert state.sub_router_startup is True
+ assert state.app_shutdown is False
+ assert state.router_shutdown is False
+ assert state.sub_router_shutdown is False
+ response = client.get("/")
+ assert response.status_code == 200, response.text
+ assert response.json() == {"message": "Hello World"}
+ assert state.app_startup is True
+ assert state.router_startup is True
+ assert state.sub_router_startup is True
+ assert state.app_shutdown is True
+ assert state.router_shutdown is True
+ assert state.sub_router_shutdown is True