探索 Pydbantic：Python数据库操作的未来

from pydantic import BaseModel
from typing import Optional
from databases import Database
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from fastapi_utils.cbv import cbv
from fastapi import FastAPI, Depends
 
app = FastAPI()
 
# 定义数据库配置
DATABASE_URL = "sqlite:///./test.db"
 
# 初始化数据库
database = Database(DATABASE_URL)
 
# 创建SQLAlchemy引擎
engine = create_engine(DATABASE_URL)
 
# 创建SQLAlchemy的基类
Base = declarative_base()
 
# 定义一个Pydantic模型，用于数据库表的映射
class Item(BaseModel):
    id: Optional[int] = None
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
 
    class Config:
        orm_mode = True
 
# 定义数据库表
class ItemTable(Base):
    __tablename__ = "items"
    id = Base.Column(Base.Integer, primary_key=True, autoincrement=True)
    name = Base.Column(Base.String(255))
    description = Base.Column(Base.String(255))
    price = Base.Column(Base.Float)
    tax = Base.Column(Base.Float)
 
    def __str__(self):
        return self.name
 
# 创建表
Base.metadata.create_all(bind=engine)
 
# 创建SQLAlchemy会话
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
 
# 获取会话
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
 
# FastAPI路由依赖项
async def get_db_session():
    async with database.session() as session:
        yield session
 
# 创建CRUD操作的基类
class CRUDBase:
    def create(self, db: Session, *, obj_in: Item):
        db_obj = ItemTable(**obj_in.dict())
        db.add(db_obj)
        db.commit()
        db.refresh(db_obj)
        return db_obj
 
    def get(self, db: Session, *, id: int):
        return db.query(ItemTable).filter(ItemTable.id == id).first()
 
    def get_multi(self, db: Session, *, skip=0, limit=100):
        return db.query(ItemTable).offset(skip).limit(limit).all()
 
# 定义CRUD操作
class ItemCRUD(CRUDBase):
    def get_by_name(self, db: Session, *, name: str):
        return db.query(ItemTable).filter(ItemTable.name == name).first()
 
# 初始化CRUD操作
item_crud = ItemCRUD()
 
# FastAPI路由
@app.post("/items/", response_model=Item)
async def create_item(*, item_in: Item, session: Session = Depends(get_db_session)):
    return item_crud.create(session, obj_in=item_in)
 
@app.get("/items/{item_id}")
asy