我刚开始学习Python，并尝试使用FastAPI构建API。
目前为止，一切正常，我已经连接了Postgres数据库，进行了POST/GET请求，并且一切都运行良好，但是我并不理解为什么我们要像这样定义架构，为什么我们必须创建以下内容：
class UserBase(BaseModel)
class UserCreate(UserBase)
class User(UserBase)
我将在所有文件中发布源代码，如果你们能帮助我更好地理解这个问题，那真的会对我帮助很大，因为我明天还有一个任务要完成。
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
模型.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