fast-api连mysql

### 使用 FastAPI 连接 MySQL 的方法 为了实现 FastAPI 应用程序与 MySQL 数据库之间的连接，通常会借助 SQLAlchemy 和 PyMySQL 来完成这一过程[^2]。 #### 安装依赖包 首先需要安装必要的 Python 包来支持数据库操作： ```bash pip install fastapi uvicorn sqlalchemy pymysql alembic ``` 这些工具提供了创建 API、定义 ORM 模型以及执行迁移所需的功能。 #### 创建项目结构 建立合理的文件夹布局有助于管理应用程序的不同部分。一个典型的项目可能看起来像这样： ``` my_project/ ├── app/ │ ├── main.py │ └── models.py └── migrations/ # Alembic migration scripts will be stored here. ``` `main.py` 文件用于启动 FastAPI 实例并配置路由；而 `models.py` 则用来声明数据表对应的类模型。 #### 设置数据库引擎和会话工厂 在 `app/models.py` 中初始化 SQLAlchemy 引擎，并设置好相应的 URL 字符串指向目标 MySQL 数据源： ```python from sqlalchemy import create_engine, Column, Integer, String from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker DATABASE_URL = "mysql+pymysql://username:password@localhost/dbname" engine = create_engine(DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base() ``` 这里通过指定驱动名称 (`pymysql`) 及其他必要参数构建完整的 URI 形式字符串给到 `create_engine()` 函数调用。 #### 定义数据模型 继续在同一文件内编写表示持久化实体的对象关系映射 (ORM) 类： ```python class User(Base): __tablename__ = 'users' id = Column(Integer, primary_key=True, index=True) name = Column(String(50)) email = Column(String(100), unique=True) ``` 这一步骤明确了应用所涉及的数据表格及其字段属性。 #### 构建 CRUD 接口逻辑 最后，在 `app/main.py` 添加路径处理器函数以响应 HTTP 请求并与上述定义好的模式交互： ```python from fastapi import Depends, FastAPI, HTTPException from sqlalchemy.orm import Session from .models import Base, engine, SessionLocal, User app = FastAPI() # Dependency injection for getting DB sessions easily within route handlers. def get_db(): db = SessionLocal() try: yield db finally: db.close() @app.post("/users/", response_model=UserSchema) async def create_user(user_data: UserCreate, db: Session = Depends(get_db)): new_user = User(**user_data.dict()) db.add(new_user) db.commit() db.refresh(new_user) return new_user @app.get("/users/{user_id}", response_model=UserRead) async def read_user(user_id: int, db: Session = Depends(get_db)): user = db.query(User).filter(User.id == user_id).first() if not user: raise HTTPException(status_code=404, detail="User not found") return user ``` 此代码片段展示了如何利用依赖注入机制简化获取数据库会话的过程，并实现了两个简单的 RESTful 路由——分别对应于新增记录和查询特定 ID 用户的操作。