Technologieaustausch

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

Was wir zuvor geteilt haben, war der Betrieb von Datenbanktabellen basierend auf einer einzelnen Tabelle. Beim Entwurf der Datenbank können wir uns auch den Entwurf der Datenbank ansehen.

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

Als nächstes schauen wir uns direkt den vollständigen Code an

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

Der Postbote bittet darum, einen Artikel zu erstellen

Alle Artikel abfragen

Fragen Sie die Elemente eines bestimmten Benutzers ab

Wir haben eine einfache Multi-Table-Abfrage implementiert.