gavicore.util API Reference

gavicore.util.dynimp

import_value

import_value(
    ref: str, *, type: type[T], name: str, example: Optional[str] = None
) -> T

Dynamically import a value given by its reference ref.

Parameters:

Name	Type	Description	Default
ref	str	A value reference of the form ":".	required
type	type[T]	Expected type of the reference	required
name	str	Name of the reference	required
example	Optional[str]	An example reference (for error messages)	None

Returns:

Type	Description
T	The imported value.

Raises:

Type	Description
ValueError	if the ref syntax is invalid, or the module cannot be imported, or attribute cannot be found.
TypeError	if the loaded value is not an instance of the expected type.

gavicore.util.ensure

ensure_type

ensure_type(
    name: str, value: Any, expected_type: type | tuple[type, ...]
) -> None

Raise a TypeError if value is not of type expected_type.

ensure_callable

ensure_callable(name: str, value: Any) -> None

Raise a TypeError if value is not callable.

ensure_condition

ensure_condition(
    condition: bool,
    message: str | Callable[[], str],
    exception_type: type[Exception] = ValueError,
) -> None

Raise a ValueError if condition is False.

gavicore.util.json

JsonCodec

Bases: ABC, Generic[T]

Convert component values to/from valid JSON values.

to_json abstractmethod

to_json(value: T | None) -> JsonValue

Return a valid JSON value from given Python value.

from_json abstractmethod

from_json(json_value: JsonValue) -> T | None

Return a Python value from given valid JSON value.

gavicore.util.model

extend_model

extend_model(model_cls: T1, extension_cls: T2) -> T1

A utility function that allows for dynamically adding custom fields to a Pydantic v2 model class. A typical use case is adding vendor-specific extension field to API model classes.

If a field from extension_cls exists in model_cls, their field properties are merged.

If extension_cls has no fields, the function is a no-op.

Parameters:

Name	Type	Description	Default
model_cls	T1	The model class to be enhanced.	required
extension_cls	T2	The model class to that provides additional fields.	required

Returns:

Type	Description
T1	The model_cls instance.

gavicore.util.request

ExecutionRequest

Bases: ProcessRequest

Process execution request. Extends ProcessRequest

• to allow the process identifier being part of the request,
• to allow creating nested object values for input names with dots.

Parameters:

Name	Type	Description	Default
process_id		Process identifier	required
dotpath		Whether dots in input names should be used to create nested object values. Defaults to False.	required
inputs		Optional process inputs given as key-value mapping. Values may be of any JSON-serializable type accepted by the given process.	required
outputs		Optional process outputs given as key-value mapping. Values are of type Output supported by the given process.	required
subscriber		Optional subscriber of type Subscriber comprising callback URLs that are informed about process status changes while the processing takes place.	required

process_id instance-attribute

process_id: Annotated[str, Field(title='Process identifier', min_length=1)]

Required process identifier.

dotpath class-attribute instance-attribute

dotpath: Annotated[
    bool, Field(title="Whether to encode nested input values using dots ('.').")
] = False

Whether dots in input names should be used to create nested object values. Defaults to False.

inputs class-attribute instance-attribute

inputs: dict[str, Any] | None = None

Optional process inputs given as key-value mapping. Values may be of any JSON-serializable type accepted by the given process.

outputs class-attribute instance-attribute

outputs: dict[str, Output] | None = None

Optional process outputs given as key-value mapping. Values are of type Output supported by the given process.

subscriber class-attribute instance-attribute

subscriber: Subscriber | None = None

Optional subscriber of type Subscriber comprising callback URLs that are informed about process status changes while the processing takes place.

response class-attribute instance-attribute

response: ResponseType | None = ResponseType.raw

Optional response type given as key-value mapping. Maybe just ignored.

to_process_request

to_process_request() -> ProcessRequest

Convert this execution request into a process request as used by the execute-process operation.

create classmethod

create(
    process_id: str | None = None,
    dotpath: bool = False,
    request_path: str | None = None,
    inputs: list[str] | None = None,
    subscribers: list[str] | None = None,
) -> ExecutionRequest

A factory method to create an execution request.

The method is intended to support CLI implementations parsing user inputs and creating validated execution requests.

Parameters:

Name	Type	Description	Default
process_id	str | None	Process identifier	None
dotpath	bool	Whether dots in input names should be used to create nested object values. Defaults to False.	False
request_path	str | None	Local path to a file that contains an execution request in YAML or JSON format.	None
inputs	list[str] | None	Optional process inputs given as a list of "=" strings.	None
subscribers	list[str] | None	Optional subscribers given as a list of "=" strings.	None

Return

A validated execution request of type ExecutionRequest.

Raise

ValueError: if a validation error occurs.

from_process_description classmethod

from_process_description(
    process_description: ProcessDescription, dotpath: bool = False
) -> ExecutionRequest

Create an execution request from the given process description.

Parameters:

Name	Type	Description	Default
process_description	ProcessDescription	The process description	required
dotpath	bool	Whether to allow for dot-separated input names for nested object values	False

Returns:

Type	Description
ExecutionRequest	The execution requests populated with default values.

gavicore.util.runsync

run_sync

run_sync(
    async_fn: Callable[P, Coroutine[object, object, T]],
    *args: P.args,
    **kwargs: P.kwargs
) -> T

Run an async function from synchronous code and return its result.

Works in

• normal scripts (no running event loop) -> uses asyncio.run()
• Jupyter notebooks (running event loop) -> submits work to background loop thread

Important notes

• If called from a thread with NO running loop, we prefer asyncio.run() because it's the simplest and most deterministic.
• If called while a loop is running (common in Jupyter), we MUST NOT call:
  • asyncio.run() (raises RuntimeError)
  • loop.run_until_complete() on the same loop (already running) Instead, we use a dedicated background event loop thread and submit the coroutine there.
• Exceptions raised inside the async function are re-raised here unchanged.

gavicore.util.testing

BaseModelMixin

Add to test classes to get the assertBaseModelEqual method.

assertBaseModelEqual

assertBaseModelEqual(a: BaseModel, b: BaseModel)

Assert that two BaseModels are equal.

set_env_cm

set_env_cm(**new_env: str | None) -> Generator[dict[str, str | None]]

Run the code in the block with a new environment new_env.

set_env

set_env(**new_env: str | None) -> Callable[[], None]

Set the new environment in new_env and return a no-arg function to restore the old environment.

gavicore.util.text

TextConverter

Bases: Generic[T], ABC

Encode a (JSON) value as plain text and decode a (JSON) value from plain text.

parse abstractmethod

parse(text: str, **kwargs) -> T

Parse text into (JSON) value of type T.

format

format(value: T, **kwargs) -> str

Format a (JSON) value of type T into text.

ArrayTextConverter

ArrayTextConverter(item_converter: TextConverter[T])

Bases: Generic[T], TextConverter[list[T]], ABC

parse

parse(text: str, **kwargs) -> list[T]

Parse plain text into an array value with item type T.

format

format(value: list[T], **kwargs) -> str

Format an array value with item type T into plain text.

gavicore.util.undefined

Undefined

The type of an undefined value.

value instance-attribute

value: Undefined

The value of an undefined value.

is_undefined staticmethod

is_undefined(value: Any) -> bool

Test if the given value is undefined.

is_defined staticmethod

is_defined(value: Any) -> bool

Test if the given value is not undefined.