一、创建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 就可以看见接口文档：
本文由 admin 创作，采用 知识共享署名4.0 国际许可协议进行许可
本站文章除注明转载/出处外，均为本站原创或翻译，转载前请务必署名
最后编辑时间为: Jul 29, 2020 at 10:27 pm