Вот асинхронный генератор get_db():
Код: Выделить всё
async def get_db() -> AsyncGenerator[AsyncSession, None]:
async with async_session() as session:
yield session
Код: Выделить всё
@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
]
Код: Выделить всё
@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)
Подробнее здесь: https://stackoverflow.com/questions/778 ... -fastapi-r