# Error Handling Mechanisms in Daily Stock Analysis: A 5-Layer Defense Strategy

> Discover the 5-layer defense strategy in ZhuLinsen/daily_stock_analysis for robust error handling. Learn about FastAPI middleware, Pydantic, custom exceptions, retries, and logging for resilient API operations.

- Repository: [mumu/daily_stock_analysis](https://github.com/ZhuLinsen/daily_stock_analysis)
- Tags: best-practices
- Published: 2026-04-30

---

**The ZhuLinsen/daily_stock_analysis repository implements a layered error handling strategy combining FastAPI middleware, Pydantic schemas, custom exception hierarchies, tenacity-based retries, and structured logging to ensure resilient API operations.**

This comprehensive guide examines the error handling mechanisms implemented across the daily_stock_analysis codebase. The repository employs a sophisticated multi-tiered approach that catches exceptions at the API boundary, enforces typed error responses, and implements domain-specific retry logic for network and database operations.

## Centralized Exception Middleware

The foundation of the error handling strategy rests in [`api/middlewares/error_handler.py`](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/api/middlewares/error_handler.py), where a global middleware intercepts all unhandled exceptions before they reach the client. The `ErrorHandlerMiddleware` class extends `BaseHTTPMiddleware` to wrap every request in a try-catch block, ensuring consistent JSON error responses regardless of where an exception originates.

```python

# api/middlewares/error_handler.py

class ErrorHandlerMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        try:
            return await call_next(request)
        except Exception as e:
            logger.error(
                f"未处理的异常: {e}\n"
                f"请求路径: {request.url.path}\n"
                f"请求方法: {request.method}\n"
                f"堆栈: {traceback.format_exc()}"
            )
            return JSONResponse(
                status_code=500,
                content={
                    "error": "internal_error",
                    "message": "服务器内部错误，请稍后重试",
                    "detail": str(e) if logger.isEnabledFor(logging.DEBUG) else None,
                },
            )

```

The middleware registers via `add_error_handlers(app)`, which adds specific handlers for `HTTPException` and validation errors. This ensures that every error—whether a validation failure or a catastrophic server crash—follows the same JSON format, preventing information leakage while maintaining debuggability.

## Structured Error Response Schemas

All API responses conform to a strict Pydantic model defined in [`api/v1/schemas/common.py`](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/api/v1/schemas/common.py). The `ErrorResponse` base class enforces a machine-readable structure that clients can reliably parse, distinguishing between user-facing messages and internal debug details.

```python

# api/v1/schemas/common.py

class ErrorResponse(BaseModel):
    error: str          # machine-readable code

    message: str        # short description

    detail: Optional[str] = None   # extra debug info (only when DEBUG)

```

Subclasses like `SystemConfigValidationErrorResponse` extend this base for specific domains while maintaining the core contract. The `detail` field is conditionally populated based on the logging level, ensuring that sensitive stack traces never expose internal implementation details in production environments.

## Domain-Specific Exception Hierarchy

Rather than raising generic `Exception` objects, the codebase implements a rich hierarchy of custom exceptions that communicate specific failure modes. These domain exceptions live in their respective service modules and enable precise error handling logic.

### Configuration Exceptions

Located in [`src/services/system_config_service.py`](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/src/services/system_config_service.py), these handle system setup failures:

- **ConfigValidationError**: Raised when configuration files fail schema validation
- **ConfigConflictError**: Triggered when configuration updates create state conflicts  
- **ConfigImportError**: Handles failures during external configuration imports

### Data Provider Exceptions

Defined in [`src/data_provider/base.py`](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/src/data_provider/base.py), these manage external API interactions:

- **DataFetchError**: General wrapper for failed data retrieval operations
- **RateLimitError**: Specific handling for API throttling scenarios
- **DataSourceUnavailableError**: Indicates complete upstream service failures

### Business Logic Exceptions

Service-layer exceptions enforce domain rules:

- **DuplicateTaskError** ([`src/services/task_queue.py`](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/src/services/task_queue.py)): Prevents redundant task processing
- **PortfolioConflictError** ([`src/services/portfolio_service.py`](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/src/services/portfolio_service.py)): Handles concurrent modification conflicts
- **PortfolioOversellError**: Enforces trading constraints when sell orders exceed holdings
- **DuplicateTradeUidError** ([`src/repositories/portfolio_repo.py`](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/src/repositories/portfolio_repo.py)): Ensures database-level trade uniqueness
- **PortfolioBusyError**: Signals when the portfolio is locked by another operation

## Retry Strategies with Tenacity

Transient failures from network I/O or rate limiting are handled using the **tenacity** library. The `@retry` decorator implements exponential back-off with jitter, automatically retrying operations on whitelisted exception types.

In [`src/services/social_sentiment_service.py`](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/src/services/social_sentiment_service.py), external API calls are wrapped with configurable retry logic:

```python

# src/services/social_sentiment_service.py

@retry(
    retry=retry_if_exception_type(_TRANSIENT_EXCEPTIONS),
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
)
def _get_with_retry(url: str, *, headers: Dict[str, str], params: Optional[Dict[str, Any]] = None):
    """GET with retry on transient network errors."""
    resp = requests.get(url, headers=headers, params=params, timeout=10)
    resp.raise_for_status()
    return resp

```

Similar tenacity-based patterns appear in [`src/search_service.py`](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/src/search_service.py) and various data provider fetchers, ensuring that temporary network blips do not cascade into system failures.

## Database Write Resilience

SQLite operations in [`src/storage.py`](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/src/storage.py) implement custom retry logic specifically for `sqlite3.OperationalError`, which typically indicates database locks or busy states. The storage layer reads configuration values `sqlite_write_retry_max` and `sqlite_write_retry_base_delay` to determine retry behavior.

```python

# src/storage.py (excerpt)

max_retries = self._sqlite_write_retry_max if self._is_sqlite_engine else 0
for attempt in range(max_retries + 1):
    try:
        # perform DB write

        break
    except sqlite3.OperationalError as exc:
        if attempt == max_retries:
            raise
        delay = self._sqlite_write_retry_base_delay * (2 ** attempt)
        time.sleep(delay)

```

This exponential back-off strategy prevents thundering herd problems when multiple processes contend for the SQLite database file.

## Notification Sender Reliability

The Telegram notification system in [`src/notification_sender/telegram_sender.py`](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/src/notification_sender/telegram_sender.py) implements custom retry logic respecting the `Retry-After` header for rate-limited requests. This approach combines fixed-delay and exponential back-off strategies.

```python

# src/notification_sender/telegram_sender.py

def _send_message(...):
    for attempt in range(5):
        response = requests.post(...)
        if response.status_code == 429:          # rate limited

            retry_after = int(response.headers.get('Retry-After', 2 ** attempt))
            logger.warning(f"Telegram rate limited, retrying in {retry_after}s ...")
            time.sleep(retry_after)
            continue
        response.raise_for_status()
        return response

```

By parsing the upstream API's rate-limit headers, the system optimizes retry timing rather than blindly applying exponential back-off.

## Logging and Observability

The error handling mechanisms integrate with a centralized logging configuration. The global middleware captures full tracebacks and request metadata (path, method, headers) for every unhandled exception. Service-layer code logs contextual information at appropriate severity levels before re-raising or handling exceptions.

The conditional inclusion of the `detail` field in error responses—only when `logger.isEnabledFor(logging.DEBUG)` returns true—ensures that production deployments remain secure while development environments retain full debugging capabilities.

## Summary

The daily_stock_analysis repository demonstrates production-grade error handling through these key mechanisms:

- **Global middleware** in [`api/middlewares/error_handler.py`](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/api/middlewares/error_handler.py) ensures uniform error responses and prevents information leakage
- **Typed Pydantic schemas** enforce consistent API contracts for error responses
- **Custom exception hierarchies** across services enable precise, domain-specific error handling
- **Tenacity decorators** provide resilient retry logic for external network calls
- **SQLite-specific retry loops** handle database contention without external dependencies
- **Rate-limit-aware retries** for notification services respect upstream API constraints
- **Structured logging** captures full context for debugging while respecting production security

## Frequently Asked Questions

### What is the purpose of the ErrorHandlerMiddleware in the FastAPI application?

The `ErrorHandlerMiddleware` serves as the last line of defense for unhandled exceptions in the daily_stock_analysis API. It catches any exception that bubbles up through the request stack and converts it into a standardized JSON response containing `error`, `message`, and conditional `detail` fields. This ensures clients always receive predictable error payloads even when unexpected failures occur, while logging full tracebacks server-side for debugging.

### How does the repository handle transient network failures when fetching stock data?

The codebase uses the **tenacity** library to wrap external HTTP calls with exponential back-off retry logic. Functions like `_get_with_retry` in [`src/services/social_sentiment_service.py`](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/src/services/social_sentiment_service.py) are decorated with `@retry` configurations that specify `stop_after_attempt(3)` and `wait_exponential` parameters. This mechanism specifically targets transient exceptions—such as connection timeouts or temporary 5xx errors—automatically retrying requests before surfacing persistent failures to the calling code.

### Why does the SQLite storage implementation use custom retry logic instead of tenacity?

The SQLite retry logic in [`src/storage.py`](https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/src/storage.py) handles `sqlite3.OperationalError` specifically, which occurs when the database file is locked by another process. Unlike network requests, SQLite contention requires tighter integration with the storage class's internal state. The custom loop respects configuration parameters `sqlite_write_retry_max` and `sqlite_write_retry_base_delay`, applying exponential delays directly through `time.sleep()` without the overhead of a decorator chain. This approach provides granular control over retry counts based on whether the engine is SQLite or another database backend.

### What information is included in production error responses versus development environments?

Production error responses include only `error` (machine-readable code) and `message` (user-friendly description) fields to prevent information leakage. The `detail` field, which may contain stack traces or internal error specifics, is only populated when `logger.isEnabledFor(logging.DEBUG)` returns true. This distinction ensures that production deployments expose minimal attack surface while development instances provide verbose debugging information for troubleshooting.