fastapi如何连接postgre数据库

导包:

from sqlalchemy import DateTime, func, select
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

创建异步引擎

定义变量

ASYNC_DATABASE_URL= "postgresql+asyncpg://postgres:[数据库密码]@[数据库地址]/[用户名]"

创建异步引擎

async_engine = create_async_engine(
    ASYNC_DATABASE_URL,
    echo = True
    pool_size = 10 
    max_overflow = 20 
)

定义模型基类,封装通用内容

class Base(DeclarativeBase):
    create_time:Mapped[datetime]=mapped_column(DateTime,default=datetime.now,comment="创建时间")
    update_time:Mapped[datetime]=mapped_column(DateTime,default=datetime.now,onupdate=datetime.now,comment="更新时间")

定义模型类

class model_base(Base):
    __tablename__ = "表名"
    id:Mapped[int]=mapped_column(primary_key=True,autoincrement=True)
    title:Mapped[str]=mapped_column(nullable=True)

建表

async def create_table():
    async with async_engine.begin() as conn:
        await conn.run_async(Base.matedata.create_all)

启动项目时连接数据库

@asynccontextmanager
async def lifespan(app: FastAPI):
    await create_table()
    yield

app = FastAPI(lifespan=lifespan)

命令行关闭进程

有时候会遇到项目没有正常关闭导致无法正常测试
这时候需要执行`netstata -ano | findstr :8000`查看是否是因为项目没有关闭导致的
接着执行`taskkill /F /IM python.exe`关闭项目