Kann mir jemand sagen, wie ich die Abhängigkeitsinjektion für meine get_db() außerhalb der FastAPI-Router-Methoden verwenden kann? Anscheinend deckt Depends() nur DI in Anforderungsfunktionen ab.
Hier ist der asynchrone Generator get_db():
async def get_db() -> AsyncGenerator[AsyncSession, None]:
async with async_session() as session:
yield session
Im FastAPI-Router kann ich Depends() einfach so verwenden:
@router.get("/interactions", response_model=List[schemas.Interaction])
async def get_all_interactions(db: Annotated[AsyncSession, Depends(get_db)]) -> List[schemas.Interaction]:
interactions = await crud.get_interactions(db=db)
return [
schemas.Interaction.model_validate(interaction) for interaction in interactions
]
Wie kann ich nun außerhalb der Anfrage die get_db in eine neue Methode einfügen und async for innerhalb der Methode entfernen?
@cli.command(name="create_superuser")
async def create_superuser(): # Note: how to pass db session here as param?
username = click.prompt("Username", type=str)
email = click.prompt("Email (optional)", type=str, default="")
password = getpass("Password: ")
confirm_password = getpass("Confirm Password: ")
if password != confirm_password:
click.echo("Passwords do not match")
return
async for db in database.get_db(): # Note: remove it from here
user = schemas.UserAdminCreate(
username=username,
email=None if not email else email,
password=password,
role="admin",
)
await crud.create_user(db=db, user=user)
PS: Der Grund für diese Anforderung ist, dass ich einen Testfall für die Funktion create_superuser() schreiben werde. das über eine eigene Datenbank und entsprechende Sitzung verfügt, daher wäre es für mich von Vorteil, die Sitzungsdatenbank in alle Methoden einzufügen.
Wie verwende ich Depends oder ähnliches als Abhängigkeitsinjektion außerhalb der FastAPI-Anfragemethoden? ⇐ Python
-
- Similar Topics
- Replies
- Views
- Last post