Service Classes

This package provides the base service abstraction and the procedure mechanism that lets service methods be exposed either as direct Python callables or as HTTP endpoints. The module-level exports make the most commonly used building blocks available for users of the data layer.

Basic Facilities

Service abstract base class for data services; provides transport helpers and common utilities.
procedure() decorator used to mark service methods as procedures.
Http convenient alias for ProcedureHttpConfig used when decorating service methods for HTTP exposure.

Base Classes

GetByIdService specialized Service interface for types that can be retrieved by integer id.
ServiceController Litestar controller that dispatches to service procedures.

Procedure Internals

Procedure: internal representation of a service procedure; manages adapters for direct and HTTP invocation (see ixmp4.data.services.procedure).
ProcedureDescriptor: descriptor returned by the procedure` decorator; binds a procedure to instances and to transports.
ProcedureRouteHandler / ProcedureHttpConfig: HTTP adapter and configuration used to expose procedures as HTTP routes.
ProcedureClient: HTTP client adapter used when a service is backed by an ixmp4.transport.HttpxTransport.
ProcedureAuthCheck / ProcedurePagination: optional wrappers to add authorization checks and pagination support to procedures.

Transports re-exported here for convenience: * DirectTransport: direct (same-process) transport implementation. * HttpxTransport: HTTP transport using httpx. * Transport: abstract transport base class.

class ixmp4.data.services.DirectTransport(session: Session)

Bases: Transport

session: Session

classmethod check_dsn(dsn: str) → str

classmethod from_dsn(dsn: str, *args: Any, **kwargs: Any) → DirectTransport

classmethod create_postgresql_engine(dsn: str) → Engine

classmethod create_sqlite_engine(dsn: str) → Engine

get_database_url() → URL | None

get_engine_info() → str

close() → None

check_versioning_compatiblity() → None

class ixmp4.data.services.HttpxTransport(client: Client | TestClient[Litestar], settings: ClientSettings, check_root: bool = True)

Bases: Transport, ServiceClient

exception_registry = <toolkit.exceptions.registry.ServiceExceptionRegistry object>

direct: DirectTransport | None = None

backoff_maximum = 16.0

backoff_factor = 0.5

backoff_exp_base = 2.0

settings: ClientSettings

executor: ThreadPoolExecutor

http_client: Client | TestClient[Litestar]

check_root() → None: Requests root api endpoint and logs messages.

classmethod from_url(url: str, settings: ClientSettings | None = None, auth: Auth | None = None) → HttpxTransport

classmethod from_asgi(asgi: Litestar, settings: ClientSettings, direct: DirectTransport | None = None, raise_server_exceptions: bool = True) → HttpxTransport

request(method: str, path: str, **kwargs: Any) → Response: Issue a request and retry when the server returns HTTP 429.

get_retry_delay_seconds(response: Response, attempt: int) → float: Return retry delay based on Retry-After or exponential backoff.

class ixmp4.data.services.Transport

Bases: ABC

check_versioning_compatiblity() → None

class ixmp4.data.services.GetByIdService(transport: Transport)

Bases: Service

Service interface for types that can be retrieved by numeric id.

Implementations must provide get_by_id() which returns a ixmp4.data.base.dto.BaseModel instance for the given id.

abstractmethod get_by_id(id: int) → BaseModel

class ixmp4.data.services.Service(transport: Transport)

Bases: ABC

Main data layer interface for a data type.

from ixmp4.data.services import (
    GetByIdService,
    Http,
    procedure,
)
from .exceptions import ExampleError

class ExampleService(Service):
    @procedure(Http(path="/", methods=("POST",)))
    def do_something(self):
        raise ExampleError("Can't do something, sorry.")

To mark service methods as interface procedures that can be called directly or via the http api, use the procedure() decorator.

Services can then be instantiated with a ixmp4.transport.Transport object:

from ixmp4.transport import DirectTransport, HttpxTransport
from .example import ExampleService

direct = DirectTranport.from_dsn("sqlite://...")
direct_svc = ExampleService(direct)

http = HttpxTransport.from_url/from_asgi(...)
http_svc = ExampleService(http)

direct_svc.do_something()
#> ExampleError

http_svc.do_something()
#> ExampleError

router_tags: ClassVar[Sequence[str]] = []

router_prefix: ClassVar[str]

http_controller: ClassVar[type[Controller] | None] = None

default_filter: Mapping[str, Any] = {}

transport: Transport

get_auth_kwargs(transport: DirectTransport) → AuthKwargs

get_dialect() → Dialect

get_datetime() → datetime

get_username() → str

get_creation_info() → dict[str, str | datetime]

get_update_info() → dict[str, str | datetime]

validate_df_or_raise(df: DataFrame, model: type[DataFrameModel]) → DataFrame

apply_filter_defaults(values: Mapping[str, Any]) → dict[str, Any]

deep_update_dict(d: dict[str, Any], o: Mapping[str, Any]) → dict[str, Any]

classmethod get_router(settings: ServerSettings) → Router

ixmp4.data.services.Http: alias of ProcedureHttpConfig

ixmp4.data.services.procedure(http_config: ProcedureHttpConfig) → Callable[[Callable[[Concatenate[ServiceT, Params]], ReturnT]], ProcedureDescriptor[ServiceT, Params, ReturnT]]

Makes a service method callable directly or via http. Constructs an internal Procedure instance and

Returns:: A special descriptor class that provides procedure functionality on a service class.
Return type:: ProcedureDescriptor

ixmp4.db.services.base module

class ixmp4.data.services.base.AuthKwargs

Bases: TypedDict

auth_ctx: AuthorizationContext | None

platform: PlatformProtocol | None

class ixmp4.data.services.base.Service(transport: Transport)

Bases: ABC

Main data layer interface for a data type.

from ixmp4.data.services import (
    GetByIdService,
    Http,
    procedure,
)
from .exceptions import ExampleError

class ExampleService(Service):
    @procedure(Http(path="/", methods=("POST",)))
    def do_something(self):
        raise ExampleError("Can't do something, sorry.")

To mark service methods as interface procedures that can be called directly or via the http api, use the procedure() decorator.

Services can then be instantiated with a ixmp4.transport.Transport object:

from ixmp4.transport import DirectTransport, HttpxTransport
from .example import ExampleService

direct = DirectTranport.from_dsn("sqlite://...")
direct_svc = ExampleService(direct)

http = HttpxTransport.from_url/from_asgi(...)
http_svc = ExampleService(http)

direct_svc.do_something()
#> ExampleError

http_svc.do_something()
#> ExampleError

router_tags: ClassVar[Sequence[str]] = []

router_prefix: ClassVar[str]

http_controller: ClassVar[type[Controller] | None] = None

default_filter: Mapping[str, Any] = {}

transport: Transport

get_auth_kwargs(transport: DirectTransport) → AuthKwargs

get_dialect() → Dialect

get_datetime() → datetime

get_username() → str

get_creation_info() → dict[str, str | datetime]

get_update_info() → dict[str, str | datetime]

validate_df_or_raise(df: DataFrame, model: type[DataFrameModel]) → DataFrame

apply_filter_defaults(values: Mapping[str, Any]) → dict[str, Any]

deep_update_dict(d: dict[str, Any], o: Mapping[str, Any]) → dict[str, Any]

classmethod get_router(settings: ServerSettings) → Router

class ixmp4.data.services.base.GetByIdService(transport: Transport)

Bases: Service

Service interface for types that can be retrieved by numeric id.

Implementations must provide get_by_id() which returns a ixmp4.data.base.dto.BaseModel instance for the given id.

abstractmethod get_by_id(id: int) → BaseModel

ixmp4.db.services.controller module

async ixmp4.data.services.controller.get_pagination(offset: int = 0, limit: int = 5000) → Pagination

class ixmp4.data.services.controller.ServiceController(owner: Router)

Bases: Controller, Generic[ServiceT]

Base Litestar controller that dispatches to service procedures.

The controller is constructed with a service instance available via dependency injection and provides helpers to look up the correct procedure handler and invoke it with request data.

dependencies: A string keyed dictionary of dependency Provider instances.

get_handler(service: ServiceT, name: str) → ProcedureRouteHandler[ServiceT, Any, Any]

async call_procedure(service: ServiceT, name: str, request: Request[Any, Any, Any]) → Any

after_request

A sync or async function executed before a Request is passed to any route handler.

If this function returns a value, the request will not reach the route handler, and instead this value will be used.

after_response

A sync or async function called after the response has been awaited.

It receives the Request instance and should not return any values.

before_request

A sync or async function called immediately before calling the route handler.

It receives the Request instance and any non-None return value is used for the response, bypassing the route handler.

cache_control

A CacheControlHeader header to add to route handlers of this controller.

Can be overridden by route handlers.

dto: AbstractDTO to use for (de)serializing and validation of request data.

etag

An etag header of type ETag to add to route handlers of this controller.

Can be overridden by route handlers.

exception_handlers: A map of handler functions to status codes and/or exception types.

guards: A sequence of Guard callables.

include_in_schema: A boolean flag dictating whether the route handler should be documented in the OpenAPI schema

middleware: A sequence of Middleware.

opt: A string key mapping of arbitrary values that can be accessed in Guards or wherever you have access to Request or ASGI Scope.

owner

The Router or Litestar app that owns the controller.

This value is set internally by Litestar and it should not be set when subclassing the controller.

parameters: A mapping of Parameter definitions available to all application paths.

path

A path fragment for the controller.

All route handlers under the controller will have the fragment appended to them. If not set it defaults to /.

request_class: A custom subclass of Request to be used as the default request for all route handlers under the controller.

request_max_body_size: Maximum allowed size of the request body in bytes. If this size is exceeded, a ‘413 - Request Entity Too Large’ error response is returned.

response_class: A custom subclass of Response to be used as the default response for all route handlers under the controller.

response_cookies: A list of Cookie instances.

response_headers: A string keyed dictionary mapping ResponseHeader instances.

return_dto: AbstractDTO to use for serializing outbound response data.

security: A sequence of dictionaries that to the schema of all route handlers under the controller.

signature_namespace: A mapping of names to types for use in forward reference resolution during signature modelling.

signature_types

A sequence of types for use in forward reference resolution during signature modelling.

These types will be added to the signature namespace using their __name__ attribute.

tags: A sequence of string tags that will be appended to the schema of all route handlers under the controller.

type_decoders: A sequence of tuples, each composed of a predicate testing for type identity and a msgspec hook for deserialization.

type_encoders: A mapping of types to callables that transform them into types supported for serialization.

websocket_class: A custom subclass of WebSocket to be used as the default websocket for all route handlers under the controller.

ixmp4.db.services.procedure modules

class ixmp4.data.services.procedure.Procedure(func: Callable[[Concatenate[ServiceT, Params]], ReturnT], http_config: ProcedureHttpConfig)

Bases: Generic[ServiceT, Params, ReturnT]

Represents a service procedure and its HTTP/transport adapters.

A Procedure wraps a service method, validates its signature, and provides adapters for direct invocation, http client calls, and registration of HTTP route handlers. It also manages authorization checks and pagination metadata attached to the procedure.

func: Callable[[Concatenate[ServiceT, Params]], ReturnT]

signature: Signature

auth_check: ProcedureAuthCheck[ServiceT, Params]

pagination: ProcedurePagination[ServiceT, Params, ReturnT]

http_config: ProcedureHttpConfig

handlers: dict[type[Service], ProcedureRouteHandler[ServiceT, Params, ReturnT]]

direct_payload_model: type[BaseModel]

build_direct_payload_model() → type[BaseModel]

validate_direct_call_args(args: tuple[Any, ...], kwargs: dict[str, Any], varargs_key: str = '__varargs__') → tuple[tuple[Any, ...], dict[str, Any]]

validate_signature(func: Callable[[...], Any]) → Signature

validate_corresponding_parameter(index: int, name: str, param: Parameter, func: Callable[[...], Any]) → bool

set_route_handler(handler: HTTPRouteHandler) → None

get_authorized_callable(service: ServiceT, func: Callable[[Params], Any]) → Callable[[Params], Any]

get_direct_callable(service: ServiceT) → Callable[[Params], ReturnT]

get_httpx_callable(service: ServiceT) → Callable[[Params], ReturnT]

get_descriptor() → ProcedureDescriptor[ServiceT, Params, ReturnT]

register_service(svc_cls: type[ServiceT]) → ProcedureRouteHandler[ServiceT, Params, ReturnT]

ixmp4.data.services.procedure.procedure(http_config: ProcedureHttpConfig) → Callable[[Callable[[Concatenate[ServiceT, Params]], ReturnT]], ProcedureDescriptor[ServiceT, Params, ReturnT]]

Makes a service method callable directly or via http. Constructs an internal Procedure instance and

Returns:: A special descriptor class that provides procedure functionality on a service class.
Return type:: ProcedureDescriptor

class ixmp4.data.services.procedure.descriptor.ProcedureDescriptor(procedure: Procedure[ServiceT, Params, ReturnT])

Bases: Generic[ServiceT, Params, ReturnT]

property auth_check: ProcedureAuthCheck[ServiceT, Params]

property paginated: ProcedurePagination[ServiceT, Params, ReturnT]

procedure: Procedure[ServiceT, Params, ReturnT]

Descriptor exposing a Procedure on a Service class.

When accessed on a service instance the descriptor returns a callable appropriate for the transport (direct call or HTTP client). When accessed on the class it exposes descriptor attributes (used by router registration).

class ixmp4.data.services.procedure.endpoint.ProcedureHttpConfig(methods: HttpMethod | Literal['GET', 'POST', 'DELETE', 'PATCH', 'PUT', 'HEAD', 'TRACE', 'OPTIONS'] | Sequence[HttpMethod | Literal['GET', 'POST', 'DELETE', 'PATCH', 'PUT', 'HEAD', 'TRACE', 'OPTIONS']], path: str | None = None, status_code: int = 200)

Configuration for exposing a procedure via HTTP.

methods: HTTP method(s) the endpoint should accept.
path: optional explicit path; if omitted a path is derived
from the procedure name.
status_code: response status code for successful responses.

methods: HttpMethod | Literal['GET', 'POST', 'DELETE', 'PATCH', 'PUT', 'HEAD', 'TRACE', 'OPTIONS'] | Sequence[HttpMethod | Literal['GET', 'POST', 'DELETE', 'PATCH', 'PUT', 'HEAD', 'TRACE', 'OPTIONS']]

path: str | None = None

status_code: int = 200

class ixmp4.data.services.procedure.endpoint.ProcedureRouteHandler(procedure: Procedure[ServiceT, Params, ReturnT], service_class: type[ServiceT], config: ProcedureHttpConfig)

HTTP route handler that adapts a procedure to a Litestar route.

Constructs request/response models from the procedure signature, binds service instances, handles validation errors, and performs serialization/deserialization for the procedure’s return type.

procedure: Procedure[ServiceT, Params, ReturnT]

config: ProcedureHttpConfig

service_class: type[ServiceT]

operation_id

proto_route: HTTPRoute

get_path_fields(path: str) → list[str]

build_payload_model(path_fields: list[str]) → type[BaseModel]

build_path_model(path_fields: list[str]) → type[BaseModel]

get_model_name(suffix: str) → str

get_openapi_operation_class() → type[Operation]

get_openapi_responses() → dict[str, OpenAPIResponse | Reference]

after_request

after_response

background

before_request

cache

cache_control

cache_key_builder

content_encoding

content_media_type

deprecated

description

etag

has_sync_callable

http_methods

include_in_schema

media_type

operation_class

raises

request_class

request_max_body_size

response_class

response_cookies

response_description

response_headers

responses

security

status_code

summary

sync_to_thread

tags

template_name

handle_request(request: Request[Any, Any, Any], service: Any, query: dict[str, Any], body: bytes) → Response[Any]

build_call_args(path: dict[str, Any], query: dict[str, Any], body: bytes, varargs_key: str = '__varargs__') → tuple[tuple[Any, ...], dict[str, Any]]

get_pagination_params(query_params: dict[str, Any]) → Pagination

bind_endpoint_func(service: ServiceT, query_params: dict[str, Any]) → Callable[[Params], Any]

ixmp4.data.services.procedure.endpoint.generate_arguments_model(signature: Signature, model_name: str, __module__: str, parameter_callback: Callable[[int, str, Parameter], Literal['skip'] | None], varargs_key: str = '__varargs__', extra: Literal['forbid', 'ignore', 'allow'] = 'forbid') → type[BaseModel]

class ixmp4.data.services.procedure.client.ProcedureClient(service: ServiceT, handler: ProcedureRouteHandler[ServiceT, Params, ReturnT])

Bases: Generic[ServiceT, Params, ReturnT]

HTTP client adapter for a ProcedureRouteHandler.

When a procedure is accessed on a service backed by an ixmp4.transport.HttpxTransport, the descriptor returns an instance of ProcedureClient which performs HTTP requests to the service endpoint, validates arguments, and handles paginated responses by dispatching concurrent requests when needed.

handler: ProcedureRouteHandler[ServiceT, Params, ReturnT]

method: str

transport: HttpxTransport

reverse_path(path_parameters: dict[str, Any]) → str

pos_args_to_named(args: tuple[Any]) → dict[str, Any]

classify_arguments(*args: ~Params, **kwargs: ~Params) → tuple[dict[str, Any], dict[str, Any]]

handle_paginated_response(response: Response, path: str, params: dict[str, Any] | None, json: dict[str, Any] | None) → ReturnT

dispatch_pagination_requests(path: str, total: int, start: int, limit: int, params: dict[str, Any] | None = None, json: dict[str, Any] | None = None) → list[ReturnT]

merge_results(results: list[ReturnT]) → ReturnT

merge_dataframes(results: list[DataFrame]) → DataFrame

merge_lists(results: list[list[Any]]) → list[Any]

class ixmp4.data.services.procedure.auth.InternalProcedureAuthCheckFunc(*args, **kwargs): Bases: Protocol[ContraServiceT]

class ixmp4.data.services.procedure.auth.ProcedureAuthCheckFuncWithParams(*args, **kwargs): Bases: Protocol[ContraServiceT, Params]

class ixmp4.data.services.procedure.auth.ProcedureAuthCheckFuncNoParams(*args, **kwargs): Bases: Protocol[ContraServiceT]

class ixmp4.data.services.procedure.auth.ProcedureAuthCheck(procedure: Procedure[ServiceT, Params, Any])

Bases: Generic[ServiceT, Params]

input_func: ProcedureAuthCheckFuncNoParams[ServiceT] | ProcedureAuthCheckFuncWithParams[ServiceT, Params]

check_func: InternalProcedureAuthCheckFunc[ServiceT]

signature: Signature

procedure: Procedure[ServiceT, Params, Any]

has_check: bool

decorator(func: ProcedureAuthCheckFuncNoParams[ServiceT] | ProcedureAuthCheckFuncWithParams[ServiceT, Params]) → ProcedureAuthCheckFuncNoParams[ServiceT] | ProcedureAuthCheckFuncWithParams[ServiceT, Params]

validate_signature(func: Callable[[...], Any]) → Signature

validate_parameter(index: int, name: str, param: Parameter, func: Callable[[...], Any]) → bool

prepend_auth_check(service: ServiceT, auth_ctx: AuthorizationContext, platform: PlatformProtocol, procedure_func: Callable[[Params], Any]) → Callable[[Params], Any]

class ixmp4.data.services.procedure.pagination.ProcedurePaginatedFunc(*args, **kwargs): Bases: Protocol[ContraServiceT, Params, CoReturnT]

class ixmp4.data.services.procedure.pagination.BoundProcedurePaginatedFunc(*args, **kwargs): Bases: Protocol[Params, CoReturnT]

class ixmp4.data.services.procedure.pagination.ProcedurePagination(procedure: Procedure[ServiceT, Params, Any])

Bases: Generic[ServiceT, Params, ReturnT]

Descriptor for marking a procedure as paginated.

Use as a decorator on a service method to declare that the procedure supports pagination. It validates the method signature and stores the paginated call target for later binding by the procedure machinery.

paginated_func: ProcedurePaginatedFunc[ServiceT, Params, PaginatedResult[TypeVar]]

procedure: Procedure[ServiceT, Params, Any]

has_pagination: bool

decorator(func: ProcedurePaginatedFunc[ServiceT, Params, PaginatedResult[TypeVar]]) → BoundProcedurePaginatedFunc[Params, PaginatedResult[TypeVar]]

validate_signature(func: Callable[[...], Any]) → Signature

validate_parameter(index: int, name: str, param: Parameter, func: Callable[[...], Any]) → bool