之前我們創(chuàng)建的文件都是在一個(gè)目錄中，但是在我們的實(shí)際開(kāi)發(fā)中，肯定不能這樣設(shè)計(jì)，那么我們?nèi)?chuàng)建一個(gè)目錄，叫models，大致如下。

主要目錄是：

· __init__.py 是一個(gè)空文件，說(shuō)明models是一個(gè)package

· crud.py 數(shù)據(jù)庫(kù)操作相關(guān)

· database.py 數(shù)據(jù)庫(kù)配置相關(guān)

· models.py 數(shù)據(jù)庫(kù)模型表

· schemas.py 模型驗(yàn)證

· main.py 主文件

database.py代碼如下：

from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import sessionmakerconn = "mysql+pymysql://{username}:{password}@{host}:{port}/{database}?charset=utf8".format(username="root", password="123456", host="10.30.10.36", port=3306, database="fastapi_learn_road")
engine = create_engine(conn)# 該類的每個(gè)實(shí)例都是一個(gè)數(shù)據(jù)庫(kù)會(huì)話，該類本身還不是數(shù)據(jù)庫(kù)會(huì)話，但是一旦我們創(chuàng)建了SessionLocal的實(shí)例，這個(gè)實(shí)例將是實(shí)際的數(shù)據(jù)庫(kù)會(huì)話
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)# 創(chuàng)建數(shù)據(jù)庫(kù)基類
Base = declarative_base()

models.py代碼如下：

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from .database import Baseclass User(Base):__tablename__ = "users"id = Column(Integer, primary_key=True, index=True)email = Column(String(10), unique=True, index=True)hashed_password = Column(String(100))is_active = Column(Boolean, default=True)items = relationship("Item", back_populates="owner")class Item(Base):__tablename__ = "items"id = Column(Integer, primary_key=True, index=True)title = Column(String(10), index=True)description = Column(String(10), index=True)owner_id = Column(Integer, ForeignKey("users.id"))owner = relationship("User", back_populates="items")

schemas.py代碼如下：定義請(qǐng)求參數(shù)模型驗(yàn)證與響應(yīng)模型驗(yàn)證的Pydantic模型。

from pydantic import BaseModel
from typing import List, Optionalclass BaseItem(BaseModel):title: strdescription: Optional[str] = Noneclass ItemModel(BaseItem):passclass ItemOut(BaseItem):id: intowner_id: intclass Config:orm_mode = Trueclass BaseUser(BaseModel):email: strclass UserModel(BaseUser):"""請(qǐng)求參數(shù)模型"""password: strclass UserOut(BaseUser):"""響應(yīng)模型"""id: intis_active: boolitems: List[ItemOut]class Comfig:orm_mode = True

crud.py代碼如下：

# 之前都是把所有邏輯寫(xiě)到了接口函數(shù)里，其實(shí)我們應(yīng)該抽出來(lái)，一起管理
from fastapi import HTTPException
from sqlalchemy.orm import Session
from .models import *
from .schemas import *def get_user_method(db: Session, uid: int):user = db.query(User).filter(User.id == uid).first()if not user:raise HTTPException(status_code=404, detail="user not exists")return userdef create_user_method(db: Session, user: UserModel):db_user = db.query(User).filter(User.email == user.email).first()if db_user:raise HTTPException(status_code=200, detail="this user already exists")fake_hashed_password = user.password + "_hashed"init_user = User(email=user.email, hashed_password=fake_hashed_password)db.add(init_user)db.commit()db.refresh(init_user)return init_userdef get_items_method(db: Session, skip: int = 0, limit: int = 10):return db.query(Item).offset(skip).limit(limit).all()def get_items_by_uid_method(db: Session, uid: int):user = db.query(User).filter(User.id == uid).first()if not user:raise HTTPException(status_code=200, detail="this user is not valid")return db.query(Item).filter(Item.owner == user).offset(0).limit(2).all()def create_item_by_user_method(db: Session, uid: int, item: ItemModel):init_item = Item(**item.dict(), owner_id=uid)db.add(init_item)db.commit()db.refresh(init_item)return init_item

main.py代碼如下：

from fastapi import FastAPI, Depends, HTTPException
from models.crud import *
from models.database import *app = FastAPI()def create_db():"""每個(gè)請(qǐng)求處理完畢后關(guān)閉當(dāng)前連接，不同的請(qǐng)求使用不同的鏈接"""db = SessionLocal()try:yield dbfinally:db.close()@app.post("/user", response_model=UserOut)
def create_user(user: UserModel, db: Session = Depends(create_db)):return create_user_method(db, user)@app.get("/user", response_model=UserOut)
def get_user(uid: int, db: Session = Depends(create_db)):return get_user_method(db, uid)@app.post("/items/{uid}", response_model=ItemOut)
def create_item_by_user(uid: int, item: ItemModel, db: Session = Depends(create_db)):return create_item_by_user_method(db, uid, item)@app.get("/items", response_model=List[ItemOut])
def get_items(skip: int = 0, limit: int = 10, db: Session = Depends(create_db)):return get_items_method(db, skip, limit)@app.get("/items/{uid}", response_model=List[ItemOut])
def get_items_by_uid(uid: int, db: Session = Depends(create_db)):return get_items_by_uid_method(db, uid)if __name__ == '__main__':import uvicornuvicorn.run("main:app", reload=True)

我們目前是這么改造的。后續(xù)還會(huì)持續(xù)改造的。目前我們沒(méi)有對(duì)API接口main文件進(jìn)行改造，下面的分享我們會(huì)對(duì)api接口做改造。