from sqlalchemy import case, exists, and_, or_, text, any_, not_ from sqlalchemy.orm import Query, aliased from models import user_label, UserProject, Project, User, ArtistTeamUser from models.user_project import UserProjectRoles PROJECT_SHARED_STATUS = {"BY_ME": "by_me", "TO_ME": "to_me"} class Queries: @staticmethod def roles_for_project_query(user_id, user_project_roles, artist_team_roles) -> Query: def int_array(*args): return text(f"ARRAY{[arg for arg in args]}::integer[]") def case_approval_role(init_role: int): return case( [ ( UserProjectRoles.APPROVER.value == any_(user_project_roles), int_array(init_role, UserProjectRoles.APPROVER.value) ) ], else_=int_array(init_role) ) def case_artist_team_and_approval_role(artist_team_roles): return case( [ ( UserProjectRoles.APPROVER.value == any_(user_project_roles), text(f"{artist_team_roles} || ARRAY{[UserProjectRoles.APPROVER.value]}::integer[]") ) ], else_=artist_team_roles ) if_admin = and_( user_label.c.user_id == user_id, user_label.c.label_id == Project.label_id, User.is_admin.is_(True) ) if_in_artist_team = and_( Project.is_confidential.is_(False), artist_team_roles.isnot(None), User.is_admin.is_(False), ) if_in_project_team = and_( user_project_roles.isnot(None), not_(user_project_roles == int_array(UserProjectRoles.APPROVER.value)), artist_team_roles.is_(None), or_( User.is_admin.is_(False), # Admin from another label should use collaborator role and_( User.is_admin.is_(True), user_label.c.user_id == user_id, user_label.c.label_id != Project.label_id ), ), ) if_labelmate = and_( Project.is_confidential.is_(False), user_label.c.label_id == Project.label_id, user_label.c.user_id == user_id, User.is_admin.is_(False), artist_team_roles.is_(None), ) return case( [ (if_admin, case_approval_role(UserProjectRoles.ADMIN.value)), (if_in_artist_team, case_artist_team_and_approval_role(artist_team_roles)), (if_in_project_team, user_project_roles), (if_labelmate, case_approval_role(UserProjectRoles.LABELMATE.value)), ], else_=user_project_roles, ) @staticmethod def role_for_project_query(user_id) -> Query: if_admin = and_( user_label.c.user_id == user_id, user_label.c.label_id == Project.label_id, User.is_admin.is_(True) ) if_in_artist_team = and_(ArtistTeamUser.role.isnot(None), User.is_admin.is_(False)) if_in_team = and_( UserProject.role.isnot(None), ArtistTeamUser.role.is_(None), or_( User.is_admin.is_(False), # Admin from another label should use collaborator role and_( User.is_admin.is_(True), user_label.c.user_id == user_id, user_label.c.label_id != Project.label_id ), ), ) if_labelmate = and_( user_label.c.label_id == Project.label_id, user_label.c.user_id == user_id, User.is_admin.is_(False), ArtistTeamUser.role.is_(None), ) return case( [ (if_admin, UserProjectRoles.ADMIN.value), (if_in_artist_team, ArtistTeamUser.role), (if_in_team, UserProject.role), (if_labelmate, UserProjectRoles.LABELMATE.value), ], else_=UserProject.role, ) @staticmethod def project_shared_status_query(user_id: int) -> Query: alias_user_project = aliased(UserProject) return case( [ ( exists().where( and_(alias_user_project.project_id == Project.id, alias_user_project.shared_by == user_id) ), PROJECT_SHARED_STATUS["BY_ME"], ), ( exists().where( and_(alias_user_project.project_id == Project.id, alias_user_project.user_id == user_id) ), PROJECT_SHARED_STATUS["TO_ME"], ), ], else_=None, )