在使用 FastAPI 构建 API 时，Python 中的 pydantic 模型（模式）是什么意思

Question

我是 python 的新人，我正在尝试使用 FastAPI 构建一个 API。
它目前正在运行，我连接了 postgres db，我发出了 post/get/ 请求，一切正常，但我不太理解为什么我们要这样定义模式，为什么我们必须创建一个

class 用户基础（基础模型）
class 用户创建（用户基础）
class 用户（用户群）

我将 post 所有文件的源代码，如果你们能帮助我更好地理解这一点，那将对我有很大帮助，因为我有一个明天的作业。

schemas.py

from typing import List
from pydantic import BaseModel

##BOOKING
class BookingBase(BaseModel):
    name:str
    description:str = None

class BookingCreate(BookingBase):
    pass

class Booking(BookingBase):
    id:int
    user_id:int

    class Config:
        orm_mode = True

##USER
class UserBase(BaseModel):
    email: str

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int
    is_active: bool
    bookings: List[Booking] = []

    class Config:
        orm_mode = True

models.py

from .database import Base
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String,DateTime
from sqlalchemy.sql import func
from sqlalchemy.orm import relationship

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True,index=True)
    email = Column(String, unique=True, index= True)
    hashed_password = Column(String)
    is_active = Column(Boolean,default=True)

    bookings = relationship("Booking", back_populates="owner")

class Booking(Base):
    __tablename__ = "bookings"

    id=Column(Integer,primary_key=True,index=True)
    name = Column(String,index=True)
    description = Column(String, index=True)
    created_date = Column(DateTime, server_default=func.now())
    user_id = Column(Integer,ForeignKey("users.id"))

    owner = relationship("User",back_populates="bookings")

crud.py

from . import models,schemas
from sqlalchemy.orm import Session

def get_user(db:Session,user_id:int):
    return db.query(models.User).filter(models.User.id == user_id).first()

def fetch_user_by_email(db:Session,email:str):
    return db.query(models.User).filter(models.User.email == email).first()

def get_all_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()


def get_bookings(db:Session,skip:int=0,limit:int=100):
    return db.query(models.Booking).offset(skip).limit(limit).all()

def create_new_user(db:Session,user:schemas.UserCreate):
    testing_hashed = user.password + "test"
    db_user = models.User(email=user.email,hashed_password=testing_hashed)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user


def create_user_booking(db: Session, booking: schemas.BookingCreate, user_id: int):
    db_item = models.Booking(**booking.dict(), user_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

database.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker


# SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
SQLALCHEMY_DATABASE_URL = "postgresql://postgres:root@localhost/meetingbookerdb"

##Creating the SQLAlchemy ORM engine..>> above we have imported create_engine method from sqlalchemy
##Since we are using Postgres we dont need anything else

create_engine

engine = create_engine(
    SQLALCHEMY_DATABASE_URL
)

#Creating SessionLocal class which will be database session on the request..

SessionLocal = sessionmaker(autocommit=False,autoflush=False,bind=engine)

## Creating the base clase, using the declerative_base() method which returns a class.
## Later we will need this Base Class to create each of the database models

Base = declarative_base()

和main.py

from typing import List

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from .app import crud, models, schemas
from .app.database  import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)


app = FastAPI()


# Dependency
def get_db():
    try:
        db = SessionLocal()
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.fetch_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_new_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_all_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/bookings/", response_model=schemas.Booking)
def create_booking_for_user(
    user_id: int,booking: schemas.BookingCreate, db: Session = Depends(get_db)
):
    return crud.create_user_booking(db=db, booking=booking, user_id=user_id)


@app.get("/bookings/", response_model=List[schemas.Booking])
def read_bookings(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    bookings = crud.get_bookings(db, skip=skip, limit=limit)
    return bookings

问题是，为什么我们必须像那样创建这些模式，好的，我明白了第一个 UserBase 必须用于使用 pydantic 进行验证，但是其他两个呢，有人可以给我一个很好的解释..

谢谢。

Answer 1

Pydantic 模式定义了属性和类型来验证一些有效负载。

在您实际允许服务完成特定操作（例如创建数据库对象）之前，它们就像守卫一样。我不确定你是否习惯了序列化程序，但除了 Pydantic 和 FastAPI 与新的 Python 3 个属性（参见 type checking）集成之外，它几乎是一样的，这使得实现你以前用框架做的事情 builtins/libraries.

在您的示例中，UserCreate 仅需要密码和电子邮件地址，因此继承使您的代码更加 DRY。

在使用 FastAPI 构建 API 时，Python 中的 pydantic 模型（模式）是什么意思

What is the meaning of pydantic models(schemas) in Python while building an API with FastAPI

python

api

postgresql

sqlalchemy

fastapi