前言
新增项目，一般都会记录是哪个用户新建的，往往一个用户可以新建多个项目，于是数据库模型就可以建立一对多的关系。
1. 新建数据库模型(一对多)
# apps/model/models.py
class User(DbModel)
......
# 一对多
projects = relationship('Project', backref='user', cascade='all,delete')
class Project(DbModel):
"""项目信息"""
__tablename__ = 'project'
id = Column(Integer, primary_key=True, autoincrement=True)
project_name = Column(String(100), unique=True, nullable=False)
publish_app = Column(String(100), nullable=True)
project_des = Column(String(600), nullable=True)
active = Column(Enum("1", "0"), default="1", nullable=True)
# 所属用户
user_id = Column(Integer, ForeignKey('user.id'))
新增数据库模型后，就可以进行表的迁移：
- alembic revision --autogenerate -m "test" # 自定义迁移文件名
- alembic upgrade head
迁移成功后，就可以在数据库中查看到新建了数据库表project。
1.1 新增crud
# apps/routers/crud.py
from apps.model.models import User, Project
def create_project(db: Session, body):
db_project = Project(**body.dict())
# 先判断是否存在，在入库
if db.query(Project).filter_by(project_name=db_project.project_name).count():
# 项目名称重复添加
raise HTTPException(status_code=400, detail=f'项目名称：{body.project_name},已存在')
try:
db.add(db_project)
db.commit()
db.refresh(db_project)
except Exception as e:
print(f"数据写入失败：{e}")
raise HTTPException(status_code=400, detail='添加项目失败！')
return db_project
创建新建项目的接口函数，并约定数据输入和输出模型。
我们先建立数据的输入和输出模型。
# apps/model/schemas.py
class ProjectIn(BaseModel):
project_name: str = Field(..., max_length=100, description="项目名称")
publish_app: str = Field(None, max_length=100)
project_des: str = Field(None, max_length=600)
user_id: int = Field(None)
class ProjectOut(BaseModel):
id: int
project_name: str
project_app: str = ''
project_des: str = ''
active: str = '1'
user_id: int = ''
class Config:
orm_mode = True # 将查询到的orm对象，转换成json
# apps/routers/view_project.py
from fastapi import APIRouter, Depends
from fastapi_jwt_auth import AuthJWT
from apps.dependency import get_db
from sqlalchemy.orm import Session
from apps.model.schemas import ProjectIn, ProjectOut
from apps.model.crud import create_project
router = APIRouter()
@router.post("/project", response_model=ProjectOut)
async def create_project_demo(
body: ProjectIn,
Authorize: AuthJWT = Depends(),
db: Session = Depends(get_db)
):
"""新增项目数据"""
Authorize.jwt_required() # 需要先登录才能请求该接口
# 获取额外的数据
user_claims= Authorize.get_raw_jwt()
# 得到用户名
print(f'用户名：{user_claims["username"]}')
body.user_id = user_claims["id"]
res = create_project(db, body)
return res
工厂函数create_app中，注册路由。
# 注册路由
app.include_router(view_project.router, prefix="/api/v1", tags=["项目"])
