Fastapi教程四:项目妹子图(3)-user模型的增删改查

in fastapi with 6026 comments

一、创建apps/users/schemas.py:

from typing import List

from pydantic import BaseModel

 
class UserBase(BaseModel):
    username: str
    email: str


class UserCreate(UserBase):
    password: str


class UserUpdate(BaseModel): 
    email:str = None
    password: str = None
    name: str= None
    last_session_key:str = None
    is_active: bool= True
    

class User(UserBase):
    id:int
    uuid:str
    name: str= None
    is_active: bool 
    is_super: bool

    class Config:
        orm_mode = True

class UserInDB(User):
    hashed_password: str

实际上就是serilazer化model.具体不介绍了,想了解的可以看文档:https://fastapi.tiangolo.com/zh/tutorial/response-model/

二、创建apps/users/crud.py:

from typing import Any, Dict, Optional, Union

from sqlalchemy.orm import Session
from fastapi.encoders import jsonable_encoder

from extensions.fastapi_password import set_password

from . import models, schemas


#获取用户BY ID
def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()

#获取用户BY username
def get_user_by_username(db: Session, username: str):
    return db.query(models.User).filter(models.User.username == username).first()

#获取用户列表,这里包含了分页,skip =(page-1)*limit
def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()

#下面是增删改
def create_user(db: Session, user: schemas.UserCreate):
    db_user = models.User(username=user.username, email=user.email)
    db_user.set_password(user.password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def update_user(db: Session, user_id:int, obj_in: Union[schemas.UserUpdate, Dict[str, Any] ]):
    user = db.query(models.User).filter(models.User.id == user_id).first()
    if user:
        print(obj_in)
        if isinstance(obj_in, dict):
            update_data = obj_in
        else:
            update_data = obj_in.dict(exclude_unset=True)
        if "password" in update_data:
            update_data["password"] = set_password(update_data["password"] )
        db.query(models.User).update(update_data)
        db.commit()
        db.refresh(user)
        return user

def delete_user(db: Session, user_id: int):
    db_user = db.query(models.User).filter(models.User.id == user_id).first()
    if db_user:
        db.delete(db_user)
        db.commit()
        db.flush()
        return db_user

三、创建API-router
apps/users/router.py

from fastapi import Depends, FastAPI, HTTPException,APIRouter
from fastapi.encoders import jsonable_encoder
from sqlalchemy.orm import Session


from core.database import get_db
from . import crud, schemas

router = APIRouter() 

 

@router.post("/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_username(db, username=user.username)
    if db_user:
        raise HTTPException(status_code=400, detail="Email or Username already registered")
    return crud.create_user(db, user)

@router.patch("/{user_id}", response_model=schemas.UserUpdate)
def update_user(user_id: int, user: schemas.UserUpdate, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id)
    print(user)
    if not db_user:
        raise HTTPException(status_code=400, detail="User is not exsit")
    return crud.update_user(db,user_id, user)


@router.get("/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@router.get("/{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

在main.py中加入user的router

import uvicorn
from core import get_app
from core.database import engine

from apps.users import models as users_model
from apps.router import router as  user_router

users_model.Base.metadata.create_all(bind=engine)

app = get_app()
+app.include_router(user_router, prefix="/users")



if __name__ == "__main__":
    uvicorn.run(app="main:app", host="0.0.0.0", port=3200, reload=True, debug=True)

打开http://127.0.0.1:3200/docs 就可以看见接口文档:
Users的接口列表

测试USER

Comments are closed.