3643375715

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/kobotoolbox/kpi/commit/<a class=hub.com/kobotoolbox/kpi/commit/<a class="double-link" href="https://git"><a class=hub.com/kobotoolbox/kpi/commit/<a class="double-link" href="https://git"><a class=hub.com/kobotoolbox/kpi/commit/<a class="double-link" href="https://git"><a class=hub.com/kobotoolbox/kpi/commit/874b39f941a2f089bd2173c761202c61537d769d">874b39f94">&lt;a href=&quot;https://github.com/kobotoolbox/kpi/commit/</a><a class="double-link" href="https://github.com/kobotoolbox/kpi/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/kobotoolbox/kpi/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/kobotoolbox/kpi/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/kobotoolbox/kpi/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/kobotoolbox/kpi/commit/874b39f941a2f089bd2173c761202c61537d769d">874b39f94</a><a href="https://github.com/kobotoolbox/kpi/commit/874b39f941a2f089bd2173c761202c61537d769d">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/kobotoolbox/kpi/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/kobotoolbox/kpi/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/kobotoolbox/kpi/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/kobotoolbox/kpi/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/kobotoolbox/kpi/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/kobotoolbox/kpi/commit/874b39f941a2f089bd2173c761202c61537d769d&quot;&gt;874b39f94&lt;/a&gt;&lt;a href=&quot;https://github.com/kobotoolbox/kpi/commit/874b39f941a2f089bd2173c761202c61537d769d&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/kobotoolbox/kpi/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/kobotoolbox/kpi/commit/&amp;amp;lt;a class=&amp;amp;quot;double-link&am... (continued)

Pull Request Pull Request #4182: Custom Views: integrate endpoints into front-end

Run Details

2515 of 4255 branches covered (59.11%)

409 of 409 new or added lines in 19 files covered. (100.0%)

11113 of 14727 relevant lines covered (75.46%)

1.51 hits per line

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

27.5

/kpi/utils/custom_project_exports.py

import csv
from io import StringIO
from typing import Union, List, Tuple

from django.conf import settings
from django.contrib.auth.models import User
from django.db.models import CharField, Count, F, Q
from django.db.models.functions import Cast
from django.db.models.query import QuerySet

from kpi.models.asset import Asset
from kpi.utils.custom_projects import (
    get_region_for_view,
)


ASSET_FIELDS = (
    'id',
    'uid',
    'name',
    'asset_type',
    'date_modified',
    'date_created',
    'owner',
    'owner__username',
    'owner__email',
    'owner__name',
    'owner__organization',
    'deployment__active',
    'deployment__submission_count',
)
SETTINGS = 'settings'
SETTINGS_FIELDS = (
    'country',
    'sector',
    'description',
)

USER_FIELDS = (
    'username',
    'is_superuser',
    'is_staff',
    'date_joined_str',
    'last_login_str',
    'is_active',
    'email',
    'mfa_is_active',
    'asset_count',
)
METADATA = 'metadata'
METADATA_FIELDS = (
    'name',
    'gender',
    'sector',
    'country',
    'city',
    'bio',
    'organization',
    'require_auth',
    'primarySector',
    'organization_website',
    'twitter',
    'linkedin',
    'instagram',
    'metadata',
)
CONFIG = {
    'assets': {
        'queryset': Asset.objects.all(),
        'q_term': 'settings__country',
        'key': SETTINGS,
        'columns': ASSET_FIELDS + SETTINGS_FIELDS,
    },
    'users': {
        'queryset': User.objects.all(),
        'q_term': 'extra_details__data__country',
        'key': METADATA,
        'columns': USER_FIELDS + METADATA_FIELDS,
    },
}


def flatten_settings_inplace(settings: dict):
    for k, v in settings.items():
        if isinstance(v, list) and v:
            settings[k] = ', '.join([item['value'] for item in v])
        if isinstance(v, dict) and 'value' in v:
            settings[k] = v['value']


def get_row_value(row: dict, col: str) -> Union[str, int, float, bool, None]:
    val = row.get(col, '')
    # remove any new lines from text
    if isinstance(val, str):
        val = val.replace('\n', '')
    return val


def get_q(countries: List[str], export_type: str) -> QuerySet:
    q_term = CONFIG[export_type]['q_term']

    if '*' in countries:
        return Q()

    q = Q(**{f'{q_term}_in': countries})
    for country in countries:
        q |= Q(**{f'{q_term}__contains': [{'value': country}]})

    return q


def get_data(filtered_queryset: QuerySet, export_type: str) -> QuerySet:
    if export_type == 'assets':
        vals = ASSET_FIELDS + (SETTINGS,)
        data = filtered_queryset.annotate(
            owner__name=F('owner__extra_details__data__name'),
            owner__organization=F('owner__extra_details__data__organization'),
            deployment__active=F('_deployment_data__active'),
            deployment__submission_count=F(
                '_deployment_data__num_of_submissions'
            ),
        )
    else:
        vals = USER_FIELDS + (METADATA,)
        data = filtered_queryset.exclude(
            pk=settings.ANONYMOUS_USER_ID
        ).annotate(
            mfa_is_active=F('mfa_methods__is_active'),
            metadata=F('extra_details__data'),
            date_joined_str=Cast('date_joined', CharField()),
            last_login_str=Cast('last_login', CharField()),
            asset_count=Count('assets'),
        )

    return data.values(*vals).order_by('id')


def create_custom_project_export(
        export_type: str, username: str, uid: str
) -> StringIO:
    config = CONFIG[export_type]
    region_for_view = get_region_for_view(uid)

    q = get_q(region_for_view, export_type)
    filtered_queryset = config['queryset'].filter(q)
    data = get_data(filtered_queryset, export_type)

    buff = StringIO()
    writer = csv.writer(buff)
    writer.writerow(config['columns'])
    for row in data:
        items = row.pop(config['key'], {}) or {}
        flatten_settings_inplace(items)
        row.update(items)
        flat_row = [get_row_value(row, col) for col in config['columns']]
        writer.writerow(flat_row)

    buff.seek(0)
    return buff

1	import csv	2✔
2	from io import StringIO	2✔
3	from typing import Union, List, Tuple	2✔
4
5	from django.conf import settings	2✔
6	from django.contrib.auth.models import User	2✔
7	from django.db.models import CharField, Count, F, Q	2✔
8	from django.db.models.functions import Cast	2✔
9	from django.db.models.query import QuerySet	2✔
10
11	from kpi.models.asset import Asset	2✔
12	from kpi.utils.custom_projects import (	2✔
13	get_region_for_view,
14	)
15
16
17	ASSET_FIELDS = (	2✔
18	'id',
19	'uid',
20	'name',
21	'asset_type',
22	'date_modified',
23	'date_created',
24	'owner',
25	'owner__username',
26	'owner__email',
27	'owner__name',
28	'owner__organization',
29	'deployment__active',
30	'deployment__submission_count',
31	)
32	SETTINGS = 'settings'	2✔
33	SETTINGS_FIELDS = (	2✔
34	'country',
35	'sector',
36	'description',
37	)
38
39	USER_FIELDS = (	2✔
40	'username',
41	'is_superuser',
42	'is_staff',
43	'date_joined_str',
44	'last_login_str',
45	'is_active',
46	'email',
47	'mfa_is_active',
48	'asset_count',
49	)
50	METADATA = 'metadata'	2✔
51	METADATA_FIELDS = (	2✔
52	'name',
53	'gender',
54	'sector',
55	'country',
56	'city',
57	'bio',
58	'organization',
59	'require_auth',
60	'primarySector',
61	'organization_website',
62	'twitter',
63	'linkedin',
64	'instagram',
65	'metadata',
66	)
67	CONFIG = {	2✔
68	'assets': {
69	'queryset': Asset.objects.all(),
70	'q_term': 'settings__country',
71	'key': SETTINGS,
72	'columns': ASSET_FIELDS + SETTINGS_FIELDS,
73	},
74	'users': {
75	'queryset': User.objects.all(),
76	'q_term': 'extra_details__data__country',
77	'key': METADATA,
78	'columns': USER_FIELDS + METADATA_FIELDS,
79	},
80	}
81
82
83	def flatten_settings_inplace(settings: dict):	2✔
84	for k, v in settings.items():	×
85	if isinstance(v, list) and v:	×
86	settings[k] = ', '.join([item['value'] for item in v])	×
87	if isinstance(v, dict) and 'value' in v:	×
88	settings[k] = v['value']	×
89
90
91	def get_row_value(row: dict, col: str) -> Union[str, int, float, bool, None]:	2✔
92	val = row.get(col, '')	×
93	# remove any new lines from text
94	if isinstance(val, str):	×
95	val = val.replace('\n', '')	×
96	return val	×
97
98
99	def get_q(countries: List[str], export_type: str) -> QuerySet:	2✔
100	q_term = CONFIG[export_type]['q_term']	×
101
102	if '*' in countries:	×
103	return Q()	×
104
105	q = Q(**{f'{q_term}_in': countries})	×
106	for country in countries:	×
107	q \|= Q(**{f'{q_term}__contains': [{'value': country}]})	×
108
109	return q	×
110
111
112	def get_data(filtered_queryset: QuerySet, export_type: str) -> QuerySet:	2✔
113	if export_type == 'assets':	×
114	vals = ASSET_FIELDS + (SETTINGS,)	×
115	data = filtered_queryset.annotate(	×
116	owner__name=F('owner__extra_details__data__name'),
117	owner__organization=F('owner__extra_details__data__organization'),
118	deployment__active=F('_deployment_data__active'),
119	deployment__submission_count=F(
120	'_deployment_data__num_of_submissions'
121	),
122	)
123	else:
124	vals = USER_FIELDS + (METADATA,)	×
125	data = filtered_queryset.exclude(	×
126	pk=settings.ANONYMOUS_USER_ID
127	).annotate(
128	mfa_is_active=F('mfa_methods__is_active'),
129	metadata=F('extra_details__data'),
130	date_joined_str=Cast('date_joined', CharField()),
131	last_login_str=Cast('last_login', CharField()),
132	asset_count=Count('assets'),
133	)
134
135	return data.values(*vals).order_by('id')	×
136
137
138	def create_custom_project_export(	2✔
139	export_type: str, username: str, uid: str
140	) -> StringIO:
141	config = CONFIG[export_type]	×
142	region_for_view = get_region_for_view(uid)	×
143
144	q = get_q(region_for_view, export_type)	×
145	filtered_queryset = config['queryset'].filter(q)	×
146	data = get_data(filtered_queryset, export_type)	×
147
148	buff = StringIO()	×
149	writer = csv.writer(buff)	×
150	writer.writerow(config['columns'])	×
151	for row in data:	×
152	items = row.pop(config['key'], {}) or {}	×
153	flatten_settings_inplace(items)	×
154	row.update(items)	×
155	flat_row = [get_row_value(row, col) for col in config['columns']]	×
156	writer.writerow(flat_row)	×
157
158	buff.seek(0)	×
159	return buff	×

kobotoolbox / kpi / 3643375715

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