Okay, let's implement the suggested improvements. This involves adding Celery for asynchronous processing, enhancing the HTTP clients with retries, adding idempotency checks, basic token authentication, correlation IDs for logging, and setting up API documentation with `drf-spectacular`. **Note:** This is a significant update involving new dependencies and concepts. You'll need to install the new requirements and set up a Celery worker and a Redis instance (or another Celery broker/backend) for this to function fully. django>=4.2,<5.0 djangorestframework>=3.14,<3.16 requests>=2.28,<3.0 python-dotenv>=1.0.0 celery>=5.3,<6.0 redis>=4.5,<5.0 # Celery Broker/Backend & Cache Backend django-redis>=5.3,<5.5 # Django Cache interface for Redis drf-spectacular>=0.26,<0.28 # OpenAPI Documentation urllib3>=1.26,<2.0 # For requests Retry logic # This ensures the Celery app is loaded when Django starts from .celery import app as celery_app __all__ = ('celery_app',) import os from celery import Celery from django.conf import settings # Set the default Django settings module for the 'celery' program. os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'loan_orchestrator.settings') app = Celery('loan_orchestrator') # Using a string here means the worker doesn't have to serialize # the configuration object to child processes. # - namespace='CELERY' means all celery-related configuration keys # should have a `CELERY_` prefix. app.config_from_object('django.conf:settings', namespace='CELERY') # Load task modules from all registered Django app configs. app.autodiscover_tasks(lambda: settings.INSTALLED_APPS) @app.task(bind=True, ignore_result=True) def debug_task(self): print(f'Request: {self.request!r}') """ Django settings for loan_orchestrator project. """ from pathlib import Path import os from dotenv import load_dotenv import logging.config # Build paths inside the project like this: BASE_DIR / 'subdir'. BASE_DIR = Path(__file__).resolve().parent.parent # Load .env file if it exists dotenv_path = BASE_DIR / '.env' if os.path.exists(dotenv_path): load_dotenv(dotenv_path=dotenv_path) # Quick-start development settings - unsuitable for production # See https://docs.djangoproject.com/en/4.2/howto/deployment/checklist/ # SECURITY WARNING: keep the secret key used in production secret! SECRET_KEY = os.environ.get('DJANGO_SECRET_KEY', 'django-insecure-placeholder-key-for-development') # SECURITY WARNING: don't run with debug turned on in production! DEBUG = os.environ.get('DJANGO_DEBUG', 'True') == 'True' ALLOWED_HOSTS = os.environ.get('DJANGO_ALLOWED_HOSTS', 'localhost,127.0.0.1').split(',') # Application definition INSTALLED_APPS = [ 'django.contrib.admin', 'django.contrib.auth', 'django.contrib.contenttypes', 'django.contrib.sessions', 'django.contrib.messages', 'django.contrib.staticfiles', # Third-party apps 'rest_framework', 'rest_framework.authtoken', # For TokenAuthentication 'drf_spectacular', # For OpenAPI documentation 'django_redis', # For Redis cache # Local apps 'orchestration_api', ] MIDDLEWARE = [ 'orchestration_api.middleware.CorrelationIDMiddleware', # Add Correlation ID middleware first 'django.middleware.security.SecurityMiddleware', 'django.contrib.sessions.middleware.SessionMiddleware', 'django.middleware.common.CommonMiddleware', 'django.middleware.csrf.CsrfViewMiddleware', # Consider CSRF exemption for webhook endpoint if needed 'django.contrib.auth.middleware.AuthenticationMiddleware', 'django.contrib.messages.middleware.MessageMiddleware', 'django.middleware.clickjacking.XFrameOptionsMiddleware', ] ROOT_URLCONF = 'loan_orchestrator.urls' TEMPLATES = [ { 'BACKEND': 'django.template.backends.django.DjangoTemplates', 'DIRS': [], 'APP_DIRS': True, 'OPTIONS': { 'context_processors': [ 'django.template.context_processors.debug', 'django.template.context_processors.request', 'django.contrib.auth.context_processors.auth', 'django.contrib.messages.context_processors.messages', ], }, }, ] WSGI_APPLICATION = 'loan_orchestrator.wsgi.application' # Database # https://docs.djangoproject.com/en/4.2/ref/settings/#databases DATABASES = { 'default': { 'ENGINE': 'django.db.backends.sqlite3', 'NAME': BASE_DIR / 'db.sqlite3', } } # Password validation # https://docs.djangoproject.com/en/4.2/ref/settings/#auth-password-validators AUTH_PASSWORD_VALIDATORS = [ { 'NAME': 'django.contrib.auth.password_validation.UserAttributeSimilarityValidator', }, { 'NAME': 'django.contrib.auth.password_validation.MinimumLengthValidator', }, { 'NAME': 'django.contrib.auth.password_validation.CommonPasswordValidator', }, { 'NAME': 'django.contrib.auth.password_validation.NumericPasswordValidator', }, ] # Internationalization # https://docs.djangoproject.com/en/4.2/topics/i18n/ LANGUAGE_CODE = 'en-us' TIME_ZONE = 'UTC' USE_I18N = True USE_TZ = True # Static files (CSS, JavaScript, Images) # https://docs.djangoproject.com/en/4.2/howto/static-files/ STATIC_URL = 'static/' STATIC_ROOT = BASE_DIR / 'staticfiles' # Default primary key field type # https://docs.djangoproject.com/en/4.2/ref/settings/#default-auto-field DEFAULT_AUTO_FIELD = 'django.db.models.BigAutoField' # REST Framework settings REST_FRAMEWORK = { 'DEFAULT_RENDERER_CLASSES': [ 'rest_framework.renderers.JSONRenderer', ], 'DEFAULT_PARSER_CLASSES': [ 'rest_framework.parsers.JSONParser', ], 'DEFAULT_AUTHENTICATION_CLASSES': [ # Use TokenAuthentication for API endpoints (except webhooks) 'rest_framework.authentication.TokenAuthentication', # SessionAuthentication might be useful for browsable API/admin 'rest_framework.authentication.SessionAuthentication', ], 'DEFAULT_PERMISSION_CLASSES': [ # Require authentication by default 'rest_framework.permissions.IsAuthenticated', ], 'DEFAULT_SCHEMA_CLASS': 'drf_spectacular.openapi.AutoSchema', # For OpenAPI docs } # DRF Spectacular Settings (for OpenAPI docs) SPECTACULAR_SETTINGS = { 'TITLE': 'Loan Orchestrator API', 'DESCRIPTION': 'API acting as an orchestration layer between internal systems and Collateral360.', 'VERSION': '1.0.0', 'SERVE_INCLUDE_SCHEMA': False, # Schema available at /api/schema/ # Optional: Add security definitions for Swagger UI 'SWAGGER_UI_SETTINGS': { 'deepLinking': True, "persistAuthorization": True, "displayOperationId": True, }, 'SECURITY': [ {'Token': []} # Maps to the 'Token' scheme defined below (or use default) ], } # Collateral360 API Configuration COLLATERAL360_API_BASE_URL = os.environ.get('COLLATERAL360_API_BASE_URL', 'https://api.example-collateral360.com/') COLLATERAL360_OAUTH_TOKEN_PATH = os.environ.get('COLLATERAL360_OAUTH_TOKEN_PATH', '/v1/los/oauth/token') COLLATERAL360_CLIENT_ID = os.environ.get('COLLATERAL360_CLIENT_ID') COLLATERAL360_CLIENT_SECRET = os.environ.get('COLLATERAL360_CLIENT_SECRET') COLLATERAL360_VERIFY_SSL = os.environ.get('COLLATERAL360_VERIFY_SSL', 'True') == 'True' COLLATERAL360_WEBHOOK_SECRET = os.environ.get('COLLATERAL360_WEBHOOK_SECRET') # For verifying incoming webhooks COLLATERAL360_TIMEOUT = int(os.environ.get('COLLATERAL360_TIMEOUT', '30')) # Request timeout in seconds COLLATERAL360_RETRY_TOTAL = int(os.environ.get('COLLATERAL360_RETRY_TOTAL', '3')) # Total number of retries COLLATERAL360_RETRY_BACKOFF = float(os.environ.get('COLLATERAL360_RETRY_BACKOFF', '0.5')) # Backoff factor (sleep = {backoff factor} * (2 ** ({number of total retries} - 1))) COLLATERAL360_RETRY_STATUS_CODES = [int(s.strip()) for s in os.environ.get('COLLATERAL360_RETRY_STATUS_CODES', '429, 500, 502, 503, 504').split(',')] # Status codes to retry on # Internal LOS Configuration (Adjust as needed) INTERNAL_LOS_API_BASE_URL = os.environ.get('INTERNAL_LOS_API_BASE_URL', 'http://internal-los-api.local/') INTERNAL_LOS_API_KEY = os.environ.get('INTERNAL_LOS_API_KEY') INTERNAL_LOS_TIMEOUT = int(os.environ.get('INTERNAL_LOS_TIMEOUT', '15')) INTERNAL_LOS_RETRY_TOTAL = int(os.environ.get('INTERNAL_LOS_RETRY_TOTAL', '3')) INTERNAL_LOS_RETRY_BACKOFF = float(os.environ.get('INTERNAL_LOS_RETRY_BACKOFF', '0.3')) INTERNAL_LOS_RETRY_STATUS_CODES = [int(s.strip()) for s in os.environ.get('INTERNAL_LOS_RETRY_STATUS_CODES', '429, 500, 502, 503, 504').split(',')] # Celery Configuration # Using Redis as the broker and result backend CELERY_BROKER_URL = os.environ.get('CELERY_BROKER_URL', 'redis://localhost:6379/0') CELERY_RESULT_BACKEND = os.environ.get('CELERY_RESULT_BACKEND', 'redis://localhost:6379/1') CELERY_ACCEPT_CONTENT = ['json'] CELERY_TASK_SERIALIZER = 'json' CELERY_RESULT_SERIALIZER = 'json' CELERY_TIMEZONE = TIME_ZONE # Use Django's timezone CELERY_TASK_TRACK_STARTED = True # Helps with monitoring CELERY_RESULT_EXPIRES = 3600 # Expire results after 1 hour # Cache Configuration (using Redis via django-redis) # Used for idempotency checks CACHE_URL = os.environ.get('CACHE_URL', 'redis://localhost:6379/2') CACHES = { "default": { "BACKEND": "django_redis.cache.RedisCache", "LOCATION": CACHE_URL, "OPTIONS": { "CLIENT_CLASS": "django_redis.client.DefaultClient", } } } IDEMPOTENCY_KEY_TIMEOUT = 60 * 60 * 24 # Timeout for idempotency keys (e.g., 24 hours) # Logging Configuration # Clear previous configuration LOGGING_CONFIG = None LOGLEVEL = os.environ.get('LOGLEVEL', 'info').upper() logging.config.dictConfig({ 'version': 1, 'disable_existing_loggers': False, 'formatters': { 'correlation': { # Include correlation ID in the log format 'format': '{levelname} {asctime} [{correlation_id}] {name} {message}', 'style': '{', }, 'standard': { 'format': '%(levelname)-8s [%(asctime)s] %(name)s: %(message)s' }, }, 'filters': { # Filter to add correlation ID to log records 'correlation_id': { '()': 'orchestration_api.logging_filters.CorrelationIDFilter', } }, 'handlers': { 'console': { 'class': 'logging.StreamHandler', 'filters': ['correlation_id'], # Apply filter to console handler 'formatter': 'correlation', # Use correlation formatter }, }, 'loggers': { 'django': { 'handlers': ['console'], 'level': os.getenv('DJANGO_LOG_LEVEL', 'INFO'), 'propagate': False, }, 'django.request': { # To include correlation ID in request logs 'handlers': ['console'], 'level': 'WARNING', # Adjust as needed 'propagate': False, }, 'celery': { 'handlers': ['console'], 'level': LOGLEVEL, 'propagate': False, }, 'orchestration_api': { # Logger for our app 'handlers': ['console'], 'level': 'DEBUG' if DEBUG else 'INFO', 'propagate': False, }, # Add other loggers as needed }, 'root': { # Optional: configure root logger if needed 'handlers': ['console'], 'level': LOGLEVEL, }, }) # Example environment variables - copy to .env and fill in actual values # DO NOT COMMIT .env to version control DJANGO_SECRET_KEY='your-django-secret-key-here' DJANGO_DEBUG=True DJANGO_ALLOWED_HOSTS=localhost,127.0.0.1 # Collateral360 API Credentials COLLATERAL360_API_BASE_URL='https://api.example-collateral360.com/' # Replace with actual base URL COLLATERAL360_OAUTH_TOKEN_PATH='/v1/los/oauth/token' # Verify/adjust if different from default COLLATERAL360_CLIENT_ID='your-collateral360-client-id' COLLATERAL360_CLIENT_SECRET='your-collateral360-client-secret' COLLATERAL360_VERIFY_SSL=True # Set to False for non-prod/proxy environments needing verify=False COLLATERAL360_TIMEOUT=30 COLLATERAL360_RETRY_TOTAL=3 COLLATERAL360_RETRY_BACKOFF=0.5 COLLATERAL360_RETRY_STATUS_CODES=429,500,502,503,504 # Collateral360 Webhook Secret (Provided by Collateral360 during webhook client setup) COLLATERAL360_WEBHOOK_SECRET='your-collateral360-webhook-secret' # Internal LOS API Credentials (if applicable) INTERNAL_LOS_API_BASE_URL='http://internal-los-api.local/' # Replace with actual base URL INTERNAL_LOS_API_KEY='your-internal-los-api-key' # Optional, if needed INTERNAL_LOS_TIMEOUT=15 INTERNAL_LOS_RETRY_TOTAL=3 INTERNAL_LOS_RETRY_BACKOFF=0.3 INTERNAL_LOS_RETRY_STATUS_CODES=429,500,502,503,504 # Celery / Redis / Cache URLs (Adjust host/port/db as needed) CELERY_BROKER_URL=redis://localhost:6379/0 CELERY_RESULT_BACKEND=redis://localhost:6379/1 CACHE_URL=redis://localhost:6379/2 # Logging Level (DEBUG, INFO, WARNING, ERROR, CRITICAL) LOGLEVEL=INFO import uuid import threading from typing import Callable from django.http import HttpRequest, HttpResponse # Use threading.local to store correlation ID per thread/request _thread_locals = threading.local() def get_correlation_id() -> str | None: """Retrieves the correlation ID for the current request.""" return getattr(_thread_locals, 'correlation_id', None) class CorrelationIDMiddleware: """ Middleware to generate or retrieve a correlation ID for each request and store it in thread-local storage for access throughout the request lifecycle, including logging. """ def __init__(self, get_response: Callable[[HttpRequest], HttpResponse]): self.get_response = get_response def __call__(self, request: HttpRequest) -> HttpResponse: # Try to get correlation ID from request header (e.g., X-Correlation-ID) # or generate a new one if not present. correlation_id = request.headers.get('X-Correlation-ID') if not correlation_id: correlation_id = uuid.uuid4().hex # Store it in thread-local storage _thread_locals.correlation_id = correlation_id response = self.get_response(request) # Add correlation ID to the response header response['X-Correlation-ID'] = correlation_id # Clean up thread-local storage after request is done if hasattr(_thread_locals, 'correlation_id'): del _thread_locals.correlation_id return response import logging from .middleware import get_correlation_id class CorrelationIDFilter(logging.Filter): """ Logging filter to add the correlation ID to log records. """ def filter(self, record): record.correlation_id = get_correlation_id() or 'N/A' # Add correlation_id attribute return True from functools import wraps from django.core.cache import cache from django.http import HttpRequest, JsonResponse from django.conf import settings import logging logger = logging.getLogger(__name__) def idempotent_view(view_func): """ Decorator to make a view idempotent based on the 'Idempotency-Key' header. Stores the result (or task ID for async views) in the cache. If the key is found in the cache, returns the cached response immediately. """ @wraps(view_func) def _wrapped_view(request: HttpRequest, *args, **kwargs): idempotency_key = request.headers.get('Idempotency-Key') if not idempotency_key: # If no key is provided, proceed without idempotency check # Alternatively, could return a 400 Bad Request if key is mandatory logger.warning("Idempotency-Key header missing, proceeding without check.") return view_func(request, *args, **kwargs) cache_key = f"idempotency:{idempotency_key}" cached_response_data = cache.get(cache_key) if cached_response_data: logger.info(f"Idempotency key '{idempotency_key}' found in cache. Returning cached response.") # Assuming cached data is a dict suitable for JsonResponse # Adjust status code if needed, maybe store it in cache too? Defaulting to 200 OK for cached hit. return JsonResponse(cached_response_data, status=200) # Key not found, execute the view function response = view_func(request, *args, **kwargs) # Cache the response data only if the request was successful (e.g., 2xx status) # For async tasks, we cache the initial 'Accepted' response data (like task_id) if 200 <= response.status_code < 300: try: # Assuming response is JsonResponse or similar with json() method response_data_to_cache = response.json() cache.set(cache_key, response_data_to_cache, timeout=settings.IDEMPOTENCY_KEY_TIMEOUT) logger.info(f"Cached response for idempotency key '{idempotency_key}'.") except Exception as e: # Handle cases where response might not be JSON serializable or other errors logger.error(f"Failed to cache response for idempotency key '{idempotency_key}': {e}", exc_info=True) return response return _wrapped_view import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry from django.conf import settings import logging import time from urllib.parse import urljoin from ..middleware import get_correlation_id # Import function to get correlation ID logger = logging.getLogger(__name__) class BaseClientError(Exception): """Base custom exception for HTTP client errors.""" def __init__(self, message, status_code=None, response_data=None, structured_errors=None): super().__init__(message) self.status_code = status_code self.response_data = response_data self.structured_errors = structured_errors class BaseClient: """ Base class for external API clients with session management, retries, and logging. """ SERVICE_NAME = "BaseService" # Override in subclasses def __init__(self, base_url, timeout, retry_total, retry_backoff, retry_status_codes, verify_ssl=True, auth_header_name=None, api_key=None): self.base_url = base_url self.timeout = timeout self.verify_ssl = verify_ssl self.auth_header_name = auth_header_name self.api_key = api_key self.session = self._create_session(retry_total, retry_backoff, retry_status_codes) if not self.verify_ssl: logger.warning(f"SSL verification is disabled for {self.SERVICE_NAME} API requests.") import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) def _create_session(self, total, backoff_factor, status_forcelist): """Creates a requests Session with retry logic.""" session = requests.Session() retries = Retry( total=total, backoff_factor=backoff_factor, status_forcelist=status_forcelist, allowed_methods=["HEAD", "GET", "POST", "PUT", "DELETE", "OPTIONS", "TRACE", "PATCH"], # Retry on all methods ) adapter = HTTPAdapter(max_retries=retries) session.mount('http://', adapter) session.mount('https://', adapter) return session def _get_headers(self): """Constructs headers for the request, including correlation ID and optional auth.""" correlation_id = get_correlation_id() headers = { 'Accept': 'application/json', 'Content-Type': 'application/json', } if correlation_id: headers['X-Correlation-ID'] = correlation_id # Add correlation ID header # Add specific auth header if configured if self.auth_header_name and self.api_key: headers[self.auth_header_name] = self.api_key return headers def _make_request(self, method, endpoint, data=None, params=None, files=None, stream=False, extra_headers=None): """ Helper method to make requests using the configured session. Handles JSON data, form data (if files are present), and optional streaming. """ url = urljoin(self.base_url, endpoint.lstrip('/')) headers = self._get_headers() if extra_headers: headers.update(extra_headers) # Adjust headers for specific content types if files: # Let requests handle Content-Type for multipart/form-data headers.pop('Content-Type', None) request_data = data # For multipart, data goes in 'data' field json_data = None elif isinstance(data, (str, bytes)): # Assuming raw data, let caller set Content-Type if needed via extra_headers request_data = data json_data = None else: # Default to JSON request_data = None json_data = data if stream: headers['Accept'] = '*/*' # Accept any content type for downloads correlation_id = headers.get('X-Correlation-ID', 'N/A') log_prefix = f"[{correlation_id}] {self.SERVICE_NAME} Request:" try: logger.debug(f"{log_prefix} {method} {url} | Params: {params} | Data: {data}") # Avoid logging full sensitive data in prod response = self.session.request( method, url, headers=headers, json=json_data, data=request_data, params=params, files=files, verify=self.verify_ssl, timeout=self.timeout * (1 + settings.CELERY_RETRY_TOTAL) if stream else self.timeout, # Increase timeout slightly for streaming/retries stream=stream ) logger.debug(f"{log_prefix} Response Status: {response.status_code}") response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx) # Handle streaming responses if stream: return response # Handle standard non-streaming responses if response.status_code == 204: return None try: if response.content: json_response = response.json() # Check for application-level errors if applicable (override in subclass if needed) # meta = json_response.get('meta', {}) # if not meta.get('success', True): ... return json_response else: return None except requests.exceptions.JSONDecodeError: logger.warning(f"{log_prefix} Non-JSON response received from {method} {url}. Status: {response.status_code}. Content: {response.text[:200]}...") return response.text except requests.exceptions.HTTPError as e: structured_errors = None error_data = e.response.text try: json_error_data = e.response.json() error_data = json_error_data # Attempt to extract structured errors (specific to Collateral360 structure) structured_errors = json_error_data.get('meta', {}).get('errors') except requests.exceptions.JSONDecodeError: pass logger.error(f"{log_prefix} HTTP error: {e.response.status_code} for {method} {url}. Response: {error_data}", exc_info=True) raise BaseClientError( f"{self.SERVICE_NAME} API request failed with HTTP status {e.response.status_code}", status_code=e.response.status_code, response_data=error_data, structured_errors=structured_errors ) from e except requests.exceptions.RequestException as e: logger.error(f"{log_prefix} Connection error: {e}", exc_info=True) raise BaseClientError(f"{self.SERVICE_NAME} Connection error: {e}") from e except Exception as e: logger.error(f"{log_prefix} Unexpected error: {e}", exc_info=True) raise BaseClientError(f"An unexpected error occurred during {self.SERVICE_NAME} API call: {e}") from e import requests from django.conf import settings import logging import time from urllib.parse import urljoin import json from .base_client import BaseClient, BaseClientError # Import base class and exception logger = logging.getLogger(__name__) # Rename specific exception to inherit from BaseClientError class Collateral360ClientError(BaseClientError): """Custom exception for Collateral360 client errors.""" pass class Collateral360Client(BaseClient): # Inherit from BaseClient """ Client for interacting with the Collateral360 LOS API using OAuth2 Client Credentials. Handles authentication, request formatting, retries, and basic error handling. """ SERVICE_NAME = "Collateral360" def __init__(self): super().__init__( base_url=settings.COLLATERAL360_API_BASE_URL, timeout=settings.COLLATERAL360_TIMEOUT, retry_total=settings.COLLATERAL360_RETRY_TOTAL, retry_backoff=settings.COLLATERAL360_RETRY_BACKOFF, retry_status_codes=settings.COLLATERAL360_RETRY_STATUS_CODES, verify_ssl=settings.COLLATERAL360_VERIFY_SSL # Auth handled separately via _get_access_token ) self.token_path = settings.COLLATERAL360_OAUTH_TOKEN_PATH self.client_id = settings.COLLATERAL360_CLIENT_ID self.client_secret = settings.COLLATERAL360_CLIENT_SECRET self._access_token = None self._token_expiry_time = 0 if not self.client_id or not self.client_secret: logger.error("Collateral360 Client ID or Secret not configured in settings.") raise ValueError("COLLATERAL360_CLIENT_ID and COLLATERAL360_CLIENT_SECRET must be set.") def _get_access_token(self): """ Retrieves/refreshes the OAuth2 access token. Uses the base class session for the request. """ buffer_seconds = 60 if self._access_token and self._token_expiry_time > (time.time() + buffer_seconds): return self._access_token logger.info(f"[{get_correlation_id() or 'N/A'}] Fetching new Collateral360 access token.") token_url = urljoin(self.base_url, self.token_path) headers = {'Content-Type': 'application/json', 'Accept': 'application/json'} payload = { 'grant_type': 'client_credentials', 'client_id': self.client_id, 'client_secret': self.client_secret, 'scope': '*' } try: # Use the session from the base class for retries etc. response = self.session.post( token_url, headers=headers, json=payload, verify=self.verify_ssl, timeout=self.timeout # Use configured timeout ) response.raise_for_status() response_data = response.json() token_data = response_data.get('data') if not token_data or 'access_token' not in token_data or 'expires_in' not in token_data: logger.error(f"[{get_correlation_id() or 'N/A'}] Invalid token response structure received: {response_data}") raise Collateral360ClientError("Received invalid token data structure from authentication server.", response_data=response_data) self._access_token = token_data['access_token'] self._token_expiry_time = time.time() + int(token_data['expires_in']) logger.info(f"[{get_correlation_id() or 'N/A'}] Successfully obtained new Collateral360 access token.") return self._access_token except requests.exceptions.HTTPError as e: error_details = e.response.text logger.error(f"[{get_correlation_id() or 'N/A'}] Failed to get Collateral360 access token ({e.response.status_code}): {error_details}", exc_info=True) raise Collateral360ClientError( f"Authentication failed with status {e.response.status_code}", status_code=e.response.status_code, response_data=error_details ) from e except requests.exceptions.RequestException as e: logger.error(f"[{get_correlation_id() or 'N/A'}] Network error during Collateral360 access token fetch: {e}", exc_info=True) raise Collateral360ClientError(f"Authentication failed due to network error: {e}", status_code=getattr(e.response, 'status_code', None)) from e except Exception as e: logger.error(f"[{get_correlation_id() or 'N/A'}] An unexpected error occurred during token fetch: {e}", exc_info=True) raise Collateral360ClientError(f"An unexpected error occurred during authentication: {e}") from e def _make_request(self, method, endpoint, data=None, params=None, files=None, stream=False): """ Overrides base _make_request to inject the Bearer token. """ token = self._get_access_token() auth_header = {'Authorization': f'Bearer {token}'} # Use the parent class's _make_request method, passing the auth header try: return super()._make_request(method, endpoint, data=data, params=params, files=files, stream=stream, extra_headers=auth_header) except BaseClientError as e: # Re-raise as specific exception if needed, or just let BaseClientError propagate # This maintains the structured_errors attribute if parsed by the base method. raise Collateral360ClientError( e.args[0], # Original message status_code=e.status_code, response_data=e.response_data, structured_errors=e.structured_errors ) from e # --- API Endpoint Methods --- # (Method implementations remain the same, docstrings updated previously) # ... (keep all existing methods like patch_collateral_overview, get_loan_info_collateral_overview, etc.) ... # Example: def patch_collateral_overview(self, location_id, update_data): # Docstring from previous step... endpoint = f"/v1/los/collateralOverview/{location_id}" if not isinstance(update_data, dict) or \ 'meta' not in update_data or \ 'updatedBy' not in update_data.get('meta', {}) or \ 'data' not in update_data: raise ValueError("Invalid update_data structure. Must include 'meta.updatedBy' and 'data'.") if not update_data['data'].get('transaction') and not update_data['data'].get('collaterals'): logger.warning(f"[{get_correlation_id() or 'N/A'}] Patch collateral overview called for location {location_id} with empty transaction and collaterals data.") logger.info(f"[{get_correlation_id() or 'N/A'}] Patching collateral overview for location ID: {location_id}") return self._make_request('PATCH', endpoint, data=update_data) # ... include ALL other endpoint methods here, ensuring they call self._make_request ... # (get_loan_info_collateral_overview, get_loan_info_environmental_risk_codes, get_service_request_fields, ...) # (create_service_request, update_service_request, add_service_to_location, add_collateral_to_loan, ...) # (get_service_request_status_by_id, list_files_by_service_request, download_file, upload_file, ...) # (get_data_capture_fields, get_utility_service_types, get_utility_document_types, ...) # (list_webhook_clients, create_webhook_client, delete_webhook_client, list_webhook_events, ...) # (list_webhook_subscriptions, create_webhook_subscription, unsubscribe_webhook) # --- Ensure all methods are present --- # Loan Information def get_loan_info_collateral_overview(self, location_id): endpoint = f"/v1/los/loan/information/collateralOverview/{location_id}" logger.info(f"[{get_correlation_id() or 'N/A'}] Getting loan info collateral overview for location ID: {location_id}") return self._make_request('GET', endpoint) def get_loan_info_environmental_risk_codes(self, location_id): endpoint = f"/v1/los/loan/information/environmental/riskCodes/{location_id}" logger.info(f"[{get_correlation_id() or 'N/A'}] Getting environmental risk codes for location ID: {location_id}") return self._make_request('GET', endpoint) # Fields def get_service_request_fields(self): endpoint = "/v1/los/serviceRequest/fields" logger.info(f"[{get_correlation_id() or 'N/A'}] Getting service request fields schema.") return self._make_request('GET', endpoint) def get_collateral_overview_fields(self): endpoint = "/v1/los/collateralOverview/fields" logger.info(f"[{get_correlation_id() or 'N/A'}] Getting collateral overview fields schema.") return self._make_request('GET', endpoint) def get_add_collateral_fields(self): endpoint = "/v1/los/addCollateral/fields" logger.info(f"[{get_correlation_id() or 'N/A'}] Getting add collateral fields schema.") return self._make_request('GET', endpoint) # Service Request Form & Actions def create_service_request(self, data): endpoint = "/v1/los/serviceRequest/form" logger.info(f"[{get_correlation_id() or 'N/A'}] Creating new service request.") return self._make_request('POST', endpoint, data=data) def update_service_request(self, service_request_id, data): endpoint = f"/v1/los/serviceRequest/form/{service_request_id}" logger.info(f"[{get_correlation_id() or 'N/A'}] Updating service request ID: {service_request_id}") return self._make_request('PUT', endpoint, data=data) def add_service_to_location(self, location_id, data): endpoint = f"/v1/los/serviceRequest/addService/locationID/{location_id}" if not isinstance(data, dict) or \ 'meta' not in data or 'updatedBy' not in data.get('meta', {}) or \ 'data' not in data or 'services' not in data.get('data', {}): raise ValueError("Invalid data structure for add_service. Must include 'meta.updatedBy' and 'data.services'.") if not data['data']['services']: raise ValueError("'data.services' array cannot be empty.") logger.info(f"[{get_correlation_id() or 'N/A'}] Requesting to add service(s) to location ID: {location_id} (asynchronous).") return self._make_request('POST', endpoint, data=data) def add_collateral_to_loan(self, loan_id, data): endpoint = f"/v1/los/serviceRequest/addCollateral/loanID/{loan_id}" if not isinstance(data, dict) or \ 'meta' not in data or \ 'updatedBy' not in data.get('meta', {}) or \ 'processAsWarnings' not in data.get('meta', {}) or \ 'data' not in data or 'collaterals' not in data.get('data', {}): raise ValueError("Invalid data structure for add_collateral. Must include 'meta.updatedBy', 'meta.processAsWarnings', and 'data.collaterals'.") if not data['data']['collaterals']: raise ValueError("'data.collaterals' array cannot be empty.") logger.info(f"[{get_correlation_id() or 'N/A'}] Adding collateral(s) to loan ID: {loan_id}") return self._make_request('POST', endpoint, data=data) def add_collateral_to_location(self, location_id, data): endpoint = f"/v1/los/serviceRequest/addCollateral/locationID/{location_id}" if not isinstance(data, dict) or \ 'meta' not in data or \ 'updatedBy' not in data.get('meta', {}) or \ 'processAsWarnings' not in data.get('meta', {}) or \ 'data' not in data or 'collaterals' not in data.get('data', {}): raise ValueError("Invalid data structure for add_collateral. Must include 'meta.updatedBy', 'meta.processAsWarnings', and 'data.collaterals'.") if not data['data']['collaterals']: raise ValueError("'data.collaterals' array cannot be empty.") logger.info(f"[{get_correlation_id() or 'N/A'}] Adding collateral(s) to loan identified by location ID: {location_id}") return self._make_request('POST', endpoint, data=data) # Service Request Status def get_service_request_status_by_id(self, service_request_id): endpoint = f"/v1/los/serviceRequest/status/serviceRequestID/{service_request_id}" logger.info(f"[{get_correlation_id() or 'N/A'}] Getting status for service request ID: {service_request_id}") return self._make_request('GET', endpoint) def get_service_request_status_by_loan(self, loan_id): endpoint = f"/v1/los/serviceRequest/status/loanID/{loan_id}" logger.info(f"[{get_correlation_id() or 'N/A'}] Getting service request status for loan ID: {loan_id}") return self._make_request('GET', endpoint) def get_service_request_status_by_location(self, location_id): endpoint = f"/v1/los/serviceRequest/status/locationID/{location_id}" logger.info(f"[{get_correlation_id() or 'N/A'}] Getting service request status for location ID: {location_id}") return self._make_request('GET', endpoint) # Service Request Files List def list_files_by_service_request(self, service_request_id): endpoint = f"/v1/los/serviceRequest/files/list/serviceRequestID/{service_request_id}" logger.info(f"[{get_correlation_id() or 'N/A'}] Listing files for service request ID: {service_request_id}") return self._make_request('GET', endpoint) def list_files_by_loan(self, loan_id): endpoint = f"/v1/los/serviceRequest/files/list/loanID/{loan_id}" logger.info(f"[{get_correlation_id() or 'N/A'}] Listing files for loan ID: {loan_id}") return self._make_request('GET', endpoint) def list_files_by_location(self, location_id): endpoint = f"/v1/los/serviceRequest/files/list/locationID/{location_id}" logger.info(f"[{get_correlation_id() or 'N/A'}] Listing files for location ID: {location_id}") return self._make_request('GET', endpoint) # Service Request File Download/Upload def download_file(self, upload_id): endpoint = f"/v1/los/serviceRequest/files/download/{upload_id}" logger.info(f"[{get_correlation_id() or 'N/A'}] Requesting download for file with upload ID: {upload_id}") return self._make_request('GET', endpoint, stream=True) def upload_file(self, location_id, uploaded_by, file_object, file_name, prepared_by=None, report_title=None, report_date=None, display_file_name=None, service_groups=None, service_types=None, document_types=None, document_status=None): endpoint = f"/v1/los/files/upload/locationId/{location_id}/uploadedBy/{uploaded_by}" files_payload = {'file': (file_name, file_object)} data_payload = {} if prepared_by: data_payload['preparedBy'] = prepared_by if report_title: data_payload['reportTitle'] = report_title if report_date: data_payload['reportDate'] = report_date if display_file_name: data_payload['displayFileName'] = display_file_name if service_groups: if not isinstance(service_groups, (list, tuple)): raise TypeError("service_groups must be a list or tuple") data_payload['serviceGroups'] = json.dumps(service_groups) if service_types: if not isinstance(service_types, (list, tuple)): raise TypeError("service_types must be a list or tuple") data_payload['serviceTypes'] = json.dumps(service_types) if document_types: if not isinstance(document_types, (list, tuple)): raise TypeError("document_types must be a list or tuple") data_payload['documentTypes'] = json.dumps(document_types) if document_status: if not isinstance(document_status, dict): raise TypeError("document_status must be a dict") data_payload['documentStatus'] = json.dumps(document_status) logger.info(f"[{get_correlation_id() or 'N/A'}] Uploading file '{file_name}' for location ID: {location_id} by {uploaded_by}") return self._make_request('POST', endpoint, data=data_payload, files=files_payload) # Data Capture def get_data_capture_fields(self, location_service_id): endpoint = f"/v1/datacaptures/locationServiceId/{location_service_id}" logger.info(f"[{get_correlation_id() or 'N/A'}] Getting data capture fields for location service ID: {location_service_id}") return self._make_request('GET', endpoint) # Utilities def get_utility_service_types(self): endpoint = "/v1/los/utility/serviceTypes" logger.info(f"[{get_correlation_id() or 'N/A'}] Getting utility service types and groups.") return self._make_request('GET', endpoint) def get_utility_document_types(self): endpoint = "/v1/los/utility/documentTypes" logger.info(f"[{get_correlation_id() or 'N/A'}] Getting utility document types.") return self._make_request('GET', endpoint) def get_utility_document_status(self): endpoint = "/v1/los/utility/documentStatus" logger.info(f"[{get_correlation_id() or 'N/A'}] Getting utility document status.") return self._make_request('GET', endpoint) # Webhooks Clients def list_webhook_clients(self): endpoint = "/v1/los/webhooks/clients" logger.info(f"[{get_correlation_id() or 'N/A'}] Listing webhook clients.") return self._make_request('GET', endpoint) def create_webhook_client(self, url: str, label: str): endpoint = "/v1/los/webhooks/clients" if not url or not label: raise ValueError("Both 'url' and 'label' are required to create a webhook client.") payload = {"url": url, "label": label} logger.info(f"[{get_correlation_id() or 'N/A'}] Creating webhook client with label '{label}' and URL '{url}'.") return self._make_request('POST', endpoint, data=payload) def delete_webhook_client(self, client_id): endpoint = f"/v1/los/webhooks/clients/{client_id}" logger.info(f"[{get_correlation_id() or 'N/A'}] Deleting webhook client ID: {client_id}") return self._make_request('DELETE', endpoint) # Webhooks Events def list_webhook_events(self): endpoint = "/v1/los/webhooks/events" logger.info(f"[{get_correlation_id() or 'N/A'}] Listing webhook events.") return self._make_request('GET', endpoint) # Webhooks Subscriptions def list_webhook_subscriptions(self): endpoint = "/v1/los/webhooks/subscriptions" logger.info(f"[{get_correlation_id() or 'N/A'}] Listing webhook subscriptions.") return self._make_request('GET', endpoint) def create_webhook_subscription(self, client_id: int, event_slug: str): endpoint = "/v1/los/webhooks/subscriptions" if not client_id or not event_slug: raise ValueError("Both 'client_id' and 'event_slug' are required.") payload = {"clientId": client_id, "eventSlug": event_slug} logger.info(f"[{get_correlation_id() or 'N/A'}] Creating webhook subscription for client ID {client_id} to event '{event_slug}'.") return self._make_request('POST', endpoint, data=payload) def unsubscribe_webhook(self, subscription_id): endpoint = f"/v1/los/webhooks/subscriptions/{subscription_id}/unsubscribe" logger.info(f"[{get_correlation_id() or 'N/A'}] Unsubscribing webhook subscription ID: {subscription_id}") return self._make_request('POST', endpoint) import requests from django.conf import settings import logging from .base_client import BaseClient, BaseClientError # Import base class and exception logger = logging.getLogger(__name__) class InternalLOSClientError(BaseClientError): """Custom exception for Internal LOS client errors.""" pass class InternalLOSClient(BaseClient): # Inherit from BaseClient """ Client for interacting with the internal Loan Origination System (LOS) API. (This is a placeholder and needs to be adapted to your specific LOS API) """ SERVICE_NAME = "InternalLOS" def __init__(self): # Determine auth header and key based on settings auth_header = None api_key = settings.INTERNAL_LOS_API_KEY if api_key: # Adjust header name based on LOS authentication mechanism auth_header = 'Authorization' # Example: 'Authorization' api_key = f'ApiKey {api_key}' # Example: Prepend 'ApiKey ' super().__init__( base_url=settings.INTERNAL_LOS_API_BASE_URL, timeout=settings.INTERNAL_LOS_TIMEOUT, retry_total=settings.INTERNAL_LOS_RETRY_TOTAL, retry_backoff=settings.INTERNAL_LOS_RETRY_BACKOFF, retry_status_codes=settings.INTERNAL_LOS_RETRY_STATUS_CODES, verify_ssl=True, # Assuming internal service uses valid SSL auth_header_name=auth_header, api_key=api_key ) def _make_request(self, method, endpoint, data=None, params=None, files=None, stream=False): """Overrides base method to handle specific error patterns if needed.""" try: # Use the parent class's _make_request method return super()._make_request(method, endpoint, data=data, params=params, files=files, stream=stream) except BaseClientError as e: # Re-raise as specific exception or handle differently if LOS errors are structured uniquely raise InternalLOSClientError( e.args[0], # Original message status_code=e.status_code, response_data=e.response_data, structured_errors=e.structured_errors # Pass along if base class found any ) from e # --- Implement methods based on your Internal LOS API capabilities --- def update_loan_status(self, loan_id, status_data): """ Example: Update the status of a loan in the LOS. """ endpoint = f"loans/{loan_id}/status" # Adjust endpoint return self._make_request('PUT', endpoint, data=status_data) def get_loan_details(self, loan_id): """ Example: Get loan details from the LOS. """ endpoint = f"loans/{loan_id}" # Adjust endpoint return self._make_request('GET', endpoint) def create_loan_record(self, loan_data): """ Example: Create a new loan record in the LOS. """ endpoint = "loans" # Adjust endpoint return self._make_request('POST', endpoint, data=loan_data) def update_loan_with_c360_ids(self, internal_loan_id, c360_srf_id, c360_location_ids): """ Example: Update internal loan with IDs from Collateral360. """ endpoint = f"loans/{internal_loan_id}/c360-refs" # Example endpoint payload = { "c360ServiceRequestId": c360_srf_id, "c360LocationIds": c360_location_ids, } # Assuming PATCH or PUT is appropriate return self._make_request('PATCH', endpoint, data=payload) # Add other methods relevant to the orchestration process from celery import shared_task import logging from .services.collateral360_client import Collateral360Client, Collateral360ClientError from .services.internal_los_client import InternalLOSClient, InternalLOSClientError from .middleware import get_correlation_id # Import function to get correlation ID logger = logging.getLogger(__name__) # Example task for creating a service request @shared_task(bind=True, max_retries=3, default_retry_delay=60) # Add retry options def process_create_service_request(self, c360_payload: dict, internal_loan_id: str, correlation_id: str): """ Celery task to create a service request in Collateral360 and update internal LOS. """ # NOTE: Correlation ID is passed explicitly as Celery tasks run in different contexts log_prefix = f"[{correlation_id}] Task process_create_service_request:" logger.info(f"{log_prefix} Starting for internal loan ID {internal_loan_id}.") c360_client = Collateral360Client() internal_los_client = InternalLOSClient() try: # Step 1: Create Service Request in Collateral360 logger.info(f"{log_prefix} Calling C360 create_service_request.") c360_response = c360_client.create_service_request(c360_payload) # Check for application-level errors even if HTTP status was OK if not c360_response or not c360_response.get('meta', {}).get('success'): logger.error(f"{log_prefix} C360 create_service_request failed at application level: {c360_response}") # Decide on retry logic or permanent failure based on the error # For now, we'll let it fail permanently if meta.success is false raise Collateral360ClientError("C360 application error during SRF creation.", response_data=c360_response) c360_srf_id = c360_response.get('data', {}).get('serviceRequestID') c360_location_ids = c360_response.get('data', {}).get('locationID', []) if not c360_srf_id: logger.error(f"{log_prefix} C360 response missing serviceRequestID: {c360_response}") raise ValueError("Missing serviceRequestID in C360 response.") logger.info(f"{log_prefix} C360 SRF created: ID {c360_srf_id}, Location IDs: {c360_location_ids}") # Step 2: Update Internal LOS try: logger.info(f"{log_prefix} Calling internal LOS update_loan_with_c360_ids.") internal_los_client.update_loan_with_c360_ids(internal_loan_id, c360_srf_id, c360_location_ids) logger.info(f"{log_prefix} Internal LOS updated successfully for loan {internal_loan_id}.") except InternalLOSClientError as e: logger.error(f"{log_prefix} Failed to update internal LOS for loan {internal_loan_id}: {e}", exc_info=True) # TODO: Implement compensating action? (e.g., flag C360 SRF, notify admin) # For now, just raise to potentially retry the whole task raise self.retry(exc=e, countdown=60 * 5) # Retry after 5 mins for LOS update failure except Collateral360ClientError as e: logger.error(f"{log_prefix} Collateral360 API error: {e}", exc_info=True) # Retry based on Celery task settings for potentially transient C360 errors # Check if status code indicates a retry might help if e.status_code in [429, 500, 502, 503, 504]: raise self.retry(exc=e) else: # Non-retryable C360 error (e.g., 400 Bad Request), fail permanently logger.error(f"{log_prefix} Non-retryable C360 error encountered.") # TODO: Update internal LOS status to 'Failed'? pass # Let the task fail except Exception as e: logger.error(f"{log_prefix} Unexpected error: {e}", exc_info=True) # Retry for unexpected errors raise self.retry(exc=e) logger.info(f"{log_prefix} Successfully processed create service request for internal loan ID {internal_loan_id}.") # Optionally return results if needed, but ensure CELERY_RESULT_BACKEND is configured return {"status": "success", "c360_srf_id": c360_srf_id, "c360_location_ids": c360_location_ids} @shared_task(bind=True, max_retries=3, default_retry_delay=30) def process_webhook_event(self, payload: dict, correlation_id: str): """ Celery task to process incoming webhook events from Collateral360. """ log_prefix = f"[{correlation_id}] Task process_webhook_event:" logger.info(f"{log_prefix} Received webhook payload: {payload}") event_type = payload.get('event', {}).get('type') # Example structure, adjust as needed event_data = payload.get('data', {}) try: if event_type == 'service.status.changed': # Example event type logger.info(f"{log_prefix} Processing service status change event.") # Extract relevant IDs (e.g., serviceRequestID, locationServiceID, new status) service_request_id = event_data.get('serviceRequestID') location_service_id = event_data.get('locationServiceID') new_status = event_data.get('status') # TODO: Implement logic to update internal LOS based on this status change # internal_los_client = InternalLOSClient() # internal_los_client.update_service_status(location_service_id, new_status) logger.info(f"{log_prefix} Placeholder: Processed status change for LSID {location_service_id} to {new_status}.") # Add handlers for other event types # elif event_type == 'another.event': # pass else: logger.warning(f"{log_prefix} Received unhandled webhook event type: {event_type}") except Exception as e: # Log error and potentially retry logger.error(f"{log_prefix} Error processing webhook event type {event_type}: {e}", exc_info=True) raise self.retry(exc=e) logger.info(f"{log_prefix} Finished processing webhook event type {event_type}.") from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import status, permissions from rest_framework.authentication import TokenAuthentication from django.views.decorators.csrf import csrf_exempt from django.utils.decorators import method_decorator from django.http import HttpResponseBadRequest, HttpResponse, JsonResponse from django.core.cache import cache from django.conf import settings import logging import json from .utils import verify_webhook_signature from .tasks import process_create_service_request, process_webhook_event # Import Celery tasks from .idempotency import idempotent_view # Import idempotency decorator from .middleware import get_correlation_id # Import function to get correlation ID # Import serializers if you create them # from .serializers import CreateServiceRequestSerializer logger = logging.getLogger(__name__) # Example View using Authentication, Idempotency, and Async Task class CreateServiceRequestView(APIView): """ API endpoint to initiate the creation of a Service Request in Collateral360. Receives loan/collateral data, validates it, triggers an asynchronous task to perform the creation in C360 and update the internal LOS, and returns a task ID for status tracking. Requires Token Authentication. Supports Idempotency via 'Idempotency-Key' header. """ authentication_classes = [TokenAuthentication] permission_classes = [permissions.IsAuthenticated] # Add serializer_class = CreateServiceRequestSerializer if using DRF serializers @method_decorator(idempotent_view) # Apply idempotency decorator def post(self, request, *args, **kwargs): correlation_id = get_correlation_id() log_prefix = f"[{correlation_id}] CreateServiceRequestView:" logger.info(f"{log_prefix} Received request.") # 1. Validate incoming request data (using DRF serializers is recommended) # serializer = CreateServiceRequestSerializer(data=request.data) # if not serializer.is_valid(): # logger.warning(f"{log_prefix} Invalid request data: {serializer.errors}") # return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) # validated_data = serializer.validated_data # internal_loan_id = validated_data.get("internal_loan_id") # Get internal ID from validated data # --- Placeholder Validation --- validated_data = request.data # Use raw data for now internal_loan_id = validated_data.get("internal_loan_id") # Assume internal ID is passed if not internal_loan_id: logger.warning(f"{log_prefix} Missing 'internal_loan_id' in request.") return Response({"error": "Missing 'internal_loan_id'"}, status=status.HTTP_400_BAD_REQUEST) # TODO: Add more robust validation # --- End Placeholder Validation --- # 2. Prepare Payload for C360 (Mapping logic goes here) # This is highly dependent on your incoming data structure vs C360's requirements try: # Example: Extract and map data - replace with actual logic c360_payload = { "meta": { "requestedBy": validated_data.get("requested_by_email", "default_user@example.com"), # Get from request or default "createdBy": validated_data.get("created_by_email", "default_user@example.com"), "cabinet": validated_data.get("c360_cabinet", "Default Cabinet"), "currency": validated_data.get("currency_code", "USD"), "srfAction": validated_data.get("srf_action", "PROCURE"), # 'DRAFT' or 'PROCURE' "processAsWarnings": validated_data.get("process_as_warnings", True) }, "data": { "transaction": validated_data.get("transaction_details", {}), # Map transaction fields "collaterals": validated_data.get("collateral_details", []) # Map collateral fields including services } } # TODO: Add detailed mapping logic based on C360 schema and incoming data logger.debug(f"{log_prefix} Prepared C360 payload.") except Exception as e: logger.error(f"{log_prefix} Error preparing C360 payload: {e}", exc_info=True) return Response({"error": "Failed to process request data"}, status=status.HTTP_400_BAD_REQUEST) # 3. Trigger Asynchronous Task try: task = process_create_service_request.delay( c360_payload=c360_payload, internal_loan_id=internal_loan_id, correlation_id=correlation_id # Pass correlation ID to task ) logger.info(f"{log_prefix} Dispatched task {task.id} for internal loan {internal_loan_id}.") # 4. Return Accepted response with Task ID response_data = { "message": "Service request creation initiated.", "task_id": task.id } # Return JsonResponse directly so idempotency decorator can cache it return JsonResponse(response_data, status=status.HTTP_202_ACCEPTED) except Exception as e: logger.error(f"{log_prefix} Failed to dispatch Celery task: {e}", exc_info=True) return Response({"error": "Failed to initiate service request creation"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR) # Mark this view as exempt from CSRF protection as Collateral360 won't send a CSRF token. # Ensure security via HMAC signature verification instead. @method_decorator(csrf_exempt, name='dispatch') class Collateral360WebhookReceiverView(APIView): """ Receives incoming webhook notifications from Collateral360. Verifies the HMAC-SHA256 signature, enqueues the event payload for asynchronous processing, and responds with HTTP 200 OK immediately to acknowledge receipt. """ # Allow any access for webhook endpoint; security is via HMAC signature authentication_classes = [] permission_classes = [permissions.AllowAny] def post(self, request, *args, **kwargs): correlation_id = get_correlation_id() # Get correlation ID from middleware log_prefix = f"[{correlation_id}] WebhookReceiver:" # 1. Verify Signature if not verify_webhook_signature(request): logger.warning(f"{log_prefix} Invalid webhook signature received.") return HttpResponseBadRequest("Invalid signature.") # 2. Parse Payload try: payload = json.loads(request.body) logger.info(f"{log_prefix} Received valid webhook payload.") logger.debug(f"{log_prefix} Payload content: {payload}") # Log full payload only if needed and secure except json.JSONDecodeError: logger.error(f"{log_prefix} Failed to parse webhook JSON payload.") # Still return 200 OK to acknowledge receipt, but log the error. return HttpResponse("Webhook received (payload parsing error).", status=status.HTTP_200_OK) except Exception as e: logger.error(f"{log_prefix} Unexpected error reading webhook payload: {e}", exc_info=True) # Still return 200 OK return HttpResponse("Webhook received (internal processing error).", status=status.HTTP_200_OK) # 3. Trigger Asynchronous Task for Processing try: process_webhook_event.delay(payload=payload, correlation_id=correlation_id) logger.info(f"{log_prefix} Dispatched task for webhook event processing.") except Exception as e: # Log the error but still acknowledge receipt to C360 logger.error(f"{log_prefix} Failed to dispatch Celery task for webhook: {e}", exc_info=True) # 4. Acknowledge Receipt Immediately (CRITICAL) # Send 200 OK *before* any potentially long-running processing. return HttpResponse("Webhook received successfully.", status=status.HTTP_200_OK) from django.urls import path from . import views # Define app_name if you plan to use namespacing, e.g., {% url 'orchestration_api:create_srf' %} # app_name = 'orchestration_api' urlpatterns = [ # Orchestration endpoints (require auth) path('service-requests/create/', views.CreateServiceRequestView.as_view(), name='create_srf'), # Add other orchestration endpoints here... # path('service-requests//status/', views.GetServiceRequestStatusView.as_view(), name='get_srf_status'), # Webhook receiver endpoint (no auth, security via HMAC) path('webhooks/collateral360/', views.Collateral360WebhookReceiverView.as_view(), name='webhook_receiver_c360'), ] """ URL configuration for loan_orchestrator project. """ from django.contrib import admin from django.urls import path, include from drf_spectacular.views import SpectacularAPIView, SpectacularRedocView, SpectacularSwaggerView urlpatterns = [ path('admin/', admin.site.urls), # Orchestration API endpoints path('api/v1/', include('orchestration_api.urls')), # Include app URLs # API Schema: path('api/schema/', SpectacularAPIView.as_view(), name='schema'), # Optional UI: path('api/schema/swagger-ui/', SpectacularSwaggerView.as_view(url_name='schema'), name='swagger-ui'), path('api/schema/redoc/', SpectacularRedocView.as_view(url_name='schema'), name='redoc'), ] **Explanation and Next Steps:** 1. **Dependencies:** Added `celery`, `redis`, `django-redis`, `drf-spectacular`, `urllib3`. Run `pip install -r requirements.txt`. 2. **Celery:** Basic Celery setup is included (`celery.py`, `__init__.py`, `settings.py`). You need a running Redis server (or other broker) specified in `CELERY_BROKER_URL` and `CELERY_RESULT_BACKEND` (e.g., in your `.env` file). Start a Celery worker: `celery -A loan_orchestrator worker -l INFO`. 3. **Redis Cache:** Configured Redis as the default cache backend via `django-redis`. Ensure the `CACHE_URL` in your environment points to your Redis instance (can be the same as Celery's, but use different database numbers). 4. **Correlation ID:** Middleware (`CorrelationIDMiddleware`) generates/retrieves an ID per request. A logging filter (`CorrelationIDFilter`) adds it to log records. The ID is passed explicitly to Celery tasks. 5. **Idempotency:** The `idempotent_view` decorator uses the Django cache (Redis) to store responses keyed by the `Idempotency-Key` header. It's applied to the example `CreateServiceRequestView`. 6. **HTTP Client Retries:** A `BaseClient` was introduced. `Collateral360Client` and `InternalLOSClient` now inherit from it, automatically getting session management and retry logic configured via `settings.py`. 7. **Asynchronous Tasks:** * `process_create_service_request`: Handles the C360 SRF creation and internal LOS update. Includes basic retry logic within the task for specific failures. * `process_webhook_event`: Handles incoming webhooks asynchronously. 8. **Views:** * `CreateServiceRequestView`: Now validates, prepares payload, triggers `process_create_service_request`, and returns `202 Accepted` with a `task_id`. Added `TokenAuthentication`. * `Collateral360WebhookReceiverView`: Verifies signature, triggers `process_webhook_event`, returns `200 OK`. Authentication/permission set to `AllowAny`. 9. **Authentication:** Default DRF settings now require `TokenAuthentication`. You'll need to generate tokens for your internal clients (`python manage.py drf_create_token `) and include them in the `Authorization: Token ` header when calling your orchestrator API endpoints (like `/api/v1/service-requests/create/`). 10. **API Documentation:** `drf-spectacular` is configured. Access the schema at `/api/schema/`, Swagger UI at `/api/schema/swagger-ui/`, and ReDoc at `/api/schema/redoc/`. 11. **Logging:** Updated format to include correlation ID. This provides a much more robust and production-ready structure. Remember to configure your `.env` file, run the necessary services (Redis, Celery worker), and implement the specific data mapping logic (`TODO` comments).