ds_provider_xledger_py_lib.utils.introspection¶

File: introspection.py Region: ds_provider_xledger_py_lib/utils

Description¶

Metadata loading helpers for dataset operation contracts.

Classes¶

`MetaField`	Single field definition from an operation metadata asset.
`IncrementalMetaData`	Incremental read section from packaged read metadata.
`PaginationMetaData`	Static pagination read definition from metadata.
`MetaData`	Operation-level metadata plus packaged GraphQL query template.
`EntryPointMetaData`	Container for all loaded operations for one entrypoint.
`IntrospectionService`	Load and cache entrypoint metadata for dataset lifecycle reuse.

Functions¶

`_load_entrypoint_metadata`(→ EntryPointMetaData)	Load all available operations for an entrypoint from packaged assets.
`_load_operation_metadata`(→ MetaData \| None)	Load metadata and query template for one operation.
`_read_operation_assets`(→ tuple[dict[str, Any], str] \| None)	Read raw metadata and query files from package assets.
`_read_operation_asset_texts`(→ tuple[str, str] \| None)	Read raw metadata/query text files with process-local memoization.

Module Contents¶

class ds_provider_xledger_py_lib.utils.introspection.MetaField[source]¶

Bases: ds_common_serde_py_lib.serializable.Serializable

Single field definition from an operation metadata asset.

name: str¶

type: str¶

description: str¶

required: bool = False¶

default: bool = False¶

class ds_provider_xledger_py_lib.utils.introspection.IncrementalMetaData[source]¶

Bases: ds_common_serde_py_lib.serializable.Serializable

Incremental read section from packaged read metadata.

field names the node attribute used for watermarks; filter_field names the GraphQL filter argument (e.g. modifiedAt_gte). They are distinct and both required when incremental is present in JSON.

kind: str¶

field: str¶

filter_field: str¶

class ds_provider_xledger_py_lib.utils.introspection.PaginationMetaData[source]¶

Bases: ds_common_serde_py_lib.serializable.Serializable

Static pagination read definition from metadata.

kind: str¶

first: int | None = None¶

class ds_provider_xledger_py_lib.utils.introspection.MetaData[source]¶

Bases: ds_common_serde_py_lib.serializable.Serializable

Operation-level metadata plus packaged GraphQL query template.

name: str¶

type: str¶

description: str¶

fields: list[MetaField]¶

query: str¶

pattern: str | None = None¶

incremental: IncrementalMetaData | None = None¶

pagination: PaginationMetaData | None = None¶

class ds_provider_xledger_py_lib.utils.introspection.EntryPointMetaData[source]¶

Container for all loaded operations for one entrypoint.

entrypoint: str¶

operations: dict[ds_provider_xledger_py_lib.enums.OperationType, MetaData]¶

get(*, operation: ds_provider_xledger_py_lib.enums.OperationType) → MetaData[source]¶

Return metadata for a specific operation.

Parameters:: operation – Operation to resolve.
Returns:: Metadata for the requested operation.
Raises:: NotSupportedError – If metadata for the operation is unavailable.

property read: MetaData | None¶: Metadata for read operation, when available.

property create: MetaData | None¶: Metadata for create operation, when available.

property update: MetaData | None¶: Metadata for update operation, when available.

property delete: MetaData | None¶: Metadata for delete operation, when available.

class ds_provider_xledger_py_lib.utils.introspection.IntrospectionService[source]¶

Load and cache entrypoint metadata for dataset lifecycle reuse.

entrypoint: str¶

_metadata: EntryPointMetaData | None = None¶

property metadata: EntryPointMetaData¶

Loaded entrypoint metadata.

Returns:: Cached metadata snapshot for the configured entrypoint.

load() → EntryPointMetaData[source]¶

Load metadata snapshot if needed and return cached value.

Returns:: Cached metadata snapshot for the configured entrypoint.

load_metadata(*, operation: ds_provider_xledger_py_lib.enums.OperationType) → MetaData[source]¶

Return metadata for a single operation.

Parameters:: operation – Operation to resolve.
Returns:: Metadata for the requested operation.

ds_provider_xledger_py_lib.utils.introspection._load_entrypoint_metadata(entrypoint: str) → EntryPointMetaData[source]¶

Load all available operations for an entrypoint from packaged assets.

Parameters:

entrypoint – Dataset entrypoint (supports nested paths).

Returns:

EntryPointMetaData containing all discovered operations.

Raises:

ValidationError – If entrypoint is missing.
NotSupportedError – If the entrypoint is unsupported.

ds_provider_xledger_py_lib.utils.introspection._load_operation_metadata(*, entrypoint: str, operation: ds_provider_xledger_py_lib.enums.OperationType) → MetaData | None[source]¶

Load metadata and query template for one operation.

Parameters:

entrypoint – Dataset entrypoint.
operation – Operation to load.

Returns:

Parsed MetaData when assets exist, otherwise None.

Raises:

ValidationError – If payload is malformed.

ds_provider_xledger_py_lib.utils.introspection._read_operation_assets(*, entrypoint: str, operation: ds_provider_xledger_py_lib.enums.OperationType) → tuple[dict[str, Any], str] | None[source]¶

Read raw metadata and query files from package assets.

Parameters:

entrypoint – Dataset entrypoint.
operation – Operation to load.

Returns:

Tuple of (metadata_json_dict, query_graphql_text) when found, otherwise None.

ds_provider_xledger_py_lib.utils.introspection._read_operation_asset_texts(*, entrypoint: str, operation: ds_provider_xledger_py_lib.enums.OperationType) → tuple[str, str] | None[source]¶

Read raw metadata/query text files with process-local memoization.

Parameters:

entrypoint – Dataset entrypoint.
operation – Operation to load.

Returns:

Tuple of (metadata_json_text, query_graphql_text) when found, otherwise None.