Skip to content

Commit

Permalink
Allow passing FastAPI directly to init
Browse files Browse the repository at this point in the history
This is useful in cases where the application is created in
a factory or otherwise doesn't have a string directly tied
to the app.
  • Loading branch information
bjchambers committed Mar 6, 2024
1 parent 4cbe2b6 commit 8264509
Showing 1 changed file with 10 additions and 6 deletions.
16 changes: 10 additions & 6 deletions taskiq_fastapi/initializator.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
from typing import Awaitable, Callable
from typing import Awaitable, Callable, Union

from fastapi import FastAPI, Request
from starlette.requests import HTTPConnection
Expand All @@ -8,7 +8,7 @@

def startup_event_generator(
broker: AsyncBroker,
app_path: str,
app_or_path: Union[str, FastAPI],
) -> Callable[[TaskiqState], Awaitable[None]]:
"""
Generate shutdown event.
Expand All @@ -24,12 +24,16 @@ def startup_event_generator(
async def startup(state: TaskiqState) -> None:
if not broker.is_worker_process:
return
app = import_object(app_path)
if not isinstance(app, FastAPI):
app = app()
app = None
if isinstance(app_or_path, FastAPI):
app = app_or_path
else:
app = import_object(app_or_path)
if not isinstance(app, FastAPI):
app = app()

if not isinstance(app, FastAPI):
raise ValueError(f"'{app_path}' is not a FastAPI application.")
raise ValueError(f"'{app_or_path}' is not a FastAPI application.")

state.fastapi_app = app
await app.router.startup()
Expand Down

0 comments on commit 8264509

Please sign in to comment.