Compare commits

...

3 Commits

Author SHA1 Message Date
github-actions[bot]
404aadf1d7 🎨 Auto format 2026-02-09 08:23:37 +00:00
Yurii Motov
267f099ac8 Add test 2026-02-09 09:16:27 +01:00
Yurii Motov
4561d575b7 Fix on_startup and on_shutdown parameters of APIRouter 2026-02-09 09:16:20 +01:00
2 changed files with 71 additions and 10 deletions

View File

@@ -952,16 +952,6 @@ class APIRouter(routing.Router):
),
] = 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
@@ -985,6 +975,17 @@ class APIRouter(routing.Router):
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 [])

View File

@@ -317,3 +317,63 @@ def test_router_async_generator_lifespan(state: State) -> None:
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