安装第三方库sqlalchemy让你用 Python 类和对象来操作数据库表和数据把 SQL 语句翻译成 Python 代码。加[asyncio]确保greenlet被安装让你能用create_async_engineaiomysql负责真正的数据库网络通信是异步引擎的底层驱动pip install sqlalchemy[asyncio] aiomysqlfrom fastapi import FastAPI,Path,Query,HTTPException,Depends from sqlalchemy.ext.asyncio import create_async_engine,async_sessionmaker,AsyncSession from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, MappedColumn from sqlalchemy import func,DateTime,String,Float,select # 数据库URL root 用户名 admin数据库密码 python_ceshi数据库名 ASYNC_DATABASE_URL mysqlaiomysql://root:adminlocalhost:3306/python_ceshi?charsetutf8 # 创建异步引擎 async_engine create_async_engine( ASYNC_DATABASE_URL, echoTrue, pool_size10, max_overflow10, ) # 创建异步会话工厂 AsyncSessionLocal async_sessionmaker( bind async_engine, class_ AsyncSession, expire_on_commit False, ) # 创建基类,所有的模型都要继承基类,为所有的模型设置创建时间和修改时间字段 class Base(DeclarativeBase): create_time: Mapped[datetime] mapped_column(DateTime,insert_defaultfunc.now(),defaultfunc.now(),comment创建时间) update_time: Mapped[datetime] mapped_column(DateTime,insert_defaultfunc.now(),onupdatefunc.now(),comment修改时间) # 定义模型类继承 Base class User(Base): # ← 继承 Base __tablename__ users # ← 指定表名 # 写n个字段名查询的时候才能查到n个 id Column(Integer, primary_keyTrue) name Column(String(50)) age Column(Integer) #创建表方法 async def create_table(): async with async_engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) # 依赖项 用户获取数据库会话 async def get_db(): async with AsyncSessionLocal() as session: try: yield session await session.commit() except Exception as e: await session.rollback() raise e finally: await session.close() #启动阶段运行 uvicorn main:app --reload 命令时FastAPI 会立刻调用这个函数。 asynccontextmanager async def lifespan_context(app: FastAPI): # 进入时执行启动阶段 await create_table() yield # 必须标记进入阶段结束运行阶段开始 # 退出时执行关闭阶段 app FastAPI(lifespanlifespan_context) app.get(/) # Depends(get_db) 是 FastAPI 的依赖注入系统 #它的作用是在每个请求中自动获取一个数据库会话AsyncSession并把它注入到你的路由函数里。 async def index(db: AsyncSession Depends(get_db)): # 获取全部的用户信息 return (await db.scalars(select(User))).all()
