#1314

Committed 24 Apr 2024 10:13PM UTC coverage: 79.923% (+0.2%) from 79.77%

Build # #1314

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Merge pull request #1405 from akvo/feature/1404-data-generator-jobs

Feature/1404 data generator jobs

Run Details

2571 of 3393 branches covered (75.77%)

Branch coverage included in aggregate %.

6151 of 7520 relevant lines covered (81.8%)

0.82 hits per line

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

45.87

backend/api/v1/v1_jobs/job.py

import logging
import os
from django_q.models import Task

import pandas as pd
from django.utils import timezone
from django_q.tasks import async_task
from django.db.models import Subquery, Max
from api.v1.v1_jobs.administrations_bulk_upload import (
    seed_administration_data,
    validate_administrations_bulk_upload,
)
from utils.upload_entities import (
    validate_entity_file,
    validate_entity_data,
)
from api.v1.v1_profile.constants import UserRoleTypes
from api.v1.v1_forms.models import Forms
from api.v1.v1_forms.constants import SubmissionTypes
from api.v1.v1_jobs.constants import JobStatus, JobTypes
# from api.v1.v1_jobs.functions import HText
from api.v1.v1_jobs.models import Jobs
from api.v1.v1_jobs.seed_data import seed_excel_data
from api.v1.v1_jobs.validate_upload import validate
from api.v1.v1_profile.models import (
    Administration,
    EntityData
)
from api.v1.v1_users.models import SystemUser
from utils import storage
from utils.email_helper import send_email, EmailTypes
from utils.export_form import (
    generate_definition_sheet,
    rearrange_definition_columns
)
from utils.functions import update_date_time_format
from utils.storage import upload
from utils.custom_generator import generate_sqlite

logger = logging.getLogger(__name__)


def download(
    form: Forms,
    administration_ids,
    download_type="all",
    submission_type=None
):
    filter_data = {}
    if administration_ids:
        filter_data['administration_id__in'] = administration_ids
    if submission_type:
        filter_data['submission_type'] = submission_type
    else:
        filter_data['submission_type__in'] = [
            SubmissionTypes.registration,
            SubmissionTypes.monitoring
        ]
    data = form.form_form_data.filter(**filter_data)
    if download_type == "recent":
        latest_per_uuid = data.values('uuid').annotate(
            latest_created=Max('created')
        ).values('latest_created')
        data = data.filter(
            created__in=Subquery(latest_per_uuid)
        )
    return [d.to_data_frame for d in data.order_by('id')]


def job_generate_download(job_id, **kwargs):
    job = Jobs.objects.get(pk=job_id)
    file_path = './tmp/{0}'.format(job.result)
    if os.path.exists(file_path):
        os.remove(file_path)
    administration_ids = False
    administration_name = "All Administration Level"
    if kwargs.get('administration'):
        administration = Administration.objects.get(
            pk=kwargs.get('administration'))
        if administration.path:
            filter_path = '{0}{1}.'.format(administration.path,
                                           administration.id)
        else:
            filter_path = f"{administration.id}."
        administration_ids = list(
            Administration.objects.filter(
                path__startswith=filter_path).values_list('id',
                                                          flat=True))
        administration_ids.append(administration.id)

        administration_name = list(
            Administration.objects.filter(
                path__startswith=filter_path).values_list('name',
                                                          flat=True))
    form = Forms.objects.get(pk=job.info.get('form_id'))
    download_type = kwargs.get('download_type')
    submission_type = kwargs.get('submission_type')
    data = download(
        form=form,
        administration_ids=administration_ids,
        download_type=download_type,
        submission_type=submission_type,
    )
    df = pd.DataFrame(data)
    col_names = rearrange_definition_columns(list(df))
    df = df[col_names]
    writer = pd.ExcelWriter(file_path, engine='xlsxwriter')
    df.to_excel(writer, sheet_name='data', index=False)
    generate_definition_sheet(form=form, writer=writer)
    context = [{
        "context": "Form Name",
        "value": form.name
    }, {
        "context": "Download Date",
        "value": update_date_time_format(job.created)
    }, {
        "context": "Administration",
        "value": ','.join(administration_name) if isinstance(
            administration_name, list) else administration_name
    }]

    context = pd.DataFrame(context).groupby(["context", "value"],
                                            sort=False).first()
    context.to_excel(writer, sheet_name='context', startrow=0, header=False)
    workbook = writer.book
    worksheet = writer.sheets['context']
    f = workbook.add_format({
        'align': 'left',
        'bold': False,
        'border': 0,
    })
    worksheet.set_column('A:A', 20, f)
    worksheet.set_column('B:B', 30, f)
    merge_format = workbook.add_format({
        'bold': True,
        'border': 1,
        'align': 'center',
        'valign': 'vcenter',
        'fg_color': '#45add9',
        'color': '#ffffff',
    })
    worksheet.merge_range('A1:B1', 'Context', merge_format)
    writer.save()
    url = upload(file=file_path, folder='download')
    return url


def job_generate_download_result(task):
    job = Jobs.objects.get(task_id=task.id)
    job.attempt = job.attempt + 1
    if task.success:
        job.status = JobStatus.done
        job.available = timezone.now()
    else:
        job.status = JobStatus.failed
    job.save()


def seed_data_job(job_id):
    try:
        job = Jobs.objects.get(pk=job_id)
        seed_excel_data(job)
    except (ValueError, OSError, NameError, TypeError):
        print("Error seed data job")
        return False
    except Exception as unknown_error:
        print("Unknown error seed data job", unknown_error)
        return False
    return True


def seed_data_job_result(task):
    job = Jobs.objects.get(task_id=task.id)
    job.attempt = job.attempt + 1
    is_super_admin = job.user.user_access.role == UserRoleTypes.super_admin
    if task.result:
        job.status = JobStatus.done
        job.available = timezone.now()
        form_id = job.info.get("form")
        form = Forms.objects.filter(pk=int(form_id)).first()
        file = job.info.get("file")
        storage.download(f"upload/{file}")
        df = pd.read_excel(f"./tmp/{file}", sheet_name='data')
        subject = "New Data Uploaded" if is_super_admin \
            else 'New Request @{0}'.format(job.user.get_full_name())
        data = {
            'subject': subject,
            'title': "New Data Submission",
            'send_to': [job.user.email],
            'listing': [{
                'name': "Upload Date",
                'value': job.created.strftime("%m-%d-%Y, %H:%M:%S"),
            }, {
                'name': "Questionnaire",
                'value': form.name
            }, {
                'name': "Number of Records",
                'value': df.shape[0]
            }],
            'is_super_admin': is_super_admin
        }
        send_email(context=data, type=EmailTypes.new_request)
    else:
        job.status = JobStatus.failed
    job.save()


def validate_excel(job_id):
    job = Jobs.objects.get(pk=job_id)
    storage.download(f"upload/{job.info.get('file')}")
    data = validate(job.info.get('form'), job.info.get('administration'),
                    f"./tmp/{job.info.get('file')}")

    if len(data):
        form_id = job.info.get("form")
        form = Forms.objects.filter(pk=int(form_id)).first()
        file = job.info.get("file")
        df = pd.read_excel(f"./tmp/{file}", sheet_name='data')
        error_list = pd.DataFrame(data)
        error_list = error_list[list(
            filter(lambda x: x != "error", list(error_list)))]
        error_file = f"./tmp/error-{job_id}.csv"
        error_list.to_csv(error_file, index=False)
        data = {
            'send_to': [job.user.email],
            'listing': [{
                'name': "Upload Date",
                'value': job.created.strftime("%m-%d-%Y, %H:%M:%S"),
            }, {
                'name': "Questionnaire",
                'value': form.name
            }, {
                'name': "Number of Records",
                'value': df.shape[0]
            }],
        }
        send_email(context=data,
                   type=EmailTypes.upload_error,
                   path=error_file,
                   content_type='text/csv')
        return False
    return True


def validate_excel_result(task):
    job = Jobs.objects.get(task_id=task.id)
    job.attempt = job.attempt + 1
    job_info = job.info
    if task.result:
        job.status = JobStatus.done
        job.available = timezone.now()
        job.save()
        job_info.update({'ref_job_id': job.id})
        new_job = Jobs.objects.create(
            result=job.info.get('file'),
            type=JobTypes.seed_data,
            status=JobStatus.on_progress,
            user=job.user,
            info=job_info
        )
        task_id = async_task(
            'api.v1.v1_jobs.job.seed_data_job', new_job.id,
            hook='api.v1.v1_jobs.job.seed_data_job_result')
        new_job.task_id = task_id
        new_job.save()
    else:
        job.status = JobStatus.failed
        job.save()


def handle_administrations_bulk_upload(filename, user_id, upload_time):
    user = SystemUser.objects.get(id=user_id)
    storage.download(f"upload/{filename}")
    file_path = f"./tmp/{filename}"
    errors = validate_administrations_bulk_upload(file_path)
    df = pd.read_excel(file_path, sheet_name='data')
    email_context = {
        'send_to': [user.email],
        'listing': [
            {
                'name': 'Upload Date',
                'value': upload_time.strftime('%m-%d-%Y, %H:%M:%S'),
            },
            {
                'name': 'Questionnaire',
                'value': 'Administrative List',
            },
            {
                'name': "Number of Records",
                'value': df.shape[0]
            },
        ]
    }
    if len(errors):
        logger.error(errors)
        error_file = (
            "./tmp/administration-error-"
            f"{upload_time.strftime('%Y%m%d%H%M%S')}-{user.id}.csv"
        )
        error_list = pd.DataFrame(errors)
        error_list.to_csv(error_file, index=False)
        send_email(context=email_context,
                   type=EmailTypes.upload_error,
                   path=error_file,
                   content_type='text/csv')
        return
    seed_administration_data(file_path)
    generate_sqlite(Administration)
    send_email(context=email_context, type=EmailTypes.administration_upload)


def handle_master_data_bulk_upload_failure(task: Task):
    if task.success:
        return
    logger.error({
        'error': 'Failed running background job',
        'id': task.id,
        'name': task.name,
        'started': task.started,
        'stopped': task.stopped,
        'args': task.args,
        'kwargs': task.kwargs,
        'body': task.result,
    })


def handle_entities_error_upload(
        errors: list,
        email_context: dict,
        user: SystemUser,
        upload_time: timezone.datetime
):
    logger.error(errors)
    error_file = (
        "./tmp/entity-error-"
        f"{upload_time.strftime('%Y%m%d%H%M%S')}-{user.id}.csv"
    )
    error_list = pd.DataFrame(errors)
    error_list.to_csv(error_file, index=False)
    send_email(
        context=email_context,
        type=EmailTypes.upload_error,
        path=error_file,
        content_type='text/csv'
    )


def handle_entities_bulk_upload(filename, user_id, upload_time):
    user = SystemUser.objects.get(id=user_id)
    storage.download(f"upload/{filename}")
    file_path = f"./tmp/{filename}"
    errors = validate_entity_file(file_path)
    email_context = {
        'send_to': [user.email],
        'listing': [{
            'name': 'Upload Date',
            'value': upload_time.strftime('%m-%d-%Y, %H:%M:%S'),
        }, {
            'name': 'Questionnaire',
            'value': 'Entity List',
        }]
    }
    if len(errors):
        handle_entities_error_upload(errors, email_context, user, upload_time)
        return
    errors = validate_entity_data(file_path)
    if len(errors):
        handle_entities_error_upload(errors, email_context, user, upload_time)
        return
    generate_sqlite(EntityData)

1	import logging	1✔
2	import os	1✔
3	from django_q.models import Task	1✔
4
5	import pandas as pd	1✔
6	from django.utils import timezone	1✔
7	from django_q.tasks import async_task	1✔
8	from django.db.models import Subquery, Max	1✔
9	from api.v1.v1_jobs.administrations_bulk_upload import (	1✔
10	seed_administration_data,
11	validate_administrations_bulk_upload,
12	)
13	from utils.upload_entities import (	1✔
14	validate_entity_file,
15	validate_entity_data,
16	)
17	from api.v1.v1_profile.constants import UserRoleTypes	1✔
18	from api.v1.v1_forms.models import Forms	1✔
19	from api.v1.v1_forms.constants import SubmissionTypes	1✔
20	from api.v1.v1_jobs.constants import JobStatus, JobTypes	1✔
21	# from api.v1.v1_jobs.functions import HText
22	from api.v1.v1_jobs.models import Jobs	1✔
23	from api.v1.v1_jobs.seed_data import seed_excel_data	1✔
24	from api.v1.v1_jobs.validate_upload import validate	1✔
25	from api.v1.v1_profile.models import (	1✔
26	Administration,
27	EntityData
28	)
29	from api.v1.v1_users.models import SystemUser	1✔
30	from utils import storage	1✔
31	from utils.email_helper import send_email, EmailTypes	1✔
32	from utils.export_form import (	1✔
33	generate_definition_sheet,
34	rearrange_definition_columns
35	)
36	from utils.functions import update_date_time_format	1✔
37	from utils.storage import upload	1✔
38	from utils.custom_generator import generate_sqlite	1✔
39
40	logger = logging.getLogger(__name__)	1✔
41
42
43	def download(	1✔
44	form: Forms,
45	administration_ids,
46	download_type="all",
47	submission_type=None
48	):
49	filter_data = {}	1✔
50	if administration_ids:	1✔
51	filter_data['administration_id__in'] = administration_ids	1✔
52	if submission_type:	1✔
53	filter_data['submission_type'] = submission_type	1✔
54	else:
55	filter_data['submission_type__in'] = [	1✔
56	SubmissionTypes.registration,
57	SubmissionTypes.monitoring
58	]
59	data = form.form_form_data.filter(**filter_data)	1✔
60	if download_type == "recent":	1✔
61	latest_per_uuid = data.values('uuid').annotate(	1✔
62	latest_created=Max('created')
63	).values('latest_created')
64	data = data.filter(	1✔
65	created__in=Subquery(latest_per_uuid)
66	)
67	return [d.to_data_frame for d in data.order_by('id')]	1✔
68
69
70	def job_generate_download(job_id, **kwargs):	1✔
71	job = Jobs.objects.get(pk=job_id)	1✔
72	file_path = './tmp/{0}'.format(job.result)	1✔
73	if os.path.exists(file_path):	1!
74	os.remove(file_path)	×
75	administration_ids = False	1✔
76	administration_name = "All Administration Level"	1✔
77	if kwargs.get('administration'):	1!
78	administration = Administration.objects.get(	×
79	pk=kwargs.get('administration'))
80	if administration.path:	×
81	filter_path = '{0}{1}.'.format(administration.path,	×
82	administration.id)
83	else:
84	filter_path = f"{administration.id}."	×
85	administration_ids = list(	×
86	Administration.objects.filter(
87	path__startswith=filter_path).values_list('id',
88	flat=True))
89	administration_ids.append(administration.id)	×
90
91	administration_name = list(	×
92	Administration.objects.filter(
93	path__startswith=filter_path).values_list('name',
94	flat=True))
95	form = Forms.objects.get(pk=job.info.get('form_id'))	1✔
96	download_type = kwargs.get('download_type')	1✔
97	submission_type = kwargs.get('submission_type')	1✔
98	data = download(	1✔
99	form=form,
100	administration_ids=administration_ids,
101	download_type=download_type,
102	submission_type=submission_type,
103	)
104	df = pd.DataFrame(data)	1✔
105	col_names = rearrange_definition_columns(list(df))	1✔
106	df = df[col_names]	1✔
107	writer = pd.ExcelWriter(file_path, engine='xlsxwriter')	1✔
108	df.to_excel(writer, sheet_name='data', index=False)	1✔
109	generate_definition_sheet(form=form, writer=writer)	1✔
110	context = [{	1✔
111	"context": "Form Name",
112	"value": form.name
113	}, {
114	"context": "Download Date",
115	"value": update_date_time_format(job.created)
116	}, {
117	"context": "Administration",
118	"value": ','.join(administration_name) if isinstance(
119	administration_name, list) else administration_name
120	}]
121
122	context = pd.DataFrame(context).groupby(["context", "value"],	1✔
123	sort=False).first()
124	context.to_excel(writer, sheet_name='context', startrow=0, header=False)	1✔
125	workbook = writer.book	1✔
126	worksheet = writer.sheets['context']	1✔
127	f = workbook.add_format({	1✔
128	'align': 'left',
129	'bold': False,
130	'border': 0,
131	})
132	worksheet.set_column('A:A', 20, f)	1✔
133	worksheet.set_column('B:B', 30, f)	1✔
134	merge_format = workbook.add_format({	1✔
135	'bold': True,
136	'border': 1,
137	'align': 'center',
138	'valign': 'vcenter',
139	'fg_color': '#45add9',
140	'color': '#ffffff',
141	})
142	worksheet.merge_range('A1:B1', 'Context', merge_format)	1✔
143	writer.save()	1✔
144	url = upload(file=file_path, folder='download')	1✔
145	return url	1✔
146
147
148	def job_generate_download_result(task):	1✔
149	job = Jobs.objects.get(task_id=task.id)	×
150	job.attempt = job.attempt + 1	×
151	if task.success:	×
152	job.status = JobStatus.done	×
153	job.available = timezone.now()	×
154	else:
155	job.status = JobStatus.failed	×
156	job.save()	×
157
158
159	def seed_data_job(job_id):	1✔
160	try:	×
161	job = Jobs.objects.get(pk=job_id)	×
162	seed_excel_data(job)	×
163	except (ValueError, OSError, NameError, TypeError):	×
164	print("Error seed data job")	×
165	return False	×
166	except Exception as unknown_error:	×
167	print("Unknown error seed data job", unknown_error)	×
168	return False	×
169	return True	×
170
171
172	def seed_data_job_result(task):	1✔
173	job = Jobs.objects.get(task_id=task.id)	×
174	job.attempt = job.attempt + 1	×
175	is_super_admin = job.user.user_access.role == UserRoleTypes.super_admin	×
176	if task.result:	×
177	job.status = JobStatus.done	×
178	job.available = timezone.now()	×
179	form_id = job.info.get("form")	×
180	form = Forms.objects.filter(pk=int(form_id)).first()	×
181	file = job.info.get("file")	×
182	storage.download(f"upload/{file}")	×
183	df = pd.read_excel(f"./tmp/{file}", sheet_name='data')	×
184	subject = "New Data Uploaded" if is_super_admin \	×
185	else 'New Request @{0}'.format(job.user.get_full_name())
186	data = {	×
187	'subject': subject,
188	'title': "New Data Submission",
189	'send_to': [job.user.email],
190	'listing': [{
191	'name': "Upload Date",
192	'value': job.created.strftime("%m-%d-%Y, %H:%M:%S"),
193	}, {
194	'name': "Questionnaire",
195	'value': form.name
196	}, {
197	'name': "Number of Records",
198	'value': df.shape[0]
199	}],
200	'is_super_admin': is_super_admin
201	}
202	send_email(context=data, type=EmailTypes.new_request)	×
203	else:
204	job.status = JobStatus.failed	×
205	job.save()	×
206
207
208	def validate_excel(job_id):	1✔
209	job = Jobs.objects.get(pk=job_id)	×
210	storage.download(f"upload/{job.info.get('file')}")	×
211	data = validate(job.info.get('form'), job.info.get('administration'),	×
212	f"./tmp/{job.info.get('file')}")
213
214	if len(data):	×
215	form_id = job.info.get("form")	×
216	form = Forms.objects.filter(pk=int(form_id)).first()	×
217	file = job.info.get("file")	×
218	df = pd.read_excel(f"./tmp/{file}", sheet_name='data')	×
219	error_list = pd.DataFrame(data)	×
220	error_list = error_list[list(	×
221	filter(lambda x: x != "error", list(error_list)))]
222	error_file = f"./tmp/error-{job_id}.csv"	×
223	error_list.to_csv(error_file, index=False)	×
224	data = {	×
225	'send_to': [job.user.email],
226	'listing': [{
227	'name': "Upload Date",
228	'value': job.created.strftime("%m-%d-%Y, %H:%M:%S"),
229	}, {
230	'name': "Questionnaire",
231	'value': form.name
232	}, {
233	'name': "Number of Records",
234	'value': df.shape[0]
235	}],
236	}
237	send_email(context=data,	×
238	type=EmailTypes.upload_error,
239	path=error_file,
240	content_type='text/csv')
241	return False	×
242	return True	×
243
244
245	def validate_excel_result(task):	1✔
246	job = Jobs.objects.get(task_id=task.id)	×
247	job.attempt = job.attempt + 1	×
248	job_info = job.info	×
249	if task.result:	×
250	job.status = JobStatus.done	×
251	job.available = timezone.now()	×
252	job.save()	×
253	job_info.update({'ref_job_id': job.id})	×
254	new_job = Jobs.objects.create(	×
255	result=job.info.get('file'),
256	type=JobTypes.seed_data,
257	status=JobStatus.on_progress,
258	user=job.user,
259	info=job_info
260	)
261	task_id = async_task(	×
262	'api.v1.v1_jobs.job.seed_data_job', new_job.id,
263	hook='api.v1.v1_jobs.job.seed_data_job_result')
264	new_job.task_id = task_id	×
265	new_job.save()	×
266	else:
267	job.status = JobStatus.failed	×
268	job.save()	×
269
270
271	def handle_administrations_bulk_upload(filename, user_id, upload_time):	1✔
272	user = SystemUser.objects.get(id=user_id)	×
273	storage.download(f"upload/{filename}")	×
274	file_path = f"./tmp/{filename}"	×
275	errors = validate_administrations_bulk_upload(file_path)	×
276	df = pd.read_excel(file_path, sheet_name='data')	×
277	email_context = {	×
278	'send_to': [user.email],
279	'listing': [
280	{
281	'name': 'Upload Date',
282	'value': upload_time.strftime('%m-%d-%Y, %H:%M:%S'),
283	},
284	{
285	'name': 'Questionnaire',
286	'value': 'Administrative List',
287	},
288	{
289	'name': "Number of Records",
290	'value': df.shape[0]
291	},
292	]
293	}
294	if len(errors):	×
295	logger.error(errors)	×
296	error_file = (	×
297	"./tmp/administration-error-"
298	f"{upload_time.strftime('%Y%m%d%H%M%S')}-{user.id}.csv"
299	)
300	error_list = pd.DataFrame(errors)	×
301	error_list.to_csv(error_file, index=False)	×
302	send_email(context=email_context,	×
303	type=EmailTypes.upload_error,
304	path=error_file,
305	content_type='text/csv')
306	return	×
307	seed_administration_data(file_path)	×
308	generate_sqlite(Administration)	×
309	send_email(context=email_context, type=EmailTypes.administration_upload)	×
310
311
312	def handle_master_data_bulk_upload_failure(task: Task):	1✔
313	if task.success:	×
314	return	×
315	logger.error({	×
316	'error': 'Failed running background job',
317	'id': task.id,
318	'name': task.name,
319	'started': task.started,
320	'stopped': task.stopped,
321	'args': task.args,
322	'kwargs': task.kwargs,
323	'body': task.result,
324	})
325
326
327	def handle_entities_error_upload(	1✔
328	errors: list,
329	email_context: dict,
330	user: SystemUser,
331	upload_time: timezone.datetime
332	):
333	logger.error(errors)	1✔
334	error_file = (	1✔
335	"./tmp/entity-error-"
336	f"{upload_time.strftime('%Y%m%d%H%M%S')}-{user.id}.csv"
337	)
338	error_list = pd.DataFrame(errors)	1✔
339	error_list.to_csv(error_file, index=False)	1✔
340	send_email(	1✔
341	context=email_context,
342	type=EmailTypes.upload_error,
343	path=error_file,
344	content_type='text/csv'
345	)
346
347
348	def handle_entities_bulk_upload(filename, user_id, upload_time):	1✔
349	user = SystemUser.objects.get(id=user_id)	1✔
350	storage.download(f"upload/{filename}")	1✔
351	file_path = f"./tmp/{filename}"	1✔
352	errors = validate_entity_file(file_path)	1✔
353	email_context = {	1✔
354	'send_to': [user.email],
355	'listing': [{
356	'name': 'Upload Date',
357	'value': upload_time.strftime('%m-%d-%Y, %H:%M:%S'),
358	}, {
359	'name': 'Questionnaire',
360	'value': 'Entity List',
361	}]
362	}
363	if len(errors):	1!
364	handle_entities_error_upload(errors, email_context, user, upload_time)	1✔
365	return	1✔
366	errors = validate_entity_data(file_path)	×
367	if len(errors):	×
368	handle_entities_error_upload(errors, email_context, user, upload_time)	×
369	return	×
370	generate_sqlite(EntityData)	×

akvo / rtmis / #1314

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