36d6dc01-6726-4ee7-a9dd-ce35c506949e

pending completion

Build # 36d6dc01-6726-4ee7-a9dd-ce35c506949e

Build Type

push

semaphore-ci

Committed by

web-flow

Commit Message

Merge pull request #5298 from akvo/cleanup-rest-endpoints

Cleanup REST endpoints

Run Details

34 of 34 new or added lines in 11 files covered. (100.0%)

17025 of 25071 relevant lines covered (67.91%)

0.68 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/akvo/rsr/filters.py

# -*- coding: utf-8 -*-

"""Akvo RSR is covered by the GNU Affero General Public License.
See more details in the license.txt file located at the root folder of the
Akvo RSR module. For additional details on the GNU license please
see < http://www.gnu.org/licenses/agpl.html >.
"""

from copy import deepcopy

from django.db.models import Q
from django.utils.translation import ugettext_lazy as _

from akvo.codelists.store.default_codelists import SECTOR_CATEGORY
from akvo.utils import codelist_choices
from .models import (Organisation, OrganisationLocation,
                     Project, ProjectLocation, ProjectUpdate, ProjectUpdateLocation,
                     RecipientCountry)
from .m49 import M49_CODES, M49_HIERARCHY

ANY_CHOICE = (('', _('All')), )


def sectors():
    sectors_list = []
    for sector in codelist_choices(SECTOR_CATEGORY):
        sectors_list.append(sector)
    return sectors_list


def get_orgs():
    orgs = list(Organisation.objects.all().values_list(
                'id', 'name', flat=False))
    return [('', _('All'))] + orgs


def remove_empty_querydict_items(request_get):
    # querydicts are immutable
    getvars = request_get.copy()
    for k, v in list(getvars.items()):
        if not v:
            getvars.pop(k, None)
    return getvars


def walk(node):
    """Walks the m49 tree and return countries"""

    if isinstance(node, str):
        return [node, ]
    elif isinstance(node, int):
        return walk(deepcopy(M49_HIERARCHY)[node])
    else:
        return (walk(node.pop()) + walk(node)) if node else []


def get_m49_filter(value, use_recipient_country=True):
    """Returns the location filter object based on value."""
    countries = walk(deepcopy(M49_HIERARCHY)[int(value)])
    countries_lower = [c.lower() for c in countries]
    filter_ = Q(locations__country__iso_code__in=countries_lower)
    if use_recipient_country:
        filter_ = (Q(recipient_countries__country__in=countries) | filter_)
    return filter_


def filter_m49(queryset, value):
    """Filters countries from the m49 list, for projects."""
    if not value:
        return queryset
    else:
        return queryset.filter(get_m49_filter(value))


def filter_m49_orgs(queryset, value):
    """Filters countries from the m49 list, for projects."""
    if not value:
        return queryset
    countries = walk(deepcopy(M49_HIERARCHY)[int(value)])
    return queryset.filter(locations__iati_country__in=countries)


def filter_title_or_subtitle(queryset, value):
    """Filters projects based on whether title or subtitle contains value."""
    if not value:
        return queryset
    return queryset.filter(Q(title__icontains=value) | Q(subtitle__icontains=value))


def get_id_for_iso(i):
    """From an iso_code e.g. 'SE' get the identifier.

    NOTE: If i is already an id, the parent id of the given id is returned!

    """
    i = [k for k, v in M49_HIERARCHY.items() if i in v]
    return None if not i else i.pop()


def get_location_hierarchy(location, locations=None):
    """Return the location > parent > ... > continent hierarchy for a location."""
    if locations is None:
        locations = [location]
    # FIXME: Actually returns parent id, when location is already an id!
    loc = get_id_for_iso(location)
    if isinstance(loc, str) or loc == 1 or loc is None:
        return locations
    else:
        locations.append(loc)
        return get_location_hierarchy(loc, locations)


def location_choices(qs):
    """Return a filterd list of locations from M49_CODES based on queryset."""

    country_ids = get_country_ids(qs)

    location_ids = {
        str(location)
        for country_id in country_ids
        for location in get_location_hierarchy(country_id)
    }

    # Add World to locations
    location_ids.add("")

    return [code for code in M49_CODES if code[0] in location_ids]


def get_country_ids(qs):
    """Return country ids for locations associated with the queryset items."""

    country_ids = get_location_country_ids(qs)
    if qs.model is Project:
        country_ids = get_recipient_country_ids(qs) + country_ids

    return set(country_ids)


def get_recipient_country_ids(projects):
    """Return countries based on recipient country of projects."""
    countries = RecipientCountry.objects.filter(project__in=projects).values('country')
    return [get_id_for_iso(country['country'].upper()) for country in countries]


def get_location_country_ids(qs):
    """Return countries for locations associated with objects in the queryset."""

    if qs.model is Project:
        location_model = ProjectLocation

    elif qs.model is ProjectUpdate:
        location_model = ProjectUpdateLocation

    elif qs.model is Organisation:
        location_model = OrganisationLocation

    locations_qs = location_model.objects.filter(
        location_target__in=qs
    ).values(
        'country__iso_code',
    ).order_by('country__iso_code').distinct()

    return [
        get_id_for_iso(location['country__iso_code'].upper())
        for location in locations_qs if location['country__iso_code']
    ]


def build_choices(qs):
    """Build choices from queryset and add an All option"""
    return [('', _('All'))] + list(qs.values_list('id', 'name', flat=False))

1	# -- coding: utf-8 --
2
3	"""Akvo RSR is covered by the GNU Affero General Public License.	×
4	See more details in the license.txt file located at the root folder of the
5	Akvo RSR module. For additional details on the GNU license please
6	see < http://www.gnu.org/licenses/agpl.html >.
7	"""
8
9	from copy import deepcopy	×
10
11	from django.db.models import Q	×
12	from django.utils.translation import ugettext_lazy as _	×
13
14	from akvo.codelists.store.default_codelists import SECTOR_CATEGORY	×
15	from akvo.utils import codelist_choices	×
16	from .models import (Organisation, OrganisationLocation,	×
17	Project, ProjectLocation, ProjectUpdate, ProjectUpdateLocation,
18	RecipientCountry)
19	from .m49 import M49_CODES, M49_HIERARCHY	×
20
21	ANY_CHOICE = (('', _('All')), )	×
22
23
24	def sectors():	×
25	sectors_list = []	×
26	for sector in codelist_choices(SECTOR_CATEGORY):	×
27	sectors_list.append(sector)	×
28	return sectors_list	×
29
30
31	def get_orgs():	×
32	orgs = list(Organisation.objects.all().values_list(	×
33	'id', 'name', flat=False))
34	return [('', _('All'))] + orgs	×
35
36
37	def remove_empty_querydict_items(request_get):	×
38	# querydicts are immutable
39	getvars = request_get.copy()	×
40	for k, v in list(getvars.items()):	×
41	if not v:	×
42	getvars.pop(k, None)	×
43	return getvars	×
44
45
46	def walk(node):	×
47	"""Walks the m49 tree and return countries"""
48
49	if isinstance(node, str):	×
50	return [node, ]	×
51	elif isinstance(node, int):	×
52	return walk(deepcopy(M49_HIERARCHY)[node])	×
53	else:
54	return (walk(node.pop()) + walk(node)) if node else []	×
55
56
57	def get_m49_filter(value, use_recipient_country=True):	×
58	"""Returns the location filter object based on value."""
59	countries = walk(deepcopy(M49_HIERARCHY)[int(value)])	×
60	countries_lower = [c.lower() for c in countries]	×
61	filter_ = Q(locations__country__iso_code__in=countries_lower)	×
62	if use_recipient_country:	×
63	filter_ = (Q(recipient_countries__country__in=countries) \| filter_)	×
64	return filter_	×
65
66
67	def filter_m49(queryset, value):	×
68	"""Filters countries from the m49 list, for projects."""
69	if not value:	×
70	return queryset	×
71	else:
72	return queryset.filter(get_m49_filter(value))	×
73
74
75	def filter_m49_orgs(queryset, value):	×
76	"""Filters countries from the m49 list, for projects."""
77	if not value:	×
78	return queryset	×
79	countries = walk(deepcopy(M49_HIERARCHY)[int(value)])	×
80	return queryset.filter(locations__iati_country__in=countries)	×
81
82
83	def filter_title_or_subtitle(queryset, value):	×
84	"""Filters projects based on whether title or subtitle contains value."""
85	if not value:	×
86	return queryset	×
87	return queryset.filter(Q(title__icontains=value) \| Q(subtitle__icontains=value))	×
88
89
90	def get_id_for_iso(i):	×
91	"""From an iso_code e.g. 'SE' get the identifier.
92
93	NOTE: If i is already an id, the parent id of the given id is returned!
94
95	"""
96	i = [k for k, v in M49_HIERARCHY.items() if i in v]	×
97	return None if not i else i.pop()	×
98
99
100	def get_location_hierarchy(location, locations=None):	×
101	"""Return the location > parent > ... > continent hierarchy for a location."""
102	if locations is None:	×
103	locations = [location]	×
104	# FIXME: Actually returns parent id, when location is already an id!
105	loc = get_id_for_iso(location)	×
106	if isinstance(loc, str) or loc == 1 or loc is None:	×
107	return locations	×
108	else:
109	locations.append(loc)	×
110	return get_location_hierarchy(loc, locations)	×
111
112
113	def location_choices(qs):	×
114	"""Return a filterd list of locations from M49_CODES based on queryset."""
115
116	country_ids = get_country_ids(qs)	×
117
118	location_ids = {	×
119	str(location)
120	for country_id in country_ids
121	for location in get_location_hierarchy(country_id)
122	}
123
124	# Add World to locations
125	location_ids.add("")	×
126
127	return [code for code in M49_CODES if code[0] in location_ids]	×
128
129
130	def get_country_ids(qs):	×
131	"""Return country ids for locations associated with the queryset items."""
132
133	country_ids = get_location_country_ids(qs)	×
134	if qs.model is Project:	×
135	country_ids = get_recipient_country_ids(qs) + country_ids	×
136
137	return set(country_ids)	×
138
139
140	def get_recipient_country_ids(projects):	×
141	"""Return countries based on recipient country of projects."""
142	countries = RecipientCountry.objects.filter(project__in=projects).values('country')	×
143	return [get_id_for_iso(country['country'].upper()) for country in countries]	×
144
145
146	def get_location_country_ids(qs):	×
147	"""Return countries for locations associated with objects in the queryset."""
148
149	if qs.model is Project:	×
150	location_model = ProjectLocation	×
151
152	elif qs.model is ProjectUpdate:	×
153	location_model = ProjectUpdateLocation	×
154
155	elif qs.model is Organisation:	×
156	location_model = OrganisationLocation	×
157
158	locations_qs = location_model.objects.filter(	×
159	location_target__in=qs
160	).values(
161	'country__iso_code',
162	).order_by('country__iso_code').distinct()
163
164	return [	×
165	get_id_for_iso(location['country__iso_code'].upper())
166	for location in locations_qs if location['country__iso_code']
167	]
168
169
170	def build_choices(qs):	×
171	"""Build choices from queryset and add an All option"""
172	return [('', _('All'))] + list(qs.values_list('id', 'name', flat=False))	×

akvo / akvo-rsr / 36d6dc01-6726-4ee7-a9dd-ce35c506949e

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous