我是 python 新手，我正在尝试使用 FastAPI 构建一个 API。
到目前为止它一直在工作，我连接了 postgres db，我发出了 post/get/ 请求，一切正常，但我不明白为什么我们这样定义模式，为什么我们必须创建一个
类 UserBase(BaseModel)
类 UserCreate(UserBase)
类 User(UserBase)
我将发布所有文件的源代码，如果你们能帮助我对此有一个很好的理解，那真的对我很有帮助，因为我明天有一个任务。
架构.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
模型.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
数据库.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 进行验证，但是另外两个呢，谁能给我一个很好的解释..
谢谢你。