### FastAPI 中使用 SQLAlchemy 2.0 和 Asyncmy 结合 MySQL 实现 CRUD 操作 在现代 Web 开发中，FastAPI 提供了一种高效的方式来构建 API 接口，而 SQLAlchemy 作为 ORM 工具则简化了数据库交互过程。结合 Asyncmy 支持异步查询功能，可以显著提高性能。以下是具体实现方法： --- #### 1. 环境配置 安装必要的依赖项： ```bash pip install fastapi sqlalchemy asyncmy pymysql uvicorn ``` --- #### 2. 数据库连接设置 创建一个异步会话工厂以管理数据库连接池。 ```python from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.declarative import declarative_base DATABASE_URL = "mysql+asyncmy://username:password@localhost/db_name" engine = create_async_engine(DATABASE_URL, echo=True) Base = declarative_base() # 创建异步会话类 async_session = sessionmaker( bind=engine, class_=AsyncSession, expire_on_commit=False, autocommit=False, autoflush=False, ) ``` --- #### 3. 定义模型 使用 SQLAlchemy 的声明式基类定义数据表结构。 ```python from sqlalchemy import Column, Integer, String class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True, index=True) name = Column(String(50), nullable=False) email = Column(String(100), unique=True, nullable=False) ``` --- #### 4. CRUD 操作实现 ##### （1）创建记录 (Create) 通过 `add()` 方法向数据库插入新记录。 ```python async def create_user(db: AsyncSession, user_data: dict): new_user = User(**user_data) db.add(new_user) await db.commit() await db.refresh(new_user) return new_user ``` ##### （2）读取记录 (Read) 利用 `select()` 查询特定条件下的记录。 ```python from sqlalchemy.future import select async def get_user_by_id(db: AsyncSession, user_id: int): query = select(User).where(User.id == user_id) result = await db.execute(query) return result.scalars().first() ``` ##### （3）更新记录 (Update) 先获取目标对象再修改属性值并提交更改。 ```python async def update_user(db: AsyncSession, user_id: int, updated_data: dict): user = await get_user_by_id(db, user_id) if not user: return None for key, value in updated_data.items(): setattr(user, key, value) await db.commit() await db.refresh(user) return user ``` ##### （4）删除记录 (Delete) 找到对应条目后调用 `delete()` 删除之。 ```python async def delete_user(db: AsyncSession, user_id: int): user = await get_user_by_id(db, user_id) if not user: return False await db.delete(user) await db.commit() return True ``` --- #### 5. 整合到 FastAPI 路由中 将上述业务逻辑嵌入至 FastAPI 的路径操作函数里。 ```python from fastapi import Depends, FastAPI, HTTPException from pydantic import BaseModel from typing import Optional app = FastAPI() # Pydantic 请求体模型 class UserCreateSchema(BaseModel): name: str email: str class UserUpdateSchema(BaseModel): name: Optional[str] = None email: Optional[str] = None @app.post("/users/", response_model=UserCreateSchema) async def add_new_user(user: UserCreateSchema, db: AsyncSession = Depends(get_db)): created_user = await create_user(db=db, user_data=user.dict()) return created_user @app.get("/users/{user_id}", response_model=dict) async def fetch_single_user(user_id: int, db: AsyncSession = Depends(get_db)): fetched_user = await get_user_by_id(db=db, user_id=user_id) if not fetched_user: raise HTTPException(status_code=404, detail="User Not Found") return {"data": fetched_user} @app.put("/users/{user_id}", response_model=dict) async def modify_existing_user( user_id: int, updates: UserUpdateSchema, db: AsyncSession = Depends(get_db) ): modified_user = await update_user(db=db, user_id=user_id, updated_data=updates.dict(exclude_unset=True)) if not modified_user: raise HTTPException(status_code=404, detail="User Not Found") return {"message": f"Updated {modified_user}"} @app.delete("/users/{user_id}") async def remove_specific_user(user_id: int, db: AsyncSession = Depends(get_db)): deletion_status = await delete_user(db=db, user_id=user_id) if not deletion_status: raise HTTPException(status_code=404, detail="User Not Found") return {"message": "Deleted Successfully"} ``` 其中 `get_db` 函数负责提供每次请求所需的独立数据库会话实例。 ```python async def get_db(): async with async_session() as session: yield session ``` --- #### 6. 启动服务 最后启动应用程序验证接口是否正常工作。 ```bash uvicorn your_module_name:app --reload ``` ---