diff --git a/app/api/v1/endpoints/jobs.py b/app/api/v1/endpoints/jobs.py index 323d4cb..6f7b042 100644 --- a/app/api/v1/endpoints/jobs.py +++ b/app/api/v1/endpoints/jobs.py @@ -501,9 +501,16 @@ def retry_job( message=f"Job retry #{job.retry_count} initiated successfully", ) + except (ValueError, KeyError, TypeError) as e: + db.rollback() + logger.error("Failed to retry job due to data issue", job_id=str(job.id), error=str(e), correlation_id=correlation_id) + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Failed to retry job due to invalid payload: {str(e)}", + ) except Exception as e: db.rollback() - logger.error("Failed to retry job", job_id=str(job.id), error=str(e), correlation_id=correlation_id) + logger.exception("Unexpected error retrying job", job_id=str(job.id), correlation_id=correlation_id) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to retry job: {e!s}", diff --git a/app/api/v1/endpoints/outages.py b/app/api/v1/endpoints/outages.py index cfe790d..eedd354 100644 --- a/app/api/v1/endpoints/outages.py +++ b/app/api/v1/endpoints/outages.py @@ -5,6 +5,7 @@ from fastapi import APIRouter, Depends, File, HTTPException, Query, Response, UploadFile from pydantic import ValidationError +from sqlalchemy.exc import IntegrityError, SQLAlchemyError from sqlalchemy.orm import Session from app.api.v1.endpoints.sla import _invalidate_analytics_cache @@ -244,8 +245,10 @@ async def import_outages( elif filename.endswith(".csv"): try: rows = list(csv.DictReader(io.StringIO(content.decode("utf-8")))) - except Exception as exc: + except (csv.Error, UnicodeDecodeError) as exc: raise HTTPException(status_code=400, detail=f"Invalid CSV: {exc}") from exc + except Exception as exc: + raise HTTPException(status_code=400, detail=f"CSV processing failed: {exc}") from exc else: raise HTTPException(status_code=400, detail="Unsupported file format. Use .json or .csv") @@ -285,7 +288,7 @@ async def import_outages( duplicate=False, ) ) - except Exception as exc: + except (ValidationError, ValueError, TypeError, KeyError) as exc: row_outcomes.append(_row_error(i, row, exc)) elif consistency == ImportConsistency.atomic: parsed: list[OutageCreate] = [] @@ -293,7 +296,7 @@ async def import_outages( try: parsed.append(OutageCreate(**row)) row_outcomes.append(ImportRowResult(row=i, id=row.get("id"), status="ok")) - except Exception as exc: + except (ValidationError, ValueError, TypeError, KeyError) as exc: row_outcomes.append(_row_error(i, row, exc)) if any(r.status == "error" for r in row_outcomes): @@ -310,9 +313,12 @@ async def import_outages( row_outcomes[i].duplicate = True row_outcomes[i].existing_id = created.id db.commit() - except Exception as exc: + except (IntegrityError, SQLAlchemyError) as exc: db.rollback() raise HTTPException(status_code=500, detail=f"Transaction failed: {exc}") from exc + except Exception as exc: + db.rollback() + raise HTTPException(status_code=500, detail=f"Unexpected import error: {exc}") from exc else: for i, row in enumerate(rows): try: @@ -331,7 +337,7 @@ async def import_outages( ) if persisted: persisted_count += 1 - except Exception as exc: + except (ValidationError, ValueError, TypeError, KeyError) as exc: db.rollback() row_outcomes.append(_row_error(i, row, exc)) diff --git a/app/core/exceptions.py b/app/core/exceptions.py index 66c991a..bb1ed21 100644 --- a/app/core/exceptions.py +++ b/app/core/exceptions.py @@ -8,92 +8,53 @@ from sqlalchemy.exc import IntegrityError -from app.utils.correlation_ctx import get_or_generate_correlation_id - - class ApexException(Exception): -114-117-webhook-concurrency-canonical-json - status_code: int = 500 - error_code: str = "error" - extra: dict[str, Any] | None = None + """Base exception for all ApexChainx domain errors. + + Every raised exception in the codebase should subclass this or one + of its children so that middleware and error handlers can act on + typed conditions rather than bare ``except Exception``. + """ def __init__( self, - detail: str, + detail: str = "An unexpected error occurred.", *, + error_code: str = "internal_error", status_code: int = 500, - error_code: str = "error", extra: dict[str, Any] | None = None, - ): + ) -> None: self.detail = detail - self.status_code = status_code self.error_code = error_code - self.extra = extra - super().__init__(detail) - - """Base exception for all ApexChainx domain errors.""" - - def __init__(self, detail: str, status_code: int = 500): - self.detail = detail self.status_code = status_code + self.extra = extra or {} + super().__init__(detail) class ApexNotFoundError(ApexException): - """Raised when a requested resource does not exist.""" + """Resource not found (404).""" - def __init__(self, detail: str = "Resource not found"): - super().__init__(detail=detail, status_code=404) + def __init__(self, detail: str = "Resource not found.", **kwargs: Any) -> None: + super().__init__(detail=detail, error_code="not_found", status_code=404, **kwargs) class ApexTransientError(ApexException): - """Raised for transient infrastructure errors that may succeed on retry.""" + """Transient / retryable error (503 by default).""" - def __init__(self, detail: str = "A transient error occurred. Please retry."): - super().__init__(detail=detail, status_code=503) -main + def __init__(self, detail: str = "A transient error occurred.", **kwargs: Any) -> None: + super().__init__(detail=detail, error_code="transient_error", status_code=503, **kwargs) class ApexConflictError(ApexException): def __init__(self, detail: str, fields: Optional[Dict[str, str]] = None): -114-117-webhook-concurrency-canonical-json - super().__init__( - detail, - status_code=409, - error_code="conflict", - extra={"fields": fields or {}}, - ) - - super().__init__(detail=detail, status_code=409) + super().__init__(detail=detail, error_code="conflict", status_code=409) self.fields = fields or {} main class ApexValidationError(ApexException): def __init__(self, detail: str, errors: Optional[List[Dict[str, Any]]] = None): -issue/114-117-webhook-concurrency-canonical-json - super().__init__( - detail, - status_code=422, - error_code="validation_error", - extra={"errors": errors or []}, - ) - - -class ApexNotFoundError(ApexException): - def __init__(self, detail: str = "Resource not found"): - super().__init__(detail, status_code=404, error_code="not_found") - - -class ApexTransientError(ApexException): - def __init__(self, detail: str = "A transient error occurred"): - super().__init__( - detail, - status_code=500, - error_code="transient_error", - extra={"retryable": True}, - ) - - super().__init__(detail=detail, status_code=422) + super().__init__(detail=detail, error_code="validation_error", status_code=422) self.errors = errors or [] main diff --git a/app/core/lock.py b/app/core/lock.py index 25ed05b..badc633 100644 --- a/app/core/lock.py +++ b/app/core/lock.py @@ -8,17 +8,22 @@ from __future__ import annotations import hashlib -from collections.abc import Generator +import logging from contextlib import contextmanager from sqlalchemy import text from sqlalchemy.orm import Session +from app.core.exceptions import ApexTransientError -class ConcurrencyLockError(Exception): +logger = logging.getLogger(__name__) + + +class ConcurrencyLockError(ApexTransientError): """Raised when a lock cannot be acquired.""" - pass + def __init__(self, detail: str = "Could not acquire lock.") -> None: + super().__init__(detail=detail, error_code="concurrency_lock", status_code=409) def _lock_id_from_key(key: str) -> int: @@ -66,9 +71,11 @@ def advisory_lock(db: Session, lock_key: str, timeout_seconds: float = 5.0) -> G try: yield - except Exception: - # Lock is automatically released on transaction rollback - raise + except ApexTransientError: + raise # let domain errors propagate naturally + except Exception as exc: + logger.exception("Error inside advisory lock for '%s'", lock_key) + raise ApexTransientError(detail=f"Unexpected error in locked section: {exc}") from exc @contextmanager @@ -103,5 +110,8 @@ def advisory_lock_nowait(db: Session, lock_key: str) -> Generator[None, None, No try: yield - except Exception: - raise + except ApexTransientError: + raise # let domain errors propagate + except Exception as exc: + logger.exception("Error inside advisory lock (nowait) for '%s'", lock_key) + raise ApexTransientError(detail=f"Unexpected error in locked section: {exc}") from exc diff --git a/app/middleware/correlation.py b/app/middleware/correlation.py index 951e00e..f666736 100644 --- a/app/middleware/correlation.py +++ b/app/middleware/correlation.py @@ -4,6 +4,7 @@ from app.utils.correlation_ctx import get_or_generate_correlation_id, set_correlation_id from app.utils.logging import get_structured_logger +from app.core.exceptions import ApexException logger = get_structured_logger("access") @@ -65,8 +66,8 @@ async def send_wrapper(message): uid = str(getattr(request.state.user, "id", "")) if uid: user_id_hash = _hash_value(uid) - except Exception: - pass + except (AttributeError, TypeError, ValueError): + pass # user object may not have expected shape; non-critical for logging duration_ms = (time.time() - start_time) * 1000 @@ -85,6 +86,8 @@ async def send_wrapper(message): try: await self.app(scope, receive, send_wrapper) + except ApexException: + raise # already typed; let FastAPI exception handlers process it except Exception as exc: duration_ms = (time.time() - start_time) * 1000 logger.error( diff --git a/app/services/sla_service.py b/app/services/sla_service.py index 1a59785..b2f8ad4 100644 --- a/app/services/sla_service.py +++ b/app/services/sla_service.py @@ -1,6 +1,6 @@ from __future__ import annotations -import re +import logging from datetime import datetime, timezone from typing import Any, Dict, List, Optional, Union from sqlalchemy.orm import Session @@ -12,6 +12,8 @@ from app.services.audit_log import audit_log from app.services.metrics import increment_counter +logger = logging.getLogger(__name__) + class SLAOrchestrator: """Orchestrates SLA computation with real domain logic for outage-centric workflows.""" @@ -226,7 +228,10 @@ def compute_device_sla( record_sla_settlement_audit_events(device_id, period, result, status="succeeded") return result + except ApexTransientError: + raise # already typed, let it propagate except Exception as e: + logger.exception("SLA computation failed for device %s", device_id) audit_log.log( event_type="sla_settlement_failed", details={"device_id": device_id, "period": period, "error": str(e)}, diff --git a/app/tasks/sla_tasks.py b/app/tasks/sla_tasks.py index 6fdb1db..3bb80d4 100644 --- a/app/tasks/sla_tasks.py +++ b/app/tasks/sla_tasks.py @@ -11,6 +11,7 @@ from app.models.job import Job, JobStatus, JobType from app.models.webhook import WebhookEvent from app.services.audit_log import audit_log +from app.core.exceptions import ApexTransientError from app.utils.correlation_ctx import set_correlation_id from app.utils.logging import get_structured_logger @@ -180,6 +181,8 @@ def compute_sla_for_device( logger.info("SLA computation complete for device=%s", device_id) return result + except ApexTransientError: + raise # let Celery handle retry of transient errors except Exception as exc: error_msg = str(exc) logger.exception("SLA computation failed for device=%s: %s", device_id, error_msg) @@ -189,7 +192,7 @@ def compute_sla_for_device( self._log_retry(db, self.request.id, self.request.retries + 1, error_msg) self._mark_failure(db, self.request.id, error_msg) - raise self.retry(exc=exc) + raise self.retry(exc=ApexTransientError(detail=f"SLA computation failed for device {device_id}: {error_msg}")) finally: db.close() @@ -244,6 +247,11 @@ def compute_bulk_sla(self: DatabaseTask, device_ids: list[str], period: str) -> processed_count += 1 + except ApexTransientError as device_exc: + logger.warning("SLA transient error for device=%s: %s", device_id, device_exc) + results.append({"device_id": device_id, "error": str(device_exc)}) + self._add_item_error(db, self.request.id, device_id, str(device_exc)) + error_count += 1 except Exception as device_exc: logger.warning("SLA failed for device=%s: %s", device_id, device_exc) results.append({"device_id": device_id, "error": str(device_exc)}) @@ -296,6 +304,8 @@ def compute_bulk_sla(self: DatabaseTask, device_ids: list[str], period: str) -> logger.info("Bulk SLA computation complete. Violations: %d/%d, Errors: %d", len(violations), total, error_count) return summary + except ApexTransientError: + raise # let Celery handle retry except Exception as exc: error_msg = str(exc) logger.exception("Bulk SLA computation failed: %s", error_msg) @@ -305,7 +315,7 @@ def compute_bulk_sla(self: DatabaseTask, device_ids: list[str], period: str) -> self._log_retry(db, self.request.id, self.request.retries + 1, error_msg) self._mark_failure(db, self.request.id, error_msg) - raise self.retry(exc=exc) + raise self.retry(exc=ApexTransientError(detail=f"Bulk SLA computation failed: {error_msg}")) finally: db.close() diff --git a/app/tasks/webhook_tasks.py b/app/tasks/webhook_tasks.py index aee86a0..f147ee7 100644 --- a/app/tasks/webhook_tasks.py +++ b/app/tasks/webhook_tasks.py @@ -4,7 +4,7 @@ from app.db.session import SessionLocal from app.services.audit_log import audit_log -from app.tasks.celery_app import celery_app +from app.core.exceptions import ApexTransientError logger = logging.getLogger(__name__) @@ -24,6 +24,8 @@ def dispatch_webhook_delivery(self, delivery_id: str) -> dict[str, Any]: dispatch_delivery(db, UUID(delivery_id)) logger.info("Webhook delivery %s dispatched.", delivery_id) return {"delivery_id": delivery_id, "dispatched": True} + except ApexTransientError: + raise # re-raise transient errors so Celery retries them except Exception as exc: error_msg = str(exc) logger.exception("Failed to dispatch webhook delivery %s: %s", delivery_id, error_msg) @@ -36,7 +38,7 @@ def dispatch_webhook_delivery(self, delivery_id: str) -> dict[str, Any]: details={"delivery_id": delivery_id, "retry_count": self.request.retries + 1, "error": error_msg}, ) - raise self.retry(exc=exc) + raise self.retry(exc=ApexTransientError(detail=f"Webhook delivery {delivery_id} failed: {error_msg}")) finally: db.close() @@ -79,6 +81,8 @@ def trigger_sla_violation_async(self, sla_data: Dict[str, Any], event: str = "sl deliveries = trigger_sla_violation_webhooks(db, sla_data=sla_data, event=WebhookEvent(event)) logger.info("Triggered %d webhook deliveries for event=%s.", len(deliveries), event) return {"triggered": len(deliveries), "event": event} + except ApexTransientError: + raise # re-raise transient errors so Celery retries them except Exception as exc: error_msg = str(exc) logger.exception("trigger_sla_violation_async failed: %s", error_msg) @@ -96,6 +100,6 @@ def trigger_sla_violation_async(self, sla_data: Dict[str, Any], event: str = "sl }, ) - raise self.retry(exc=exc) + raise self.retry(exc=ApexTransientError(detail=f"SLA violation webhook failed: {error_msg}")) finally: db.close()