feat: added porters and new serializer based exporter

Author: henit-chobisa

apps/api/plane/bgtasks/export_task.py

@@ -1,23 +1,64 @@
 # Python imports
 import io
+import logging
 import zipfile
 from typing import List
-import boto3
-from botocore.client import Config
 from uuid import UUID
 
 # Third party imports
 from celery import shared_task
 
 # Django imports
-from django.conf import settings
-from django.utils import timezone
 from django.db.models import Prefetch
+from django.utils import timezone
 
 # Module imports
-from plane.db.models import ExporterHistory, Issue, IssueRelation
+from plane.db.models import ExporterHistory, Issue, IssueRelation, StateGroup
+from plane.settings.storage import S3Storage
 from plane.utils.exception_logger import log_exception
-from plane.utils.exporters import Exporter, IssueExportSchema
+from plane.utils.porters import DataExporter, CSVFormatter, JSONFormatter, XLSXFormatter, IssueExportSerializer
+from plane.utils.filters import ComplexFilterBackend, IssueFilterSet
+from plane.utils.host import base_host
+from plane.utils.issue_filters import issue_filters
+
+# Logger
+logger = logging.getLogger("plane.worker")
+
+
+class _FakeDjangoRequest:
+    def __init__(self):
+        from urllib.parse import urlparse
+
+        from django.http import QueryDict
+
+        self.GET = QueryDict(mutable=True)
+
+        # Get the public URL from environment variables
+        web_url = base_host(is_app=True, request=None)
+        parsed_url = urlparse(web_url)
+
+        # Add scheme and host attributes needed by S3Storage
+        self.scheme = parsed_url.scheme or "http"
+        self._host = parsed_url.netloc or "localhost"
+
+    def get_host(self):
+        return self._host
+
+
+class _FakeDRFRequest:
+    def __init__(self):
+        self._request = _FakeDjangoRequest()
+
+    @property
+    def query_params(self):
+        return self._request.GET
+
+
+class _ExportFilterView:
+    filterset_class = IssueFilterSet
+
+    def __init__(self, request):
+        self.request = request
 
 
 def create_zip_file(files: List[tuple[str, str | bytes]]) -> io.BytesIO:
@@ -39,83 +80,63 @@ def upload_to_s3(zip_file: io.BytesIO, workspace_id: UUID, token_id: str, slug:
     Upload a ZIP file to S3 and generate a presigned URL.
     """
     file_name = f"{workspace_id}/export-{slug}-{token_id[:6]}-{str(timezone.now().date())}.zip"
-    expires_in = 7 * 24 * 60 * 60
 
-    if settings.USE_MINIO:
-        upload_s3 = boto3.client(
-            "s3",
-            endpoint_url=settings.AWS_S3_ENDPOINT_URL,
-            aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
-            aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
-            config=Config(signature_version="s3v4"),
-        )
-        upload_s3.upload_fileobj(
-            zip_file,
-            settings.AWS_STORAGE_BUCKET_NAME,
-            file_name,
-            ExtraArgs={"ACL": "public-read", "ContentType": "application/zip"},
-        )
+    logger.info(
+        "Uploading export file to S3",
+        {
+            "file_name": file_name,
+        },
+    )
 
-        # Generate presigned url for the uploaded file with different base
-        presign_s3 = boto3.client(
-            "s3",
-            endpoint_url=(
-                f"{settings.AWS_S3_URL_PROTOCOL}//{str(settings.AWS_S3_CUSTOM_DOMAIN).replace('/uploads', '')}/"
-            ),
-            aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
-            aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
-            config=Config(signature_version="s3v4"),
-        )
+    expires_in = 7 * 24 * 60 * 60
 
-        presigned_url = presign_s3.generate_presigned_url(
-            "get_object",
-            Params={"Bucket": settings.AWS_STORAGE_BUCKET_NAME, "Key": file_name},
-            ExpiresIn=expires_in,
-        )
-    else:
-        # If endpoint url is present, use it
-        if settings.AWS_S3_ENDPOINT_URL:
-            s3 = boto3.client(
-                "s3",
-                endpoint_url=settings.AWS_S3_ENDPOINT_URL,
-                aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
-                aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
-                config=Config(signature_version="s3v4"),
-            )
-        else:
-            s3 = boto3.client(
-                "s3",
-                region_name=settings.AWS_REGION,
-                aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
-                aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
-                config=Config(signature_version="s3v4"),
-            )
+    storage = S3Storage(request=None)
 
-        # Upload the file to S3
-        s3.upload_fileobj(
-            zip_file,
-            settings.AWS_STORAGE_BUCKET_NAME,
-            file_name,
-            ExtraArgs={"ContentType": "application/zip"},
-        )
+    # Upload the file to S3
+    is_uploaded = storage.upload_file(
+        file_obj=zip_file,
+        object_name=file_name,
+        content_type="application/zip",
+    )
+    if not is_uploaded:
+        logger.error("Failed to upload export file to S3")
+        return
 
-        # Generate presigned url for the uploaded file
-        presigned_url = s3.generate_presigned_url(
-            "get_object",
-            Params={"Bucket": settings.AWS_STORAGE_BUCKET_NAME, "Key": file_name},
-            ExpiresIn=expires_in,
-        )
+    # Generate a presigned URL for the uploaded file
+    fake_request = _FakeDjangoRequest()
+    storage = S3Storage(request=fake_request)
+
+    presigned_url = storage.generate_presigned_url(
+        file_name,
+        expiration=expires_in,
+        http_method="GET",
+        disposition="inline",
+        filename=file_name,
+    )
 
     exporter_instance = ExporterHistory.objects.get(token=token_id)
 
     # Update the exporter instance with the presigned url
     if presigned_url:
+        logger.info(
+            "Uploaded export file to S3",
+            {
+                "file_name": file_name,
+            },
+        )
         exporter_instance.url = presigned_url
         exporter_instance.status = "completed"
         exporter_instance.key = file_name
     else:
         exporter_instance.status = "failed"
+        logger.error("Failed to upload export file to S3")
 
+    logger.info(
+        "Saving exporter instance",
+        {
+            "exporter_instance": exporter_instance,
+        },
+    )
     exporter_instance.save(update_fields=["status", "url", "key"])
 
 
@@ -127,21 +148,49 @@ def issue_export_task(
     token_id: str,
     multiple: bool,
     slug: str,
+    export_type: str | None = None,
 ):
     """
     Export issues from the workspace.
     provider (str): The provider to export the issues to csv | json | xlsx.
     token_id (str): The export object token id.
     multiple (bool): Whether to export the issues to multiple files per project.
+    export_type (str | None): The type of export (epic, intake, issue, etc.).
     """
+
+    logger.info(f"Export started for work-items for project {project_ids} in workspace {workspace_id}")
+
     try:
         exporter_instance = ExporterHistory.objects.get(token=token_id)
         exporter_instance.status = "processing"
         exporter_instance.save(update_fields=["status"])
 
-        # Build base queryset for issues
+        logger.info(
+            "Building base queryset for issues",
+            {
+                "workspace_id": workspace_id,
+                "type": exporter_instance.type,
+                "export_type": export_type,
+            },
+        )
+
+        # Build base queryset with export_type-specific manager and filters
+        if export_type == "epic":
+            # Use issue_and_epics_objects manager for epics with epic filter
+            base_queryset = Issue.issue_and_epics_objects.filter(type__is_epic=True)
+        elif export_type == "intake":
+            # Use objects manager for intake with triage state filter
+            base_queryset = Issue.objects.filter(state__group=StateGroup.TRIAGE.value)
+        elif export_type == "issue":
+            # Use issue_objects manager for regular issues (workitem, cycle, module, view)
+            base_queryset = Issue.issue_objects.all()
+        else:
+            # Default: Use objects manager to export all types of issues (workspace export)
+            base_queryset = Issue.objects.all()
+
+        # Apply common filters
         workspace_issues = (
-            Issue.objects.filter(
+            base_queryset.filter(
                 workspace__id=workspace_id,
                 project_id__in=project_ids,
                 project__project_projectmember__member=exporter_instance.initiated_by_id,
@@ -152,11 +201,12 @@ def issue_export_task(
                 "project",
                 "workspace",
                 "state",
+                "type",
                 "created_by",
                 "estimate_point",
             )
             .prefetch_related(
-                "labels",
+                "label_issue__label",
                 "issue_cycle__cycle",
                 "issue_module__module",
                 "issue_comments",
@@ -178,33 +228,67 @@ def issue_export_task(
             )
         )
 
-        # Create exporter for the specified format
-        try:
-            exporter = Exporter(
-                format_type=provider,
-                schema_class=IssueExportSchema,
-                options={"list_joiner": ", "},
+        # Apply filters if present
+        rich_filters = exporter_instance.rich_filters
+        logger.info(
+            "Applying rich filters",
+            {
+                "rich_filters": rich_filters,
+            },
+        )
+        if rich_filters:
+            backend = ComplexFilterBackend()
+            fake_request = _FakeDRFRequest()
+            view = _ExportFilterView(fake_request)
+            workspace_issues = backend.filter_queryset(
+                fake_request,
+                workspace_issues,
+                view,
+                filter_data=rich_filters,
             )
-        except ValueError as e:
-            # Invalid format type
-            exporter_instance = ExporterHistory.objects.get(token=token_id)
+
+        # Apply legacy filters if present
+        filters = exporter_instance.filters
+        logger.info(
+            "Applying legacy filters",
+            {
+                "filters": filters,
+            },
+        )
+        if filters:
+            filters = issue_filters(filters, "GET")
+            workspace_issues = workspace_issues.filter(**filters)
+
+        # Create exporter for the specified format
+        formatters = {"csv": CSVFormatter(), "json": JSONFormatter(), "xlsx": XLSXFormatter()}
+        if provider not in formatters:
             exporter_instance.status = "failed"
-            exporter_instance.reason = str(e)
+            exporter_instance.reason = f"Unsupported format: {provider}. Available: csv, json, xlsx"
             exporter_instance.save(update_fields=["status", "reason"])
             return
 
+        formatter = formatters[provider]
+        exporter = DataExporter(IssueExportSerializer)
+
+        logger.info(
+            "Creating files",
+            {
+                "multiple": multiple,
+                "project_ids": project_ids,
+            },
+        )
         files = []
         if multiple:
             # Export each project separately with its own queryset
             for project_id in project_ids:
                 project_issues = workspace_issues.filter(project_id=project_id)
-                export_filename = f"{slug}-{project_id}"
-                filename, content = exporter.export(export_filename, project_issues)
+                filename = f"{slug}-{project_id}.{formatter.extension}"
+                content = exporter.to_string(project_issues, formatter)
                 files.append((filename, content))
         else:
             # Export all issues in a single file
-            export_filename = f"{slug}-{workspace_id}"
-            filename, content = exporter.export(export_filename, workspace_issues)
+            filename = f"{slug}-{workspace_id}.{formatter.extension}"
+            content = exporter.to_string(workspace_issues, formatter)
             files.append((filename, content))
 
         zip_buffer = create_zip_file(files)

apps/api/plane/utils/porters/__init__.py

@@ -0,0 +1,15 @@
+from .formatters import BaseFormatter, CSVFormatter, JSONFormatter, XLSXFormatter
+from .exporter import DataExporter
+from .serializers import IssueExportSerializer
+
+__all__ = [
+    # Formatters
+    "BaseFormatter",
+    "CSVFormatter",
+    "JSONFormatter",
+    "XLSXFormatter",
+    # Exporters
+    "DataExporter",
+    # Export Serializers
+    "IssueExportSerializer",
+]

apps/api/plane/utils/porters/exporter.py

@@ -0,0 +1,37 @@
+from typing import Dict, List
+from .formatters import BaseFormatter
+
+class DataExporter:
+    """
+    Export data using DRF serializers.
+
+    Usage:
+        exporter = DataExporter(BookSerializer, exclude=['password'])
+        exporter.to_file(queryset, 'books.csv', CSVFormatter())
+        csv_string = exporter.to_string(queryset, CSVFormatter())
+    """
+
+    def __init__(self, serializer_class, **serializer_kwargs):
+        self.serializer_class = serializer_class
+        self.serializer_kwargs = serializer_kwargs
+
+    def serialize(self, queryset) -> List[Dict]:
+        """QuerySet → list of dicts"""
+        serializer = self.serializer_class(
+            queryset,
+            many=True,
+            **self.serializer_kwargs
+        )
+        return serializer.data
+
+    def to_string(self, queryset, formatter: BaseFormatter) -> str:
+        """Export to formatted string"""
+        data = self.serialize(queryset)
+        return formatter.encode(data)
+
+    def to_file(self, queryset, filepath: str, formatter: BaseFormatter) -> str:
+        """Export to file"""
+        content = self.to_string(queryset, formatter)
+        with open(filepath, 'w', encoding='utf-8') as f:
+            f.write(content)
+        return filepath