当前位置：首页 > news >正文

FastAPI 学习之路（三十四）数据库多表操作

news 来源：原创 2024/9/19 10:13:59

之前我们分享的是基于单个表的数据库表的操作，我们在设计数据库的时候也设计了跨表，我们可以看下数据库的设计

class 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")

接下来我们就直接看完整的代码

from typing import Optional, List"""
数据库多表操作
"""from sqlalchemy import create_engine, ForeignKey
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import Column, Integer, String, Booleanconn = "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)# 该类的每个实例都是一个数据库会话，该类本身还不是数据库会话，但是一旦我们创建了SessionLocal的实例，这个实例将是实际的数据库会话
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)# 创建数据库基类
Base = declarative_base()# 创建两个数据库模型，继承自Base基类
class 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")from pydantic import BaseModel
from sqlalchemy.orm import Session
from fastapi import FastAPI, Depends, HTTPExceptiondef create_db():"""每个请求处理完毕后关闭当前连接，不同的请求使用不同的链接"""db = SessionLocal()try:yield dbfinally:db.close()app = FastAPI()# -------------------------以下是多表操作
class BaseItem(BaseModel):title: strdescription: Optional[str] = Noneclass ItemModel(BaseItem):passclass ItemOut(BaseItem):id: intowner_id: intclass Config:orm_mode = True@app.post("/items/{uid}", response_model=ItemOut)
def create_item_by_user(uid: int, item: ItemModel, db: Session = Depends(create_db)):init_item = Item(**item.dict(), owner_id=uid)db.add(init_item)db.commit()db.refresh(init_item)return init_item# 分页获取所有的items
@app.get("/items", response_model=List[ItemOut])
def get_items(skip: int = 0, limit: int = 10, db: Session = Depends(create_db)):return db.query(Item).offset(skip).limit(limit).all()# 如何查询用户的items
@app.get("/items/{uid}", response_model=List[ItemOut])
def get_items_by_uid(uid: int, db: Session = Depends(create_db)):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()

postman请求创建item