VB.net 2010 视频教程 VB.net 2010 视频教程 python基础视频教程
SQL Server 2008 视频教程 c#入门经典教程 Visual Basic从门到精通视频教程
当前位置:
首页 > temp > python入门教程 >
  • 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="账号不能重复")

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

 

 我们看下最后的数据库的存储。

 

 其实操作数据库还是很简单的。

来源:https://www.cnblogs.com/leiziv5/p/15416790.html

 


相关教程