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

雷子

发布于 2021-09-17 11:13:45

6590

发布于 2021-09-17 11:13:45

文章被收录于专栏：雷子说测试开发

前言

我们之前分享分享使用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="账号不能重复")

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