Fastapi连接Mysql数据库

文件目录

• run.py
• app- database.py- main.py- model.py- schemas.py- crud.py

步骤 1: 准备工作

首先，确保你的 Python 环境中已经安装了 FastAPI、uvicorn 、sqlalchemy和pymysql。

pip install fastapi
pip ntsall uvicorn
pip install sqlalchemy
pip install pymysql

步骤 2: 创建 database.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 设置数据库的URL
SQLALCHEMY_DATABASE_URL = "mysql+pymysql://用户名:密码@主机地址:端口号/数据库名"

# 创建数据库引擎
engine = create_engine(
    SQLALCHEMY_DATABASE_URL,    # 数据库连接地址
    # echo=True,                  # 打印SQL语句
    # connect_args={"check_same_thread": False}    # 解决多线程问题  sqlite使用
)

# 创建会话
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine, expire_on_commit=True)

# 映射
Base = declarative_base()

步骤 3: 创建model.py

from sqlalchemy import Column, Integer, String
from .DateBase_ms import Base

# 定义User类，继承自Base
class User(Base):
    # 定义表名
    __tablename__ = "user"
    # 定义id字段，主键，自增，索引，可以为空
    id = Column(Integer, primary_key=True, index=True, nullable=True, autoincrement=True)
    # 定义name字段，不能为空，长度为100
    name = Column(String(100), nullable=False, comment="姓名")
    # 定义gender字段，长度为10
    gender = Column(String(10), comment="性别")
    

步骤 4: 创建schemas.py

from pydantic import BaseModel
from typing import Union, Optional

# 定义UserBase类，继承自BaseModel
class UserBase(BaseModel):
    # 定义id属性，类型为int，默认值为1
    id: int = 1

    # 定义Config属性，用于配置
    class Config:
        # 设置orm_mode属性，值为True
        orm_mode = True

# 创建用户类
class create_user(UserBase):
    # 用户名
    name: str
    # 性别
    gender: Optional[str] = None
    

    # 定义一个Config类
    class Config:
        # 定义一个orm_mode属性，默认值为True
        orm_mode = True

class read_user(UserBase):
    # 用户名
    name: str
    # 性别
    gender: Optional[str] = None

    # 路径操作函数中直接返回SQLAlchemy数据模型
    class Config:
        orm_mode = True

步骤 5: 创建crud.py

from sqlalchemy.orm import Session
import model, schemas

# 查看用户信息
def get_user(db: Session, skip: int = 0, limit: int = 10):
    # 返回skip和limit之间的用户信息
    return db.query(model.User).offset(skip).limit(limit).all()

# 创建新用户
def create_user(db: Session, user: schemas.create_user):
    # 创建一个User模型，并将加密后的密码赋值给password
    db_user = model.User(id = user.id,
                            name = user.name,
                            gender = user.gender
                            )
    # 将User模型添加到数据库
    db.add(db_user)
    # 提交数据库操作
    db.commit()
    # 刷新数据库
    db.refresh(db_user)
    # 返回创建的用户
    return db_user

步骤6：创建main.py

from fastapi import Depends, APIRouter, HTTPException
from sqlalchemy.orm import Session
import crud, model, schemas
from DateBase import SessionLocal, engine

# 创建数据表
model.Base.metadata.create_all(bind=engine)
# 创建所有模型
app = APIRouter()

# 定义一个函数get_db，用于获取数据库连接
def get_db():
    # 创建一个SessionLocal对象
    db = SessionLocal()
    try:
        # 尝试执行yield db
        yield db
    finally:
        # 关闭数据库连接
        db.close()

# 获取user数据
@app_ms.get("/read/", response_model = List[schemas.read_user])
def read_data(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    # 从数据库中获取用户数据
    users = crud.get_user(db, skip=skip, limit=limit)
    # 返回用户数据
    return users

# 录入user
@app_ms.post("/create/", response_model = schemas.read_user)
def create_data(user: schemas.create_user, db: Session = Depends(get_db)):
    # 检查数据库中是否存在该用户的账号
    db_user = crud.get_user_by_account(db, account = user.account)
    # 如果存在，则抛出异常
    if db_user:
        raise HTTPException(status_code=400, detail="user already registered")
    # 检查数据库中是否存在该用户的手机号
    db_user = crud.get_user_by_phone_number(db, phone_number = user.phone_number)
    # 如果存在，则抛出异常
    if db_user:
        raise HTTPException(status_code=400, detail="phone number already used")
    # 返回创建的用户
    return crud.create_user(db=db, user=user)

步骤7：创建run.py

from fastapi import FastAPI
import uvicorn
from app import app_ms

# 创建FastAPI应用
app = FastAPI()

# 包含app_ms路由，前缀为/app_ms，标签为api_ms
app.include_router(app_ms, prefix="/app_ms", tags=["api_ms"])

if __name__ == "__main__":
    # 启动uvicorn服务，参数为demo_run:app，主机地址为127.0.0.1，端口号为8000，自动重新加载，工作进程数为1
    uvicorn.run('demo_run:app', host="127.0.0.1", port=8000, reload=True, workers=1)

运行run.py

    开启数据库后，浏览器输入127.0.0.1:8000/docs，即可对其进行测试。