From 38e065201d969377cc114e5104db6f678133d17a Mon Sep 17 00:00:00 2001 From: peaceshallom37-rgb Date: Wed, 29 Jul 2026 17:37:40 +0000 Subject: [PATCH] refactor(error-handling): tighten 17 bare except Exception sites with typed handling - Add ApexException base class, ApexNotFoundError, ApexTransientError - Make ApexConflictError and ApexValidationError inherit from ApexException - Fix bare except Exception in webhook_tasks.py, sla_tasks.py, jobs.py, outages.py, sla_service.py, lock.py, correlation.py - Each site now uses ApexException subclasses or specific Python types - logger.exception() added for incident context Closes #93 --- app/api/v1/endpoints/jobs.py | 9 ++++++- app/api/v1/endpoints/outages.py | 16 ++++++++---- app/core/exceptions.py | 45 ++++++++++++++++++++++++++++++--- app/core/lock.py | 25 +++++++++++++----- app/middleware/correlation.py | 7 +++-- app/services/sla_service.py | 6 +++++ app/tasks/sla_tasks.py | 14 ++++++++-- app/tasks/webhook_tasks.py | 9 +++++-- 8 files changed, 108 insertions(+), 23 deletions(-) diff --git a/app/api/v1/endpoints/jobs.py b/app/api/v1/endpoints/jobs.py index a38d84c..abede26 100644 --- a/app/api/v1/endpoints/jobs.py +++ b/app/api/v1/endpoints/jobs.py @@ -503,9 +503,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: {str(e)}", diff --git a/app/api/v1/endpoints/outages.py b/app/api/v1/endpoints/outages.py index 5b15429..31e8e4c 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.db.session import get_db @@ -220,8 +221,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") @@ -261,7 +264,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] = [] @@ -269,7 +272,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): @@ -286,9 +289,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: @@ -307,7 +313,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 cea766d..7247669 100644 --- a/app/core/exceptions.py +++ b/app/core/exceptions.py @@ -8,15 +8,52 @@ from sqlalchemy.exc import IntegrityError -class ApexConflictError(Exception): - def __init__(self, detail: str, fields: Optional[Dict[str, str]] = None): +class ApexException(Exception): + """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 = "An unexpected error occurred.", + *, + error_code: str = "internal_error", + status_code: int = 500, + extra: dict[str, Any] | None = None, + ) -> None: self.detail = detail + self.error_code = error_code + self.status_code = status_code + self.extra = extra or {} + super().__init__(detail) + + +class ApexNotFoundError(ApexException): + """Resource not found (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): + """Transient / retryable error (503 by default).""" + + 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): + super().__init__(detail=detail, error_code="conflict", status_code=409) self.fields = fields or {} -class ApexValidationError(Exception): +class ApexValidationError(ApexException): def __init__(self, detail: str, errors: Optional[List[Dict[str, Any]]] = None): - self.detail = detail + super().__init__(detail=detail, error_code="validation_error", status_code=422) self.errors = errors or [] diff --git a/app/core/lock.py b/app/core/lock.py index 2453155..ca79a13 100644 --- a/app/core/lock.py +++ b/app/core/lock.py @@ -8,17 +8,23 @@ from __future__ import annotations import hashlib +import logging from contextlib import contextmanager from typing import Generator 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 +72,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 +111,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 c87940a..0f68bdb 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") @@ -64,8 +65,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 @@ -84,6 +85,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 be36a22..57c1da0 100644 --- a/app/services/sla_service.py +++ b/app/services/sla_service.py @@ -1,5 +1,6 @@ from __future__ import annotations +import logging from datetime import datetime, timezone from typing import Any, Dict, List, Optional from sqlalchemy.orm import Session @@ -10,6 +11,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.""" @@ -202,7 +205,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 19eabdc..af7543a 100644 --- a/app/tasks/sla_tasks.py +++ b/app/tasks/sla_tasks.py @@ -10,6 +10,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 @@ -179,6 +180,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) @@ -188,7 +191,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() @@ -243,6 +246,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)}) @@ -295,6 +303,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) @@ -304,7 +314,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 8dba02d..aa23b53 100644 --- a/app/tasks/webhook_tasks.py +++ b/app/tasks/webhook_tasks.py @@ -5,6 +5,7 @@ from app.tasks.celery_app import celery_app from app.db.session import SessionLocal from app.services.audit_log import audit_log +from app.core.exceptions import ApexTransientError logger = logging.getLogger(__name__) @@ -24,6 +25,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 +39,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 +82,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 +101,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()