#1109

Committed 16 Apr 2024 05:26AM UTC coverage: 79.111% (-0.04%) from 79.155%

Build # #1109

Build Type

Pull #1343

coveralls-python

Committed by

web-flow

Commit Message

Merge branch 'develop' into feature/1148-update-sqlite-when-bulk-uploads

Pull Request Pull Request #1343: [#1148] Generate sqlite after administration bulk uploads

Run Details

2426 of 3228 branches covered (75.15%)

Branch coverage included in aggregate %.

5800 of 7170 relevant lines covered (80.89%)

0.81 hits per line

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

21.05

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 api.v1.v1_jobs.administrations_bulk_upload import (
        seed_administration_data,
        validate_administrations_bulk_upload)

from api.v1.v1_profile.constants import UserRoleTypes
from api.v1.v1_forms.models import Forms
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
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
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):
    filter_data = {}
    if administration_ids:
        filter_data['administration_id__in'] = administration_ids
    data = form.form_form_data.filter(**filter_data).order_by('-id')
    return [d.to_data_frame for d in data]


def rearrange_columns(col_names: list):
    meta_columns = ["id", "created_at", "created_by", "updated_at",
                    "updated_by", "datapoint_name", "administration",
                    "geolocation"]
    col_question = list(filter(lambda x: x not in meta_columns, col_names))
    if len(col_question) == len(col_names):
        return col_question
    col_names = meta_columns + col_question
    return col_names


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'))
    data = download(form=form, administration_ids=administration_ids)
    df = pd.DataFrame(data)
    col_names = rearrange_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_administrations_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,
    })

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

akvo / rtmis / #1109

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