eleanor.backend.api

Eleanor Framework API layer.

The API layer is responsible for handling incoming HTTP requests and invoking backend service(s). Versioned implementations are implemented as submodules to eleanor.backend.api and must all start with v, ex v1, v2``, etc.

Specific API version implementations must handle backwards-incompatible changes and Data Transfer Object (DTO) mapping. Moreover, the API layer is responsible for mapping incoming and outgoing DTOs to the corresponding service layer objects.

Note

A major future enhancement is planned to enable long-running job execution services, see: #180. This feature will allow for a request to be submitted via the API and create a unit of work object to be processed asynchronously. The API will return a task ID that can be used to query the status of the task. Callback notifications may also be supported.

Attributes

DEFAULT_HTTP_ERROR_RESPONSES module-attribute

DEFAULT_HTTP_ERROR_RESPONSES = {}

FASTAPI_PATH_SEP_CHAR module-attribute

FASTAPI_PATH_SEP_CHAR = '-'

R module-attribute

R = TypeVar('R')

SERVICE_ERROR_MAP module-attribute

SERVICE_ERROR_MAP = {
    GENERAL_SERVICE_ERROR: HTTP_500_INTERNAL_SERVER_ERROR,
    INVALID_INPUT: HTTP_400_BAD_REQUEST,
    RESOURCE_INVALID_STATE: HTTP_400_BAD_REQUEST,
    RESOURCE_NOT_FOUND: HTTP_404_NOT_FOUND,
    RESOURCE_ALREADY_EXISTS: HTTP_409_CONFLICT,
    RESOURCE_DELETED: HTTP_410_GONE,
    RESOURCE_DISABLED: HTTP_403_FORBIDDEN,
    RESOURCE_INVALID_OWNERSHIP: HTTP_403_FORBIDDEN,
}

cur_status_code module-attribute

cur_status_code = getattr(status, cur_status)

Classes

BaseRequest

Bases: BaseDataModel

Base request object for all API endpoints

BaseResponse

Bases: BaseDataModel

Base response object for all API endpoints

Attributes

model_config class-attribute instance-attribute

model_config = ConfigDict(
    populate_by_name=True,
    use_enum_values=True,
    extra="forbid",
    strict=False,
)

status class-attribute instance-attribute

status: ResponseStatus = Field(
    default_factory=lambda: ResponseStatus(
        message=name,
        code=SUCCESS.value,
        trace=gen_trace_id(),
    ),
    description="API call response status information",
)

ResponseStatus

Bases: BaseDataModel

API call response status object

Attributes

code class-attribute instance-attribute

code: int = Field(
    title="Numeric status code",
    description="Numeric code representing the response status. A value of 0 indicates success, a non-zero value indicates a failure.",
)

message class-attribute instance-attribute

message: str = Field(
    title="Message",
    description="Human-readable message that describes the result of the API call.",
)

trace class-attribute instance-attribute

trace: str = Field(
    title="Trace identifier",
    description="Unique identifier for tracing the operation on the server.",
)

Functions

format_path

format_path(*args) -> str

Formats a list of FastAPI path components to a valid path string.

Parameters:

• components (List[str] | None) –

  A list of path components to format, if None then a single “/” is returned. Defaults to None.

Returns:

• str ( str ) –

  The formatted tag string.

Raises:

• ValueError –

  If an empty path component is detected.

gen_trace_id

gen_trace_id() -> str

Generate a trace ID with a prefix.

Returns:

• str ( str ) –

  The generated trace ID.

run_async async

run_async(
    fn: Callable[..., R],
    executor: Executor,
    *args: Any,
    **kwargs: Any
) -> R

Asynchronously runs a function with the given arguments and keyword arguments using the specified executor.

Parameters:

• fn (Callable[..., R]) –

  The function to be executed asynchronously.

• executor (Executor) –

  The executor to be used for running the function.

• *args (Any, default: () ) –

  Positional arguments to be passed to the function.

• **kwargs (Any, default: {} ) –

  Keyword arguments to be passed to the function.

Returns:

• R –

  The result of the function execution.

run_background

run_background(
    fn: Callable[..., R],
    executor: Executor,
    task_id: str,
    *args: Any,
    **kwargs: Any
) -> None

Run a function in the background using the specified executor.

Parameters:

• fn (Callable[..., R]) –

  The function to be executed in the background.

• executor (Executor) –

  The executor to use for running the function.

• task_id (str) –

  The ID of the background task.

• *args (Any, default: () ) –

  Positional arguments to be passed to the function.

• **kwargs (Any, default: {} ) –

  Keyword arguments to be passed to the function.

Returns:

• None –

  None