Core Module

Core module containing the SurfaceDriver class.

This module contains the SurfaceDriver class, an abstract base class responsible for being the base class developers subclass to create a dialect driver while also providing core functionality of the package.

SurfaceDriver:

Abstract base class for managing and running driver toolchains.

class plugorm.core.SurfaceDriver(conn_driver: ConnectionDriver, il_driver: ILDriver | None = None, simplifier: Simplifier | None = None, logger: Logger | None = None, fallback: bool = False)

Bases: Driver

Abstract base class for a surface driver managing multiple components.

A SurfaceDriver orchestrates the execution of connected drivers, including a mandatory connection driver, an optional internal language driver, and an optional simplifier. It supports building and running both synchronous and asynchronous toolchains, with optional fallback mechanisms for handling async/sync mismatches.

conn_driver

The primary connection driver.

Type:

ConnectionDriver

il_driver

Optional internal language driver.

Type:

ILDriver | None

simplifier

Optional simplifier for post-processing.

Type:

Simplifier | None

logger

Logger instance for debugging and info messages.

Type:

logging.Logger

sync_toolchain

Ordered list of synchronous steps.

Type:

list[SyncStep]

async_toolchain

Ordered list of asynchronous steps.

Type:

list[AsyncStep]

fallback

Enables fallback between sync and async execution when a step does not natively support the requested mode.

Type:

bool

async aclose() → None

Closes the asynchronous connection.

Calls aclose() on the connection driver if it supports asynchronous operation. If not, falls back to threaded sync execution if enabled.

Raises:

NotAsyncError – If the connection driver is not async-capable and fallback is disabled.

async aconnect() → None

Establishes an asynchronous connection.

Calls aconnect() on the connection driver if it supports asynchronous operation. If not, falls back to threaded sync execution if enabled.

Raises:

NotAsyncError – If the connection driver is not async-capable and fallback is disabled.

static adialect(func: Callable[[...], Awaitable[Any]]) → Callable[[...], Awaitable[Any]]

Decorator for asynchronous dialect parsing.

Wraps a coroutine function that produces a dialect representation, ensuring that any exceptions raised during parsing are caught and re-raised as DialectParsingError. After parsing, the result is executed through the asynchronous toolchain.

Parameters:

func (Callable[..., Awaitable[Any]]) – An async function that parses and returns a dialect representation.

Returns:

A wrapped coroutine function that validates parsing and runs the result

through the async toolchain.

Return type:

Callable[…, Awaitable[Any]]

Raises:

DialectParsingError – If the wrapped function fails to parse.

async arun_toolchain(value: Any) → Any

Executes the asynchronous toolchain.

Runs the configured async toolchain steps sequentially, awaiting the result of each step before passing it to the next.

Parameters:

value (Any) – The initial input passed to the first step.

Returns:

The final result after all steps have been executed.

Return type:

Any

async_toolchain: list[Callable[[Any], Awaitable[Any]]]

build_async_toolchain()

Builds the asynchronous toolchain.

Invokes the internal _build_toolchain with allow_async=True to construct the async-capable sequence of steps and stores it in self.async_toolchain.

build_sync_toolchain()

Builds the synchronous toolchain.

Invokes the internal _build_toolchain with allow_async=False to construct the sync-only sequence of steps and stores it in self.sync_toolchain.

build_toolchains()

Builds both sync and async toolchains.

Calls the internal builder methods to populate the synchronous and asynchronous toolchains with their respective steps.

close() → None

Closes the synchronous connection.

Calls close() on the connection driver if it supports synchronous operation. Raises an error if the driver is async-only.

Raises:

NotSyncError – If the connection driver is async-only.

conn_driver: ConnectionDriver

connect() → None

Establishes a synchronous connection.

Calls connect() on the connection driver if it supports synchronous operation. Raises an error if the driver is async-only.

Raises:

NotSyncError – If the connection driver is async-only.

static dialect(func: Callable[[...], Any]) → Callable[[...], Any]

Decorator for synchronous dialect parsing.

Wraps a function that produces a dialect representation, ensuring that any exceptions raised during parsing are caught and re-raised as DialectParsingError. After parsing, the result is executed through the synchronous toolchain.

Parameters:

func (Callable[..., Any]) – A function that parses and returns a dialect representation.

Returns:

A wrapped function that validates parsing and runs the result through the sync

toolchain.

Return type:

Callable[…, Any]

Raises:

DialectParsingError – If the wrapped function fails to parse.

fallback: bool

il_driver: ILDriver | None

logger: Logger

reset_toolchains()

Resets both sync and async toolchains.

Clears the internal lists that store the synchronous and asynchronous toolchains. This effectively removes any previously built steps.

run_toolchain(value: Any) → Any

Executes the synchronous toolchain.

Runs the configured sync toolchain steps sequentially, passing the output of each step as the input to the next.

Parameters:

value (Any) – The initial input passed to the first step.

Returns:

The final result after all steps have been executed.

Return type:

Any

simplifier: Simplifier | None

sync_toolchain: list[Callable[[Any], Any]]

validate_link(source_name: str, target_name: str, source_output: set[str], target_input: set[str]) → None

Validates compatibility between two driver components.

This method ensures that the output types of a source driver overlap with the input types of a target driver. If no overlap exists, the validation fails and a LinkValidationError is raised.

Parameters:

• source_name (str) – Human-readable name of the source driver.

• target_name (str) – Human-readable name of the target driver.

• source_output (set[str]) – The possible output types of the source driver.

• target_input (set[str]) – The expected input types of the target driver.

Raises:

LinkValidationError – If the source and target driver types are incompatible (no overlapping elements).