| | |
| | | from app.models import RoleParameter, GroupModel, KnowledgeModel |
| | | from app.models.base_model import get_db |
| | | from app.models.user_model import UserModel |
| | | from app.service.knowledge import get_knowledge_list |
| | | |
| | | knowledge_router = APIRouter() |
| | | |
| | | @knowledge_router.get("/list", response_model=Response) |
| | | async def knowledge_list(role:str, current_user: UserModel = Depends(get_current_user), |
| | | async def knowledge_list(role:str="", current_user: UserModel = Depends(get_current_user), |
| | | db=Depends(get_db)): |
| | | if role == "admin": |
| | | kld_list = db.query(KnowledgeModel).all() |
| | | else: |
| | | user = db.query(UserModel).filter(UserModel.id == current_user.id).first() |
| | | |
| | | group_list = [i.id for i in user.groups] |
| | | |
| | | kld_list = db.query(KnowledgeModel).filter(KnowledgeModel.tenant_id==current_user.id).all() |
| | | kld_set = set([i.id for i in kld_list]) |
| | | for group in db.query(GroupModel).filter(GroupModel.id.in_(group_list)).all(): |
| | | for k in group.knowledges: |
| | | if k.id not in kld_set: |
| | | kld_list.append(k) |
| | | kld_set.add(k.id) |
| | | |
| | | data = [kld.to_json() for kld in kld_list] |
| | | return Response(code=200, msg="", data={"rows": data}) |
| | | return Response(code=200, msg="", data = await get_knowledge_list(db, role, current_user.id)) |