FastAPI 学习之路（三十三）操作数据库

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

通过创建pydantic模型进行验证提交数据

from pydantic import BaseModel

class UserBase(BaseModel):

    email: str

class UserCreate(UserBase):

    """

    请求模型验证：

    email:

    password:

    """

    password: str

class Users(UserBase):

    """

    响应模型：

    id:email: is_active并且设置orm_mode与之兼容

    """

    id: int

    is_active: bool

    class Config:

        orm_mode = True

我们去通过传入数据库连接以及参数等进行数据库操作，包括创建用户、查询用户等，返回的是orm模型对象。

from sqlalchemy.orm import Session

# 通过id查询用户

def get_user(db: Session, user_id: int):

    return db.query(User).filter(User.id == user_id).first()

# 新建用户

def db_create_user(db: Session, user: UserCreate):

    fake_hashed_password = user.password + "notreallyhashed"

    db_user = User(email=user.email, hashed_password=fake_hashed_password)

    db.add(db_user)

    db.commit()  # 提交保存到数据库中

    db.refresh(db_user)  # 刷新

    return db_user

接下来，我们就是在实际的接口中调用

app = FastAPI()

# Dependency

def get_db():

    """

    每一个请求处理完毕后会关闭当前连接，不同的请求使用不同的连接

    :return:

    """

    db = SessionLocal()

    try:

        yield db

    finally:

        db.close()

# 新建用户

@app.post("/users/", response_model=Users)

def create_user(user: UserCreate, db: Session = Depends(get_db)):

    return db_create_user(db=db, user=user)

# 通过id查询用户

@app.get("/user/{user_id}", response_model=Users)

def read_user(user_id: int, db: Session = Depends(get_db)):

    db_user = get_user(db, user_id=user_id)

    if not db_user:

        raise HTTPException(status_code=404, detail="User not found")

    return db_user

我们去启动下，看下对应的接口。

uvicorn main:app --reload

我们去用postman去请求下

创建下用户

查询下账户

我们去查询不存在的id

我们在用postman去请求的时候，发现

报错了，我们看下如何处理

因为是唯一的索引，所以我们在去创建用户的时候，要去判断用户是否存在。

我们先去创建一个通过email查询用户的

def get_user_emai(db:Session,email:str):

    return db.query(User).filter(User.email==email).first()

我们在创建用户的时候，我们去判断下

@app.post("/users/", response_model=Users)

def create_user(user: UserCreate, db: Session = Depends(get_db)):

    db_crest=get_user_emai(db,user.email)

    if not db_crest:

        return db_create_user(db=db, user=user)

    raise HTTPException(status_code=200, detail="账号不能重复")

如果不存在，我们就去创建，如果存在，我们就直接返回。我们看下接口