Compartir tecnología

한어Русский языкEnglishFrançaisIndonesianSanskrit日本語DeutschPortuguêsΕλληνικάespañolItalianoSuomalainenLatina

Lo que compartimos antes fue el funcionamiento de las tablas de la base de datos basadas en una sola tabla. Al diseñar la base de datos, también diseñamos operaciones entre tablas. Podemos echar un vistazo al diseño de la base de datos.

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

A continuación veremos directamente el código completo.

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, Boolean
 
conn = "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, HTTPException
 
 
def create_db():
    """每个请求处理完毕后关闭当前连接，不同的请求使用不同的链接"""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
 
 
app = FastAPI()
 
 
# -------------------------以下是多表操作
class BaseItem(BaseModel):
    title: str
    description: Optional[str] = None
 
 
class ItemModel(BaseItem):
    pass
 
 
class ItemOut(BaseItem):
    id: int
    owner_id: int
 
    class 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()

El cartero solicita crear un artículo

Consultar todos los artículos

Consultar los elementos de un usuario específico.

Hemos implementado una consulta simple de varias tablas.