86 lines
3.0 KiB
Python
86 lines
3.0 KiB
Python
from typing import Any, Dict, Optional, Union
|
|
|
|
from fastapi.encoders import jsonable_encoder
|
|
from sqlalchemy.orm import Session
|
|
|
|
from crud.base import CRUDBase
|
|
from models.authority import Authority
|
|
from schemas import AuthorityCreate, AuthorityUpdate
|
|
|
|
|
|
class CRUDAuthority(CRUDBase[Authority, AuthorityCreate, AuthorityUpdate]):
|
|
def create_with_authority(
|
|
self, db: Session, *, obj_in: AuthorityCreate, user_id: int
|
|
) -> Authority:
|
|
obj_in_data = jsonable_encoder(obj_in)
|
|
|
|
db_obj = self.model(**obj_in_data, user_id=user_id)
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
def get_my_dashboard(self, db: Session, project_id: int, user_id: int):
|
|
dashboards = db.query(Authority).filter(Authority.project_id == project_id, Authority.user_id == user_id).all()
|
|
res = {'kanban': dict(), 'space': dict()}
|
|
for item in dashboards:
|
|
# 是空间的
|
|
dashboard = item.dashboard
|
|
folder = item.folder
|
|
space = item.space
|
|
if space:
|
|
s = res['space'].setdefault(space.id, {'dashboard': [], 'folder': {}})
|
|
s['name'] = space.name
|
|
s['id'] = space.id
|
|
if folder:
|
|
f = s['folder'].setdefault(folder.id, {'dashboard': []})
|
|
f['name'] = folder.name
|
|
f['id'] = folder.id
|
|
if dashboard:
|
|
f['dashboard'].append({
|
|
'name': dashboard.name,
|
|
'id': dashboard.id
|
|
})
|
|
elif dashboard:
|
|
d = s['dashboard']
|
|
d.append({
|
|
'name': dashboard.name,
|
|
'id': dashboard.id
|
|
})
|
|
|
|
else:
|
|
# 是看板的文件夹
|
|
folder = item.folder
|
|
f = res['kanban'].setdefault(folder.id, {'dashboard': []})
|
|
f['name'] = folder.name
|
|
f['id'] = folder.id
|
|
if dashboard:
|
|
f['dashboard'].append({
|
|
'name': dashboard.name,
|
|
'id': dashboard.id
|
|
})
|
|
return res
|
|
|
|
def delete(self, db: Session, user_id: int, project_id: int = None, folder_id: int = None, space_id: int = None,
|
|
dashboard_id: int = None):
|
|
where = [
|
|
Authority.user_id == user_id
|
|
]
|
|
if project_id is not None:
|
|
where.append(Authority.project_id == project_id)
|
|
|
|
if space_id is not None:
|
|
where.append(Authority.space_id == space_id)
|
|
|
|
if folder_id is not None:
|
|
where.append(Authority.folder_id == folder_id)
|
|
|
|
if dashboard_id is not None:
|
|
where.append(Authority.dashboard_id == dashboard_id)
|
|
|
|
if not db.query(Authority).filter(*where).delete():
|
|
return
|
|
|
|
|
|
authority = CRUDAuthority(Authority)
|