Skip to content

dataenginex.api

API helper components — schemas, errors, and pagination.

These are building blocks for applications that expose an HTTP layer on top of dataenginex (e.g. a FastAPI server). They have no dependency on any web framework and can be imported without extras.

Public API::

from dataenginex.api import (
    BadRequestError, NotFoundError, ServiceUnavailableError,
    PaginatedResponse, paginate,
)
from dataenginex.api.schemas import PipelineResultResponse, PredictionRequest, ...

BadRequestError

Bases: DexAPIError

Invalid input or malformed request.

Source code in src/dataenginex/api/errors.py 
25
26
27
28
29
class BadRequestError(DexAPIError):
    """Invalid input or malformed request."""

    def __init__(self, message: str = "Bad request") -> None:
        super().__init__(message, code="bad_request")

DexAPIError

Bases: Exception

Base error for dataenginex API operations.

Source code in src/dataenginex/api/errors.py 
17
18
19
20
21
22
class DexAPIError(Exception):
    """Base error for dataenginex API operations."""

    def __init__(self, message: str, code: str = "api_error") -> None:
        super().__init__(message)
        self.code = code

NotFoundError

Bases: DexAPIError

Requested resource does not exist.

Source code in src/dataenginex/api/errors.py 
32
33
34
35
36
class NotFoundError(DexAPIError):
    """Requested resource does not exist."""

    def __init__(self, message: str = "Resource not found") -> None:
        super().__init__(message, code="not_found")

ServiceUnavailableError

Bases: DexAPIError

A required dependency is unavailable.

Source code in src/dataenginex/api/errors.py 
39
40
41
42
43
class ServiceUnavailableError(DexAPIError):
    """A required dependency is unavailable."""

    def __init__(self, message: str = "Service unavailable") -> None:
        super().__init__(message, code="service_unavailable")

PaginatedResponse

Bases: BaseModel

Generic paginated response wrapper.

Source code in src/dataenginex/api/pagination.py 
43
44
45
46
47
class PaginatedResponse(BaseModel):
    """Generic paginated response wrapper."""

    data: list[Any] = Field(default_factory=list)
    pagination: PaginationMeta

PaginationMeta

Bases: BaseModel

Pagination metadata returned alongside results.

Source code in src/dataenginex/api/pagination.py 
33
34
35
36
37
38
39
40
class PaginationMeta(BaseModel):
    """Pagination metadata returned alongside results."""

    total: int = Field(description="Total number of items")
    limit: int = Field(description="Page size")
    has_next: bool = Field(description="Whether more items exist")
    next_cursor: str | None = Field(None, description="Opaque cursor for next page")
    has_previous: bool = Field(default=False, description="Whether previous items exist")

paginate(items, *, cursor=None, limit=20, max_limit=100)

Slice items and return a PaginatedResponse.

Parameters

items: Full list of items to paginate. cursor: Opaque cursor from a previous response (or None for the first page). limit: Number of items per page. max_limit: Hard ceiling on limit to prevent abuse.

Source code in src/dataenginex/api/pagination.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def paginate(
    items: list[Any],
    *,
    cursor: str | None = None,
    limit: int = 20,
    max_limit: int = 100,
) -> PaginatedResponse:
    """Slice *items* and return a ``PaginatedResponse``.

    Parameters
    ----------
    items:
        Full list of items to paginate.
    cursor:
        Opaque cursor from a previous response (or *None* for the first page).
    limit:
        Number of items per page.
    max_limit:
        Hard ceiling on *limit* to prevent abuse.
    """
    limit = min(max(1, limit), max_limit)

    if cursor:
        try:
            offset = decode_cursor(cursor)
        except ValueError:
            offset = 0  # Reset to first page on invalid cursor
    else:
        offset = 0
    total = len(items)

    page = items[offset : offset + limit]
    has_next = (offset + limit) < total
    next_cursor = encode_cursor(offset + limit) if has_next else None
    has_previous = offset > 0

    return PaginatedResponse(
        data=page,
        pagination=PaginationMeta(
            total=total,
            limit=limit,
            has_next=has_next,
            next_cursor=next_cursor,
            has_previous=has_previous,
        ),
    )