Celery task to generate a CSV file containing the given invoice IDs.
Updates the user's InvoiceExportManager object with status/final data, then
adds a download task to the user's My Tasks when completed.
Source code in hypha/apply/projects/tasks.py
| @shared_task
def generate_invoice_csv(qs_ids: List[int], request_user_id: int) -> None:
"""Celery task to generate a CSV file containing the given invoice IDs.
Updates the user's InvoiceExportManager object with status/final data, then
adds a download task to the user's `My Tasks` when completed.
"""
try:
from hypha.apply.projects.models.payment import Invoice
qs = (
Invoice.objects.filter(id__in=qs_ids)
.select_related("project", "project__user")
.prefetch_related("tags")
)
request_user = User.objects.get(pk=request_user_id)
if current := InvoiceExportManager.objects.filter(user=request_user):
current.delete()
export_manager = InvoiceExportManager.objects.create(
user=request_user, total_export=len(qs_ids)
)
export_manager.export_data = export_invoices_to_csv(
qs.iterator(chunk_size=2000)
)
export_manager.set_completed_and_save()
user_task = DOWNLOAD_INVOICES_EXPORT
except Exception as exc:
export_manager.set_failed_and_save()
user_task = FAILED_INVOICES_EXPORT
if settings.SENTRY_DSN:
from sentry_sdk import capture_exception
capture_exception(exc)
else:
raise exc
finally:
if not settings.CELERY_TASK_ALWAYS_EAGER:
add_task_to_user(
code=user_task,
user=request_user,
related_obj=export_manager,
)
|