in backend/services/project_search_service.py [0:0]
def _filter_projects(search_dto: ProjectSearchDTO, user):
""" Filters all projects based on criteria provided by user"""
query = ProjectSearchService.create_search_query(user)
query = query.join(ProjectInfo).filter(
ProjectInfo.locale.in_([search_dto.preferred_locale, "en"])
)
project_status_array = []
if search_dto.project_statuses:
project_status_array = [
ProjectStatus[project_status].value
for project_status in search_dto.project_statuses
]
query = query.filter(Project.status.in_(project_status_array))
else:
if not search_dto.created_by:
project_status_array = [ProjectStatus.PUBLISHED.value]
query = query.filter(Project.status.in_(project_status_array))
if search_dto.interests:
query = query.join(
project_interests, project_interests.c.project_id == Project.id
).filter(project_interests.c.interest_id.in_(search_dto.interests))
if search_dto.created_by:
query = query.filter(Project.author_id == search_dto.created_by)
if search_dto.mapped_by:
projects_mapped = UserService.get_projects_mapped(search_dto.mapped_by)
query = query.filter(Project.id.in_(projects_mapped))
if search_dto.favorited_by:
projects_favorited = user.favorites
query = query.filter(
Project.id.in_([project.id for project in projects_favorited])
)
if search_dto.mapper_level and search_dto.mapper_level.upper() != "ALL":
query = query.filter(
Project.mapper_level == MappingLevel[search_dto.mapper_level].value
)
if search_dto.action and search_dto.action != "any":
if search_dto.action == "map":
query = ProjectSearchService.filter_projects_to_map(query, user)
if search_dto.action == "validate":
query = ProjectSearchService.filter_projects_to_validate(query, user)
if search_dto.organisation_name:
query = query.filter(Organisation.name == search_dto.organisation_name)
if search_dto.organisation_id:
query = query.filter(Organisation.id == search_dto.organisation_id)
if search_dto.team_id:
query = query.join(
ProjectTeams, ProjectTeams.project_id == Project.id
).filter(ProjectTeams.team_id == search_dto.team_id)
if search_dto.campaign:
query = query.join(Campaign, Project.campaign).group_by(Campaign.name)
query = query.filter(Campaign.name == search_dto.campaign)
if search_dto.mapping_types:
# Construct array of mapping types for query
mapping_type_array = []
if search_dto.mapping_types_exact:
mapping_type_array = [
{
MappingTypes[mapping_type].value
for mapping_type in search_dto.mapping_types
}
]
query = query.filter(Project.mapping_types.in_(mapping_type_array))
else:
mapping_type_array = [
MappingTypes[mapping_type].value
for mapping_type in search_dto.mapping_types
]
query = query.filter(Project.mapping_types.overlap(mapping_type_array))
if search_dto.text_search:
# We construct an OR search, so any projects that contain or more of the search terms should be returned
or_search = " | ".join(
[x for x in search_dto.text_search.split(" ") if x != ""]
)
opts = [
ProjectInfo.text_searchable.match(
or_search, postgresql_regconfig="english"
),
ProjectInfo.name.ilike(f"%{or_search}%"),
]
try:
opts.append(Project.id == int(search_dto.text_search))
except ValueError:
pass
query = query.filter(or_(*opts))
if search_dto.country:
# Unnest country column array.
sq = Project.query.with_entities(
Project.id, func.unnest(Project.country).label("country")
).subquery()
query = query.filter(
sq.c.country.ilike("%{}%".format(search_dto.country))
).filter(Project.id == sq.c.id)
if search_dto.last_updated_gte:
last_updated_gte = validate_date_input(search_dto.last_updated_gte)
query = query.filter(Project.last_updated >= last_updated_gte)
if search_dto.last_updated_lte:
last_updated_lte = validate_date_input(search_dto.last_updated_lte)
query = query.filter(Project.last_updated <= last_updated_lte)
if search_dto.created_gte:
created_gte = validate_date_input(search_dto.created_gte)
query = query.filter(Project.created >= created_gte)
if search_dto.created_lte:
created_lte = validate_date_input(search_dto.created_lte)
query = query.filter(Project.created <= created_lte)
order_by = search_dto.order_by
if search_dto.order_by_type == "DESC":
order_by = desc(search_dto.order_by)
query = query.order_by(order_by).distinct(search_dto.order_by, Project.id)
if search_dto.managed_by and user.role != UserRole.ADMIN.value:
# Get all the projects associated with the user and team.
orgs_projects_ids = [[p.id for p in u.projects] for u in user.organisations]
orgs_projects_ids = [
item for sublist in orgs_projects_ids for item in sublist
]
team_project_ids = [
[
p.project_id
for p in u.team.projects
if p.role == TeamRoles.PROJECT_MANAGER.value
]
for u in user.teams
]
team_project_ids = [
item for sublist in team_project_ids for item in sublist
]
orgs_projects_ids.extend(team_project_ids)
ids = tuple(set(orgs_projects_ids))
query = query.filter(Project.id.in_(ids))
all_results = []
if not search_dto.omit_map_results:
query_result = query
query_result.column_descriptions.clear()
query_result.add_column(Project.id)
query_result.add_column(Project.centroid.ST_AsGeoJSON().label("centroid"))
query_result.add_column(Project.priority)
all_results = query_result.all()
paginated_results = query.paginate(search_dto.page, 14, True)
return all_results, paginated_results