36 lines
1.1 KiB
Python
36 lines
1.1 KiB
Python
from typing import Any, Dict, Optional, Union
|
|
|
|
from fastapi import HTTPException
|
|
from fastapi.encoders import jsonable_encoder
|
|
from sqlalchemy.orm import Session
|
|
|
|
from crud.base import CRUDBase
|
|
from models.authority import Authority
|
|
|
|
from models.project import Project
|
|
|
|
from schemas.project import ProjectCreate, ProjectUpdate
|
|
|
|
|
|
class CRUDProject(CRUDBase[Project, ProjectCreate, ProjectUpdate]):
|
|
def create_with_project(
|
|
self, db: Session, *, obj_in: ProjectCreate, user_id: int
|
|
) -> Project:
|
|
if db.query(self.model).filter(self.model.name == obj_in.name).first():
|
|
raise HTTPException(status_code=406, detail="项目名已存在")
|
|
|
|
obj_in_data = jsonable_encoder(obj_in)
|
|
db_obj = Project(**obj_in_data, user_id=user_id)
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
def get_my_project(
|
|
self, db: Session, *, user_id: int
|
|
) -> Project:
|
|
return db.query(self.model).join(Authority).filter(Authority.user_id == user_id).all()
|
|
|
|
|
|
project = CRUDProject(Project)
|