UN-2470 [FEAT] Remove Django dependency from Celery workers with internal APIs (#1494)

* UN-2470 [MISC] Remove Django dependency from Celery workers

This commit introduces a new worker architecture that decouples
Celery workers from Django where possible, enabling support for
gevent/eventlet pool types and reducing worker startup overhead.

Key changes:
- Created separate worker modules (api-deployment, callback, file_processing, general)
- Added internal API endpoints for worker communication
- Implemented Django-free task execution where appropriate
- Added shared utilities and client facades
- Updated container configurations for new worker architecture

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>

* Fix pre-commit issues: file permissions and ruff errors

Setup the docker for new workers

- Add executable permissions to worker entrypoint files
- Fix import order in namespace package __init__.py
- Remove unused variable api_status in general worker
- Address ruff E402 and F841 errors

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>

* refactoreed, Dockerfiles,fixes

* flexibility on celery run commands

* added debug logs

* handled filehistory for API

* cleanup

* cleanup

* cloud plugin structure

* minor changes in import plugin

* added notification and logger workers under new worker module

* add docker compatibility for new workers

* handled docker issues

* log consumer worker fixes

* added scheduler worker

* minor env changes

* cleanup the logs

* minor changes in logs

* resolved scheduler worker issues

* cleanup and refactor

* ensuring backward compatibbility to existing wokers

* added configuration internal apis and cache utils

* optimization

* Fix API client singleton pattern to share HTTP sessions

- Fix flawed singleton implementation that was trying to share BaseAPIClient instances
- Now properly shares HTTP sessions between specialized clients
- Eliminates 6x BaseAPIClient initialization by reusing the same underlying session
- Should reduce API deployment orchestration time by ~135ms (from 6 clients to 1 session)
- Added debug logging to verify singleton pattern activation

* cleanup and structuring

* cleanup in callback

* file system connectors  issue

* celery env values changes

* optional gossip

* variables for sync, mingle and gossip

* Fix for file type check

* Task pipeline issue resolving

* api deployement failed response handled

* Task pipline fixes

* updated file history cleanup with active file execution

* pipline status update and workflow ui page execution

* cleanup and resolvinf conflicts

* remove unstract-core from conenctoprs

* Commit uv.lock changes

* uv locks updates

* resolve migration issues

* defer connector-metadtda

* Fix connector migration for production scale

- Add encryption key handling with defer() to prevent decryption failures
- Add final cleanup step to fix duplicate connector names
- Optimize for large datasets with batch processing and bulk operations
- Ensure unique constraint in migration 0004 can be created successfully

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>

* hitl fixes

* minor fixes on hitl

* api_hub related changes

* dockerfile fixes

* api client cache fixes with actual response class

* fix: tags and llm_profile_id

* optimized clear cache

* cleanup

* enhanced logs

* added more handling on is file dir and added loggers

* cleanup the runplatform script

* internal apis are excempting from csrf

* sonal cloud issues

* sona-cloud issues

* resolving sonar cloud issues

* resolving sonar cloud issues

* Delta: added Batch size fix in workers

* comments addressed

* celery configurational changes for new workers

* fiixes in callback regaurding the pipline type check

* change internal url registry logic

* gitignore changes

* gitignore changes

* addressng pr cmmnets and cleanup the codes

* adding missed profiles for v2

* sonal cloud blocker issues resolved

* imlement otel

* Commit uv.lock changes

* handle execution time and some cleanup

* adding user_data in metadata Pr: #1544

* scheduler backward compatibitlity

* replace user_data with custom_data

* Commit uv.lock changes

* celery worker command issue resolved

* enhance package imports in connectors by changing to lazy imports

* Update runner.py by removing the otel from it

Update runner.py by removing the otel from it

Signed-off-by: ali <117142933+muhammad-ali-e@users.noreply.github.com>

* added delta changes

* handle erro to destination db

* resolve tool instances id validation and hitl queu name in API

* handled direct execution from workflow page to worker and logs

* handle cost logs

* Update health.py

Signed-off-by: Ritwik G <100672805+ritwik-g@users.noreply.github.com>

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* minor log changes

* introducing log consumer scheduler to bulk create, and socket .emit from worker for ws

* Commit uv.lock changes

* time limit or timeout celery config cleanup

* implemented redis client class in worker

* pipline status enum mismatch

* notification worker fixes

* resolve uv lock conflicts

* workflow log fixes

* ws channel name issue resolved. and handling redis down in status tracker, and removing redis keys

* default TTL changed for unified logs

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

---------

Signed-off-by: ali <117142933+muhammad-ali-e@users.noreply.github.com>
Signed-off-by: Ritwik G <100672805+ritwik-g@users.noreply.github.com>
Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: Ritwik G <100672805+ritwik-g@users.noreply.github.com>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: 4 people

.gitignore

@@ -1,6 +1,14 @@
 # Created by https://www.toptal.com/developers/gitignore/api/windows,macos,linux,pycharm,pycharm+all,pycharm+iml,python,visualstudiocode,react,django
 # Edit at https://www.toptal.com/developers/gitignore?templates=windows,macos,linux,pycharm,pycharm+all,pycharm+iml,python,visualstudiocode,react,django
 
+# Development helper scripts
+*.sh
+# list Exceptional files with ! like !fix-and-test.sh
+!run-platform.sh
+!workers/run-worker.sh
+!workers/run-worker-docker.sh
+!workers/log_consumer/scheduler.sh
+
 ### Django ###
 *.log
 *.pot
@@ -622,6 +630,7 @@ backend/plugins/processor/*
 # Subscription Plugins
 backend/plugins/subscription/*
 
+
 # API Deployment Plugins
 backend/plugins/api/**
 
@@ -685,6 +694,7 @@ backend/requirements.txt
 backend/backend/*_urls.py
 !backend/backend/base_urls.py
 !backend/backend/public_urls.py
+!backend/backend/internal_base_urls.py
 # TODO: Remove after v2 migration is completed
 backend/backend/*_urls_v2.py
 !backend/backend/public_urls_v2.py

backend/account_v2/custom_auth_middleware.py

@@ -8,6 +8,7 @@
 from account_v2.authentication_service import AuthenticationService
 from account_v2.constants import Common
 from backend.constants import RequestHeader
+from backend.internal_api_constants import INTERNAL_API_PREFIX
 
 
 class CustomAuthMiddleware:
@@ -22,6 +23,10 @@ def __call__(self, request: HttpRequest) -> HttpResponse:
         if any(request.path.startswith(path) for path in settings.WHITELISTED_PATHS):
             return self.get_response(request)
 
+        # Skip internal API paths - they are handled by InternalAPIAuthMiddleware
+        if request.path.startswith(f"{INTERNAL_API_PREFIX}/"):
+            return self.get_response(request)
+
         # Authenticating With API_KEY
         x_api_key = request.headers.get(RequestHeader.X_API_KEY)
         if (

backend/account_v2/internal_serializers.py

@@ -0,0 +1,15 @@
+"""Account Internal API Serializers
+Handles serialization for organization context related endpoints.
+"""
+
+from rest_framework import serializers
+
+
+class OrganizationContextSerializer(serializers.Serializer):
+    """Serializer for organization context information."""
+
+    organization_id = serializers.CharField()
+    organization_name = serializers.CharField()
+    organization_slug = serializers.CharField(required=False, allow_blank=True)
+    created_at = serializers.CharField(required=False, allow_blank=True)
+    settings = serializers.DictField(required=False)

backend/account_v2/internal_urls.py

@@ -0,0 +1,20 @@
+"""Internal API URLs for Organization Context
+URL patterns for organization-related internal APIs.
+"""
+
+from django.urls import path
+
+from .internal_views import OrganizationContextAPIView
+
+urlpatterns = [
+    # Organization context endpoint (backward compatibility)
+    path(
+        "<str:org_id>/", OrganizationContextAPIView.as_view(), name="organization-context"
+    ),
+    # Organization context endpoint (explicit path)
+    path(
+        "<str:org_id>/context/",
+        OrganizationContextAPIView.as_view(),
+        name="organization-context-explicit",
+    ),
+]

backend/account_v2/internal_views.py

@@ -0,0 +1,40 @@
+"""Account Internal API Views
+Handles organization context related endpoints for internal services.
+"""
+
+import logging
+
+from rest_framework import status
+from rest_framework.response import Response
+from rest_framework.views import APIView
+from utils.organization_utils import get_organization_context, resolve_organization
+
+from .internal_serializers import OrganizationContextSerializer
+
+logger = logging.getLogger(__name__)
+
+
+class OrganizationContextAPIView(APIView):
+    """Internal API endpoint for getting organization context."""
+
+    def get(self, request, org_id):
+        """Get organization context information."""
+        try:
+            # Use shared utility to resolve organization
+            organization = resolve_organization(org_id, raise_on_not_found=True)
+
+            # Use shared utility to get context data
+            context_data = get_organization_context(organization)
+
+            serializer = OrganizationContextSerializer(context_data)
+
+            logger.info(f"Retrieved organization context for {org_id}")
+
+            return Response(serializer.data)
+
+        except Exception as e:
+            logger.error(f"Failed to get organization context for {org_id}: {str(e)}")
+            return Response(
+                {"error": "Failed to get organization context", "detail": str(e)},
+                status=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            )

backend/account_v2/organization_internal_urls.py

@@ -0,0 +1,16 @@
+"""Account Internal API URLs
+Defines internal API endpoints for organization operations.
+"""
+
+from django.urls import path
+
+from .internal_views import OrganizationContextAPIView
+
+urlpatterns = [
+    # Organization context API
+    path(
+        "<str:org_id>/context/",
+        OrganizationContextAPIView.as_view(),
+        name="organization-context",
+    ),
+]

backend/account_v2/subscription_loader.py

@@ -21,15 +21,28 @@ class SubscriptionConfig:
     METADATA_IS_ACTIVE = "is_active"
 
 
+# Cache for loaded plugins to avoid repeated loading
+_subscription_plugins_cache: list[Any] = []
+_plugins_loaded = False
+
+
 def load_plugins() -> list[Any]:
     """Iterate through the subscription plugins and register them."""
+    global _subscription_plugins_cache, _plugins_loaded
+
+    # Return cached plugins if already loaded
+    if _plugins_loaded:
+        return _subscription_plugins_cache
+
     plugins_app = apps.get_app_config(SubscriptionConfig.PLUGINS_APP)
     package_path = plugins_app.module.__package__
     subscription_dir = os.path.join(plugins_app.path, SubscriptionConfig.PLUGIN_DIR)
     subscription_package_path = f"{package_path}.{SubscriptionConfig.PLUGIN_DIR}"
     subscription_plugins: list[Any] = []
 
     if not os.path.exists(subscription_dir):
+        _subscription_plugins_cache = subscription_plugins
+        _plugins_loaded = True
         return subscription_plugins
 
     for item in os.listdir(subscription_dir):
@@ -56,10 +69,13 @@ def load_plugins() -> list[Any]:
                         SubscriptionConfig.METADATA: module.metadata,
                     }
                 )
+                name = metadata.get(
+                    SubscriptionConfig.METADATA_NAME,
+                    getattr(module, "__name__", "unknown"),
+                )
+                is_active = metadata.get(SubscriptionConfig.METADATA_IS_ACTIVE, False)
                 logger.info(
-                    "Loaded subscription plugin: %s, is_active: %s",
-                    module.metadata[SubscriptionConfig.METADATA_NAME],
-                    module.metadata[SubscriptionConfig.METADATA_IS_ACTIVE],
+                    "Loaded subscription plugin: %s, is_active: %s", name, is_active
                 )
             else:
                 logger.info(
@@ -75,6 +91,10 @@ def load_plugins() -> list[Any]:
     if len(subscription_plugins) == 0:
         logger.info("No subscription plugins found.")
 
+    # Cache the results for future requests
+    _subscription_plugins_cache = subscription_plugins
+    _plugins_loaded = True
+
     return subscription_plugins
 
 

backend/api_v2/internal_api_views.py

@@ -0,0 +1,74 @@
+"""Internal API Views for API v2
+
+This module provides internal API endpoints for worker communication,
+specifically optimized for type-aware pipeline data fetching.
+
+Since we know the context from worker function calls:
+- process_batch_callback_api -> APIDeployment model
+- process_batch_callback -> Pipeline model (handled in workflow_manager)
+
+This provides direct access to APIDeployment model data without
+the overhead of checking both Pipeline and APIDeployment models.
+"""
+
+import logging
+
+from rest_framework import status
+from rest_framework.response import Response
+from rest_framework.views import APIView
+
+from api_v2.models import APIDeployment
+from api_v2.serializers import APIDeploymentSerializer
+
+logger = logging.getLogger(__name__)
+
+
+class APIDeploymentDataView(APIView):
+    """Internal API endpoint for fetching APIDeployment data.
+
+    This endpoint is optimized for callback workers that know they're dealing
+    with API deployments. It directly queries the APIDeployment model without
+    checking the Pipeline model, improving performance.
+
+    Endpoint: GET /v2/api-deployments/{api_id}/data/
+    """
+
+    def get(self, request, api_id):
+        """Get APIDeployment model data by API ID.
+
+        Args:
+            request: HTTP request object
+            api_id: APIDeployment UUID
+
+        Returns:
+            Response with APIDeployment model data
+        """
+        try:
+            logger.debug(f"Fetching APIDeployment data for ID: {api_id}")
+
+            # Query APIDeployment model directly (organization-scoped via DefaultOrganizationMixin)
+            api_deployment = APIDeployment.objects.get(id=api_id)
+
+            # Serialize the APIDeployment model
+            serializer = APIDeploymentSerializer(api_deployment)
+
+            # Use consistent response format with pipeline endpoint
+            response_data = {"status": "success", "pipeline": serializer.data}
+
+            logger.info(
+                f"Found APIDeployment {api_id}: name='{api_deployment.api_name}', display_name='{api_deployment.display_name}'"
+            )
+            return Response(response_data, status=status.HTTP_200_OK)
+
+        except APIDeployment.DoesNotExist:
+            logger.warning(f"APIDeployment not found for ID: {api_id}")
+            return Response(
+                {"error": f"APIDeployment with ID {api_id} not found"},
+                status=status.HTTP_404_NOT_FOUND,
+            )
+        except Exception as e:
+            logger.error(f"Error fetching APIDeployment data for {api_id}: {str(e)}")
+            return Response(
+                {"error": f"Failed to fetch APIDeployment data: {str(e)}"},
+                status=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            )

backend/api_v2/internal_urls.py

@@ -0,0 +1,20 @@
+"""Internal API URLs for API v2
+
+Internal endpoints for worker communication, specifically optimized
+for type-aware pipeline data fetching.
+"""
+
+from django.urls import path
+from rest_framework.urlpatterns import format_suffix_patterns
+
+from api_v2.internal_api_views import APIDeploymentDataView
+
+urlpatterns = format_suffix_patterns(
+    [
+        path(
+            "<uuid:api_id>/",
+            APIDeploymentDataView.as_view(),
+            name="api_deployment_data_internal",
+        ),
+    ]
+)

backend/backend/base_urls.py

@@ -23,4 +23,6 @@
         include("pipeline_v2.public_api_urls"),
     ),
     path("", include("health.urls")),
+    # Internal API for worker communication
+    path("internal/", include("backend.internal_base_urls")),
 ]