mirror of
https://github.com/testdrivenio/fastapi-sqlmodel-alembic.git
synced 2025-08-15 01:20:38 +08:00
24 lines
671 B
Python
24 lines
671 B
Python
import os
|
|
|
|
from sqlmodel import SQLModel, create_engine
|
|
from sqlmodel.ext.asyncio.session import AsyncSession, AsyncEngine
|
|
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
|
|
DATABASE_URL = os.environ.get("DATABASE_URL")
|
|
|
|
engine = AsyncEngine(create_engine(DATABASE_URL, echo=True, future=True))
|
|
|
|
async def init_db():
|
|
async with engine.begin() as conn:
|
|
# await conn.run_sync(SQLModel.metadata.drop_all)
|
|
await conn.run_sync(SQLModel.metadata.create_all)
|
|
|
|
|
|
async def get_session() -> AsyncSession:
|
|
async_session = sessionmaker(
|
|
engine, class_=AsyncSession, expire_on_commit=False
|
|
)
|
|
async with async_session() as session:
|
|
yield session |