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 (for error messages)	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.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.

Attributes:

Name	Type	Description
process_id	Annotated[str, Field(title='Process identifier', min_length=1)]	Process identifier
dotpath	Annotated[bool, Field(title="Whether to encode nested input values using dots ('.').")]	Whether dots in input names should be used to create nested object values. Defaults to False.
inputs	dict[str, Any] | None	Optional process inputs given as key-value mapping. Values may be of any JSON-serializable type accepted by the given process.
outputs	dict[str, Output] | None	Optional process outputs given as key-value mapping. Values are of type Output supported by the given process.
subscriber	Subscriber | None	Optional subscriber of type Subscriber comprising callback URLs that are informed about process status changes while the processing takes place.

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.

use_temp_dir

use_temp_dir(
    temp_dir: str | os.PathLike[str] | Path | None = None,
) -> Generator[Path]

A context manager that creates a temporary directory and changes the current working directory to it. This isolates tests that affect the contents of the CWD to prevent them from interfering with each other.

Vendored from click.CliRunner.isolated_filesystem(), as it is not available in typer.CliRunner (as of typer 0.26.8).

Parameters:

Name	Type	Description	Default
temp_dir	str | os.PathLike[str] | Path | None	Create the temporary directory under this directory. If given, the created directory is not removed when exiting.	None