【FastAPI】Python FastAPI + MySQL 操作流程

本文将使用Python Web框架 FastAPI 实现与关系数据库MySQL的连接与操作。
FastAPI 官方教程

一、连接 MySQL

在database.py中，完成与MySQL的连接

 # database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker


SQLALCHEMY_DATABASE_URL = f'mysql+mysqlconnector://{USER}:{PWD}@localhost:3306/{DB_NAME}?charset=utf8&auth_plugin=mysql_native_password'
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, pool_pre_ping=True
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

二、设计数据库表

使用ORM在models.py中对数据库表的model进行设计。以user表为例：

 # models.py
 
from sqlalchemy import Column, String

from database import Base, engine


class User(Base):
    __tablename__ = 'user'  # 数据库表名

    account = Column(String(255), primary_key=True, index=True)
    screen_name = Column(String(255), nullable=False)
    hash_pwd = Column(String(255), nullable=False)


if __name__ == '__main__':
    Base.metadata.create_all(engine)

三、（可选）定义与数据库表相对应的 Pydantic models

建立该schema的原因可详见官方教程，大致就是在普通数据与ORM Model中间进行数据的处理与格式化的。

 # schemas.py
 
from pydantic import BaseModel

class User(BaseModel):
    account: str
    screen_name: str
    hash_pwd: str

四、使用DB Session对数据库进行操作

db session 就是对之前定义的 SessionLocal 的实例化。

 # main.py
 
from fastapi import FastAPI, Depends
import hashlib
from sqlalchemy.orm import Session

from database import SessionLocal
import models
import schemas


app = FastAPI()

# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post('/sign_in')
def create_user(user: schemas.User, db: Session = Depends(get_db)):
    db_user = models.User()
    db_user.hash_pwd = hashlib.new('md5', user.pwd.encode()).hexdigest()
    db_user.account, db_user.screen_name = user.account, user.screen_name
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return {
        'error': 0,
        'data': 'success'
    }

五、部署 FastAPI

 # startup.py
 
import uvicorn
from main import app

if __name__ == '__main__':
    uvicorn.run(app, host="127.0.0.1", port=8895)