Skip to content

Commit

Permalink
Merge pull request #52 from community-of-python/46-use-lifespan-inste…
Browse files Browse the repository at this point in the history
…ad-of-on_shutdown-and-on_startup

Use lifespan instead of on_shutdown and on_startup for fastapi
  • Loading branch information
lesnik512 authored Dec 9, 2024
2 parents a667f89 + 5ac2300 commit 149a716
Show file tree
Hide file tree
Showing 2 changed files with 26 additions and 9 deletions.
26 changes: 24 additions & 2 deletions microbootstrap/bootstrappers/fastapi.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import contextlib
import typing

import fastapi
Expand Down Expand Up @@ -29,11 +30,32 @@ class FastApiBootstrapper(
application_config = FastApiConfig()
application_type = fastapi.FastAPI

@contextlib.asynccontextmanager
async def _lifespan_manager(self, _: fastapi.FastAPI) -> typing.AsyncIterator[dict[str, typing.Any]]:
try:
self.console_writer.print_bootstrap_table()
yield {}
finally:
self.teardown()

@contextlib.asynccontextmanager
async def _wrapped_lifespan_manager(self, app: fastapi.FastAPI) -> typing.AsyncIterator[dict[str, typing.Any]]:
assert self.application_config.lifespan # noqa: S101
async with self._lifespan_manager(app), self.application_config.lifespan(app):
yield {}

def _choose_lifespan_manager(
self,
) -> typing.Callable[[fastapi.FastAPI], typing.AsyncContextManager[dict[str, typing.Any]]]:
if self.application_config.lifespan:
return self._wrapped_lifespan_manager

return self._lifespan_manager

def bootstrap_before(self) -> dict[str, typing.Any]:
return {
"debug": self.settings.service_debug,
"on_shutdown": [self.teardown],
"on_startup": [self.console_writer.print_bootstrap_table],
"lifespan": self._choose_lifespan_manager(),
}


Expand Down
9 changes: 2 additions & 7 deletions tests/bootstrappers/test_fastapi.py
Original file line number Diff line number Diff line change
Expand Up @@ -49,14 +49,9 @@ def test_fastapi_configure_application() -> None:
assert application.title == test_title


def test_fastapi_configure_application_add_startup_event(magic_mock: MagicMock) -> None:
def test_startup() -> None:
magic_mock()

def test_fastapi_configure_application_lifespan(magic_mock: MagicMock) -> None:
application: typing.Final = (
FastApiBootstrapper(FastApiSettings())
.configure_application(FastApiConfig(on_startup=[test_startup]))
.bootstrap()
FastApiBootstrapper(FastApiSettings()).configure_application(FastApiConfig(lifespan=magic_mock)).bootstrap()
)

with TestClient(app=application):
Expand Down

0 comments on commit 149a716

Please sign in to comment.