Return a session factory bound to the given engine. Usage:: factory = get_session_factory(engine) async with factory() as session: await session.execute(...)
(engine: AsyncEngine)
| 60 | |
| 61 | |
| 62 | def get_session_factory(engine: AsyncEngine) -> type[AsyncSession]: |
| 63 | """ |
| 64 | Return a session factory bound to the given engine. |
| 65 | |
| 66 | Usage:: |
| 67 | |
| 68 | factory = get_session_factory(engine) |
| 69 | async with factory() as session: |
| 70 | await session.execute(...) |
| 71 | """ |
| 72 | from sqlalchemy.orm import sessionmaker |
| 73 | return sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) |
| 74 | |
| 75 | |
| 76 | async def init_db(engine: AsyncEngine | None = None) -> None: |
nothing calls this directly
no outgoing calls
no test coverage detected