backend/services/organisation_service.py (231 lines of code) (raw):
from flask import current_app
from sqlalchemy.exc import IntegrityError
from backend import db
from backend.models.dtos.organisation_dto import (
OrganisationDTO,
NewOrganisationDTO,
ListOrganisationsDTO,
UpdateOrganisationDTO,
)
from backend.models.dtos.stats_dto import (
OrganizationStatsDTO,
OrganizationProjectsStatsDTO,
OrganizationTasksStatsDTO,
)
from backend.models.postgis.campaign import campaign_organisations
from backend.models.postgis.organisation import Organisation
from backend.models.postgis.project import Project, ProjectInfo
from backend.models.postgis.task import Task
from backend.models.postgis.statuses import ProjectStatus, TaskStatus
from backend.models.postgis.utils import NotFound
from backend.services.users.user_service import UserService
class OrganisationServiceError(Exception):
""" Custom Exception to notify callers an error occurred when handling organisations """
def __init__(self, message):
if current_app:
current_app.logger.debug(message)
class OrganisationService:
@staticmethod
def get_organisation_by_id(organisation_id: int) -> Organisation:
org = Organisation.get(organisation_id)
if org is None:
raise NotFound()
return org
@staticmethod
def get_organisation_by_id_as_dto(
organisation_id: int, user_id: int, abbreviated: bool
):
org = Organisation.get(organisation_id)
return OrganisationService.get_organisation_dto(org, user_id, abbreviated)
@staticmethod
def get_organisation_by_slug_as_dto(slug: str, user_id: int, abbreviated: bool):
org = Organisation.query.filter_by(slug=slug).first()
return OrganisationService.get_organisation_dto(org, user_id, abbreviated)
@staticmethod
def get_organisation_dto(org, user_id: int, abbreviated: bool):
if org is None:
raise NotFound()
organisation_dto = org.as_dto(abbreviated)
if user_id != 0:
organisation_dto.is_manager = (
OrganisationService.can_user_manage_organisation(org.id, user_id)
)
else:
organisation_dto.is_manager = False
if abbreviated:
return organisation_dto
organisation_dto.teams = [team.as_dto_inside_org() for team in org.teams]
return organisation_dto
@staticmethod
def get_organisation_by_name(organisation_name: str) -> Organisation:
organisation = Organisation.get_organisation_by_name(organisation_name)
if organisation is None:
raise NotFound()
return organisation
@staticmethod
def get_organisation_name_by_id(organisation_id: int) -> str:
return Organisation.get_organisation_name_by_id(organisation_id)
@staticmethod
def create_organisation(new_organisation_dto: NewOrganisationDTO) -> int:
"""
Creates a new organisation using an organisation dto
:param new_organisation_dto: Organisation DTO
:returns: ID of new Organisation
"""
try:
org = Organisation.create_from_dto(new_organisation_dto)
return org.id
except IntegrityError:
raise OrganisationServiceError(
f"Organisation name already exists: {new_organisation_dto.name}"
)
@staticmethod
def update_organisation(organisation_dto: UpdateOrganisationDTO) -> Organisation:
"""
Updates an organisation
:param organisation_dto: DTO with updated info
:returns updated Organisation
"""
org = OrganisationService.get_organisation_by_id(
organisation_dto.organisation_id
)
OrganisationService.assert_validate_name(org, organisation_dto.name)
OrganisationService.assert_validate_users(organisation_dto)
org.update(organisation_dto)
return org
@staticmethod
def delete_organisation(organisation_id: int):
""" Deletes an organisation if it has no projects """
org = OrganisationService.get_organisation_by_id(organisation_id)
if org.can_be_deleted():
org.delete()
else:
raise OrganisationServiceError(
"Organisation has projects, cannot be deleted"
)
@staticmethod
def get_organisations(manager_user_id: int):
if manager_user_id is None:
""" Get all organisations """
return Organisation.get_all_organisations()
else:
return Organisation.get_organisations_managed_by_user(manager_user_id)
@staticmethod
def get_organisations_as_dto(
manager_user_id: int, authenticated_user_id: int, omit_managers: bool
):
orgs = OrganisationService.get_organisations(manager_user_id)
orgs_dto = ListOrganisationsDTO()
for org in orgs:
org_dto = org.as_dto(omit_managers)
if not authenticated_user_id:
del org_dto.managers
orgs_dto.organisations.append(org_dto)
return orgs_dto
@staticmethod
def get_organisations_managed_by_user(user_id: int):
""" Get all organisations a user manages """
if UserService.is_user_an_admin(user_id):
return Organisation.get_all_organisations()
return Organisation.get_organisations_managed_by_user(user_id)
@staticmethod
def get_organisations_managed_by_user_as_dto(user_id: int) -> ListOrganisationsDTO:
orgs = OrganisationService.get_organisations_managed_by_user(user_id)
orgs_dto = ListOrganisationsDTO()
orgs_dto.organisations = [org.as_dto() for org in orgs]
return orgs_dto
@staticmethod
def get_projects_by_organisation_id(organisation_id: int) -> Organisation:
projects = (
db.session.query(Project.id, ProjectInfo.name)
.join(ProjectInfo)
.filter(Project.organisation_id == organisation_id)
.all()
)
if projects is None:
raise NotFound()
return projects
@staticmethod
def get_organisation_stats(organisation_id: int) -> OrganizationStatsDTO:
projects = db.session.query(Project.id, Project.status).filter(
Project.organisation_id == organisation_id
)
published_projects = projects.filter(
Project.status == ProjectStatus.PUBLISHED.value
)
active_tasks = db.session.query(
Task.id, Task.project_id, Task.task_status
).filter(Task.project_id.in_([i.id for i in published_projects.all()]))
# populate projects stats
projects_dto = OrganizationProjectsStatsDTO()
projects_dto.draft = projects.filter(
Project.status == ProjectStatus.DRAFT.value
).count()
projects_dto.published = published_projects.count()
projects_dto.archived = projects.filter(
Project.status == ProjectStatus.ARCHIVED.value
).count()
# populate tasks stats
tasks_dto = OrganizationTasksStatsDTO()
tasks_dto.ready = active_tasks.filter(
Task.task_status == TaskStatus.READY.value
).count()
tasks_dto.locked_for_mapping = active_tasks.filter(
Task.task_status == TaskStatus.LOCKED_FOR_MAPPING.value
).count()
tasks_dto.mapped = active_tasks.filter(
Task.task_status == TaskStatus.MAPPED.value
).count()
tasks_dto.locked_for_validation = active_tasks.filter(
Task.task_status == TaskStatus.LOCKED_FOR_VALIDATION.value
).count()
tasks_dto.validated = active_tasks.filter(
Task.task_status == TaskStatus.VALIDATED.value
).count()
tasks_dto.invalidated = active_tasks.filter(
Task.task_status == TaskStatus.INVALIDATED.value
).count()
tasks_dto.badimagery = active_tasks.filter(
Task.task_status == TaskStatus.BADIMAGERY.value
).count()
# populate and return main dto
stats_dto = OrganizationStatsDTO()
stats_dto.projects = projects_dto
stats_dto.active_tasks = tasks_dto
return stats_dto
@staticmethod
def assert_validate_name(org: Organisation, name: str):
""" Validates that the organisation name doesn't exist """
if org.name != name and Organisation.get_organisation_by_name(name) is not None:
raise OrganisationServiceError(f"Organisation name already exists: {name}")
@staticmethod
def assert_validate_users(organisation_dto: OrganisationDTO):
""" Validates that the users exist"""
if organisation_dto.managers and len(organisation_dto.managers) == 0:
raise OrganisationServiceError("Must have at least one admin")
if organisation_dto.managers and len(organisation_dto.managers) > 0:
managers = []
for user in organisation_dto.managers:
try:
admin = UserService.get_user_by_username(user)
except NotFound:
raise NotFound(f"User {user} does not exist")
managers.append(admin.username)
organisation_dto.managers = managers
@staticmethod
def can_user_manage_organisation(organisation_id: int, user_id: int):
""" Check that the user is an admin for the org or a global admin"""
if UserService.is_user_an_admin(user_id):
return True
else:
return OrganisationService.is_user_an_org_manager(organisation_id, user_id)
@staticmethod
def is_user_an_org_manager(organisation_id: int, user_id: int):
""" Check that the user is an manager for the org """
org = Organisation.get(organisation_id)
if org is None:
raise NotFound()
user = UserService.get_user_by_id(user_id)
return user in org.managers
@staticmethod
def get_campaign_organisations_as_dto(campaign_id: int, user_id: int):
"""
Returns organisations under a particular campaign
"""
organisation_list_dto = ListOrganisationsDTO()
orgs = (
Organisation.query.join(campaign_organisations)
.filter(campaign_organisations.c.campaign_id == campaign_id)
.all()
)
for org in orgs:
if user_id != 0:
logged_in = OrganisationService.can_user_manage_organisation(
org.id, user_id
)
else:
logged_in = False
organisation_dto = OrganisationDTO()
organisation_dto.organisation_id = org.id
organisation_dto.name = org.name
organisation_dto.logo = org.logo
organisation_dto.url = org.url
organisation_dto.is_manager = logged_in
organisation_list_dto.organisations.append(organisation_dto)
return organisation_list_dto