`dataenginex.api`

Reusable API components — auth, health, errors, pagination, rate limiting.

Public API::

from dataenginex.api import (
    HealthChecker, HealthStatus, ComponentHealth,
    APIHTTPException, BadRequestError, NotFoundError, ServiceUnavailableError,
    PaginatedResponse, paginate,
    AuthMiddleware, AuthUser, create_token, decode_token,
    RateLimiter, RateLimitMiddleware,
)

Routers are defined in application packages that build on dataenginex.

Requires the [api] extra::

pip install dataenginex[api]

`AuthMiddleware`

Bases: BaseHTTPMiddleware

Starlette middleware that enforces JWT auth when enabled.

When DEX_AUTH_ENABLED is "true" (case-insensitive), every request to a non-public path must carry a valid Authorization: Bearer <token> header. The decoded claims are stored on request.state.auth_user.

Source code in src/dataenginex/api/auth.py

class AuthMiddleware(BaseHTTPMiddleware):
    """Starlette middleware that enforces JWT auth when enabled.

    When ``DEX_AUTH_ENABLED`` is ``"true"`` (case-insensitive), every request
    to a non-public path must carry a valid ``Authorization: Bearer <token>``
    header. The decoded claims are stored on ``request.state.auth_user``.
    """

    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
        """Validate the bearer token and attach ``auth_user`` to request state."""
        enabled = os.getenv("DEX_AUTH_ENABLED", "false").lower() == "true"
        if not enabled:
            return await call_next(request)

        # Skip public endpoints
        if request.url.path in _PUBLIC_PATHS:
            return await call_next(request)

        secret = os.getenv("DEX_JWT_SECRET", "")
        if not secret:
            logger.error("DEX_AUTH_ENABLED=true but DEX_JWT_SECRET is not set")
            return JSONResponse(
                status_code=500,
                content={"error": "auth_config_error", "message": "Auth secret not configured"},
            )

        auth_header = request.headers.get("Authorization", "")
        if not auth_header.startswith("Bearer "):
            return JSONResponse(
                status_code=401,
                content={"error": "unauthorized", "message": "Missing bearer token"},
            )

        token = auth_header[7:]
        try:
            claims = decode_token(token, secret)
        except ValueError:
            logger.exception("JWT validation failed")
            return JSONResponse(
                status_code=401,
                content={
                    "error": "unauthorized",
                    "message": "Invalid or expired authentication token",
                },
            )

        request.state.auth_user = AuthUser(
            sub=claims.get("sub", "anonymous"),
            roles=claims.get("roles", []),
            claims=claims,
        )
        return await call_next(request)

`dispatch(request, call_next)` `async`

Validate the bearer token and attach auth_user to request state.

Source code in src/dataenginex/api/auth.py

async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
    """Validate the bearer token and attach ``auth_user`` to request state."""
    enabled = os.getenv("DEX_AUTH_ENABLED", "false").lower() == "true"
    if not enabled:
        return await call_next(request)

    # Skip public endpoints
    if request.url.path in _PUBLIC_PATHS:
        return await call_next(request)

    secret = os.getenv("DEX_JWT_SECRET", "")
    if not secret:
        logger.error("DEX_AUTH_ENABLED=true but DEX_JWT_SECRET is not set")
        return JSONResponse(
            status_code=500,
            content={"error": "auth_config_error", "message": "Auth secret not configured"},
        )

    auth_header = request.headers.get("Authorization", "")
    if not auth_header.startswith("Bearer "):
        return JSONResponse(
            status_code=401,
            content={"error": "unauthorized", "message": "Missing bearer token"},
        )

    token = auth_header[7:]
    try:
        claims = decode_token(token, secret)
    except ValueError:
        logger.exception("JWT validation failed")
        return JSONResponse(
            status_code=401,
            content={
                "error": "unauthorized",
                "message": "Invalid or expired authentication token",
            },
        )

    request.state.auth_user = AuthUser(
        sub=claims.get("sub", "anonymous"),
        roles=claims.get("roles", []),
        claims=claims,
    )
    return await call_next(request)

`AuthUser` `dataclass`

Resolved identity from a valid JWT.

Source code in src/dataenginex/api/auth.py

@dataclass
class AuthUser:
    """Resolved identity from a valid JWT."""

    sub: str
    roles: list[str]
    claims: dict[str, Any]

`APIHTTPException`

Bases: HTTPException

Base HTTP exception with error code and details.

Source code in src/dataenginex/api/errors.py

class APIHTTPException(HTTPException):
    """Base HTTP exception with error code and details."""

    def __init__(
        self,
        status_code: int,
        message: str,
        code: str = "api_error",
        details: list[ErrorDetail] | None = None,
    ) -> None:
        self.code = code
        self.details = details
        super().__init__(status_code=status_code, detail=message)

`BadRequestError`

Bases: APIHTTPException

Raised for 400 validation or malformed requests.

Source code in src/dataenginex/api/errors.py

class BadRequestError(APIHTTPException):
    """Raised for 400 validation or malformed requests."""

    def __init__(
        self,
        message: str = "Bad request",
        details: list[ErrorDetail] | None = None,
    ) -> None:
        super().__init__(
            status_code=status.HTTP_400_BAD_REQUEST,
            message=message,
            code="bad_request",
            details=details,
        )

`NotFoundError`

Bases: APIHTTPException

Raised for 404 not found errors.

Source code in src/dataenginex/api/errors.py

class NotFoundError(APIHTTPException):
    """Raised for 404 not found errors."""

    def __init__(
        self,
        message: str = "Resource not found",
        details: list[ErrorDetail] | None = None,
    ) -> None:
        super().__init__(
            status_code=status.HTTP_404_NOT_FOUND,
            message=message,
            code="not_found",
            details=details,
        )

`ServiceUnavailableError`

Bases: APIHTTPException

Raised when a dependency is unavailable.

Source code in src/dataenginex/api/errors.py

class ServiceUnavailableError(APIHTTPException):
    """Raised when a dependency is unavailable."""

    def __init__(
        self,
        message: str = "Service unavailable",
        details: list[ErrorDetail] | None = None,
    ) -> None:
        super().__init__(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            message=message,
            code="service_unavailable",
            details=details,
        )