Partage de technologie

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

Ce que nous avons partagé auparavant était le fonctionnement des tables de base de données basées sur une seule table. Lors de la conception de la base de données, nous avons également conçu des opérations entre tables. Nous pouvons jeter un œil à la conception de la base de données.

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

Ensuite, nous examinerons directement le code complet

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

Le facteur demande de créer un article

Interroger tous les éléments

Interroger les éléments d'un utilisateur spécifié

Nous avons implémenté une simple requête multi-tables.