ds_provider_xledger_py_lib.utils

File: __init__.py Region: ds_provider_xledger_py_lib/utils

Description

Utility helpers for provider internals.

Submodules

• ds_provider_xledger_py_lib.utils.dataframe
• ds_provider_xledger_py_lib.utils.graphql
• ds_provider_xledger_py_lib.utils.introspection
• ds_provider_xledger_py_lib.utils.query_builder
• ds_provider_xledger_py_lib.utils.rules

Classes

EntryPointMetaData	Container for all loaded operations for one entrypoint.
IntrospectionService	Load and cache entrypoint metadata for dataset lifecycle reuse.
MetaData	Operation-level metadata plus packaged GraphQL query template.
GraphQLErrorRuleBook	Registry for GraphQL-to-provider exception mapping rules.
Rule	Simple rule for mapping GraphQL errors to provider exceptions.

Functions

raise_for_graphql_errors(→ dict[str, Any])	Raise typed DS exceptions for GraphQL payload errors.
build_mutation(→ str)	Build a rendered GraphQL mutation for write operations.
build_query(→ str)	Build a rendered GraphQL query for read operations.
build_variables(→ dict[str, Any])	Build GraphQL variables payload for an operation.

Package Contents

ds_provider_xledger_py_lib.utils.raise_for_graphql_errors(*, body: Any) → dict[str, Any]

Raise typed DS exceptions for GraphQL payload errors.

GraphQL servers may return HTTP 200 while reporting failures under an errors field in the response payload.

Parameters:

body – The GraphQL response body to inspect.

Returns:

The original GraphQL response body when no error is present.

class ds_provider_xledger_py_lib.utils.EntryPointMetaData

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

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.IntrospectionService

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

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

Return metadata for a single operation.

Parameters:

operation – Operation to resolve.

Returns:

Metadata for the requested operation.

class ds_provider_xledger_py_lib.utils.MetaData

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

ds_provider_xledger_py_lib.utils.build_mutation(*, metadata: ds_provider_xledger_py_lib.utils.introspection.MetaData, variables: dict[str, Any], return_fields: list[str] | None = None) → str

Build a rendered GraphQL mutation for write operations.

Parameters:

• metadata – Operation metadata.

• variables – Resolved variables required by the mutation template.

• return_fields – Optional explicit fields to return.

Returns:

GraphQL mutation text.

ds_provider_xledger_py_lib.utils.build_query(*, metadata: ds_provider_xledger_py_lib.utils.introspection.MetaData, **kwargs: Any) → str

Build a rendered GraphQL query for read operations.

Parameters:

• metadata – Operation metadata.

• **kwargs – Query keyword arguments. fields controls selection set.

Returns:

GraphQL query text.

ds_provider_xledger_py_lib.utils.build_variables(*, obj: pandas.DataFrame, operation: ds_provider_xledger_py_lib.enums.OperationType, metadata: ds_provider_xledger_py_lib.utils.introspection.MetaData) → dict[str, Any]

Build GraphQL variables payload for an operation.

Parameters:

• obj – Input dataframe.

• operation – Dataset operation type.

• metadata – Loaded operation metadata.

Returns:

Variables dictionary matching the query template.

class ds_provider_xledger_py_lib.utils.GraphQLErrorRuleBook

Registry for GraphQL-to-provider exception mapping rules.

_rules: tuple[Rule, Ellipsis]

classmethod resolve(*, code: str, extension_code: str, error_message: str) → ResolvedRule | None

Resolve a GraphQL error to a mapped provider exception.

Parameters:

• code – The error code to resolve.

• extension_code – The extension code to resolve.

• error_message – The error message to resolve.

Returns:

The resolved rule.

static _match_source(*, rule: Rule, message_lower: str, code: str, extension_code: str) → str | None

Return the source used to match the provided rule.

Parameters:

• rule – The rule to match.

• message_lower – The lowercased error message to match.

• code – The error code to match.

• extension_code – The extension code to match.

Returns:

A source label when the current error matches the provided rule.

class ds_provider_xledger_py_lib.utils.Rule

Bases: NamedTuple

Simple rule for mapping GraphQL errors to provider exceptions.

exc_cls: type[Exception]

code: str | None = None

extension_code: str | None = None

message_keywords: tuple[str, Ellipsis] = ()

default_message: str | None = None