一、安裝三方庫

pip install sqlalchemy
pip install pymysql

二、創(chuàng)建表

為了更好的實現(xiàn)項目代碼管理，可以將數(shù)據(jù)庫操作放到其他文件中進行，而不是全部代碼都堆積到main.py中

1、根據(jù)截圖創(chuàng)建對應文件

• database.py：用于創(chuàng)建可調用的數(shù)據(jù)庫引擎使用
• models.py：用于定義數(shù)據(jù)表模型
• __init__.py: 空文件，表示mysql是一個自定義的庫包，可以在其他文件中直接使用import引用到

database.py內容如下：

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

# 定義數(shù)據(jù)庫連接的url
# 格式: 數(shù)據(jù)庫類型+數(shù)據(jù)庫驅動://用戶名:密碼@Ip:端口/數(shù)據(jù)庫名
# 注意：數(shù)據(jù)庫需要先前置創(chuàng)建好，不會自動創(chuàng)建數(shù)據(jù)庫
SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:123456@127.0.0.1:3306/fastapi_data"

# 使用url創(chuàng)建一個數(shù)據(jù)庫引擎
engine = create_engine(SQLALCHEMY_DATABASE_URL)

# 創(chuàng)建數(shù)據(jù)庫庫會話
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 返回一個base基類，用于繼承
Base = declarative_base()

models.py內容如下：

from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import relationship

from .database import Base


# 數(shù)據(jù)庫模型
class User(Base):
    # 創(chuàng)建一個名為users的表
    __tablename__ = "users"

    # 字段
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(20))

main.py內容如下：

from fastapi import FastAPI
from mysql import models, database

app = FastAPI()


@app.get("/")
def root():
    # 創(chuàng)建數(shù)據(jù)表
    models.Base.metadata.create_all(bind=database.engine)
    return {"create table success"}

然后啟動服務，瀏覽器訪問root節(jié)點

2、驗證數(shù)據(jù)庫是否創(chuàng)建成功

然后使用下述命令查表是否創(chuàng)建成功

# 連接數(shù)據(jù)庫，會車后需要輸入root用戶密碼再回車才能連上數(shù)據(jù)庫
mysql -uroot -p
# 顯示出當前有哪些庫
show databases;
# 使用 fastApiData 庫
use fastapi_data;
# 查看數(shù)據(jù)庫下有哪些表
show tables;
# 查看users表下有哪些字段
show Columns from users;

當然，也可以使用一些可視化的數(shù)據(jù)庫查看工具去查看，省去命令記憶

三、增刪改查

在之前的基礎上對main.py進行如下修改：

from fastapi import FastAPI, Depends
from mysql import models, database

app = FastAPI()


# 創(chuàng)建數(shù)據(jù)庫表
@app.get("/")
def root():
    # 創(chuàng)建數(shù)據(jù)表
    models.Base.metadata.create_all(bind=database.engine)
    return {"create table success"}


# 新增數(shù)據(jù), 這里為了演示用的get請求，一般用post
@app.get('/add/{userName}')
def add_user(userName):
    userInfo = models.User(name=userName)  # 創(chuàng)建新增實例
    db = database.SessionLocal()  # 獲取數(shù)據(jù)庫會話
    db.add(userInfo)  # 新增
    db.commit()  # 提交
    db.refresh(userInfo)  # 刷新數(shù)據(jù)
    return f"數(shù)據(jù)庫新增數(shù)據(jù)成功, 用戶名： {userName}"


# 查詢數(shù)據(jù)
@app.get('/search/{searchParm}')
def search_data(searchParm):
    db = database.SessionLocal()  # 獲取數(shù)據(jù)庫會話
    if searchParm == 'all':
        users = db.query(models.User).all()  # 查看所有用戶
        return users
    else:
        user = db.query(models.User).filter(models.User.id == int(searchParm)).first()  # 使用過濾條件查看單個用戶
        return user


# 修改數(shù)據(jù)，這里為了演示用了get請求，一般用put
@app.get('/modify/{oldName}&{newName}')
def modify_data(oldName, newName):
    db = database.SessionLocal()  # 獲取數(shù)據(jù)庫會話
    existing_user = db.query(models.User).filter(models.User.name == oldName).first()
    existing_user.name = newName
    db.commit()
    db.refresh(existing_user)
    return existing_user


# 刪除數(shù)據(jù)，這邊為了演示用了get請求，一般用delete
@app.get('/delete/{userName}')
def delete_data(userName):
    db = database.SessionLocal()  # 獲取數(shù)據(jù)庫會話
    db.query(models.User).filter(models.User.name == userName).delete()
    db.commit()
    return f"名為： {userName} 的用戶已經(jīng)被刪除"