main.py
from fastapi import FastAPI
import uvicorn
from database.connection import init_models
app = FastAPI()
@app.on_event('startup')
async def init_db() -> None:
await init_models()
if __name__ == '__main__':
uvicorn.run('main:app', host='0.0.0.0', port=8000, reload=True)
连接.py
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from dotenv import load_dotenv
import os
load_dotenv()
DATABASE_URL: str = f'postgresql+asyncpg://{os.getenv("POSTGRES_USER")}:{os.getenv("POSTGRES_PASSWORD")}@localhost:5432/postgres'
SECRET_KEY: str = os.getenv('SECRET_KEY')
engine = create_async_engine(DATABASE_URL, echo=True)
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
Base = declarative_base()
async def init_models() -> None:
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
await conn.run_sync(Base.metadata.create_all)
async def get_session() -> AsyncSession:
async with async_session() as session:
yield session
我得到这样的输出:
INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
INFO: Started reloader process [4529] using StatReload
INFO: Started server process [4534]
INFO: Waiting for application startup.
2023-03-14 13:48:57,835 INFO sqlalchemy.engine.Engine select pg_catalog.version()
2023-03-14 13:48:57,835 INFO sqlalchemy.engine.Engine [raw sql] ()
2023-03-14 13:48:57,836 INFO sqlalchemy.engine.Engine select current_schema()
2023-03-14 13:48:57,836 INFO sqlalchemy.engine.Engine [raw sql] ()
2023-03-14 13:48:57,836 INFO sqlalchemy.engine.Engine show standard_conforming_strings
2023-03-14 13:48:57,836 INFO sqlalchemy.engine.Engine [raw sql] ()
2023-03-14 13:48:57,837 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2023-03-14 13:48:57,837 INFO sqlalchemy.engine.Engine COMMIT
INFO: Application startup complete.
SQLAlchemy 不创建任何表。我尝试使用不异步的普通会话,尝试使用 alembic,但没有任何帮助。模型一切正常,我不明白问题与哪个有关。