FastAPI + Celery + Next.js + Postgres/Redis app with company monitoring, source collection, LLM-based change analysis, enrichment, and account security (Turnstile, escalating lockout, email verification).
47 lines
1.0 KiB
Python
47 lines
1.0 KiB
Python
"""App-level exceptions, mapped to HTTP responses in one place (main.py).
|
|
|
|
Services raise these instead of `fastapi.HTTPException` so business logic
|
|
stays importable/testable from Celery tasks, which don't have an HTTP
|
|
response to raise into.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
|
|
class AppError(Exception):
|
|
"""Base class for all app-level errors."""
|
|
|
|
|
|
class NotFoundError(AppError):
|
|
pass
|
|
|
|
|
|
class ConflictError(AppError):
|
|
pass
|
|
|
|
|
|
class AuthenticationError(AppError):
|
|
pass
|
|
|
|
|
|
class ForbiddenError(AppError):
|
|
pass
|
|
|
|
|
|
class ValidationAppError(AppError):
|
|
pass
|
|
|
|
|
|
class RateLimitedError(AppError):
|
|
pass
|
|
|
|
|
|
class ThrottledError(RateLimitedError):
|
|
"""Raised by the IP throttle/ban engine (app/services/ip_throttle_service.py)
|
|
- carries a machine-readable retry_after_seconds so the frontend can
|
|
drive a live countdown instead of just showing a generic message."""
|
|
|
|
def __init__(self, message: str, retry_after_seconds: int | None = None) -> None:
|
|
super().__init__(message)
|
|
self.retry_after_seconds = retry_after_seconds
|