Condivisione della tecnologia

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

Ciò che abbiamo condiviso in precedenza era il funzionamento delle tabelle del database basate su una singola tabella. Durante la progettazione del database, abbiamo progettato anche le operazioni tra tabelle. Possiamo dare un'occhiata alla progettazione del database.

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

Successivamente esamineremo direttamente il codice 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()

Il postino richiede di creare un articolo

Interroga tutti gli elementi

Interrogare gli elementi di un utente specificato

Abbiamo implementato una semplice query multitabella.