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.

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.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
supportPagination: bool = False
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.