ds_resource_plugin_py_lib.common.resource.client

File: client.py Region: ds_resource_plugin_py_lib/common/resource

Description

Resource client for discovering and managing datasets and linked services using entry points.

Example

from ds_resource_plugin_py_lib.common.resource.client import ResourceClient

client = ResourceClient()

# Inspect discovered resources (populated via Python entry points).
print(client.resources.keys())
print(client.linked_services.keys())
print(client.datasets.keys())

dataset = client.dataset(config={"type": "dataset.example", "version": "1.0.0"})
linked_service = client.linked_service(config={"type": "linked_service.example", "version": "1.0.0"})

linked_service.connect()
print(linked_service.connection)
print(dataset.read())

Attributes

logger

Classes

ResourceClient

Module Contents

ds_resource_plugin_py_lib.common.resource.client.logger

class ds_resource_plugin_py_lib.common.resource.client.ResourceClient

PROTOCOL_GROUP = 'ds.protocols'

PROVIDER_GROUP = 'ds.providers'

_resource_dict: dict[str, dict[str, Any]]

_linked_services: dict[tuple[str, str], ds_resource_plugin_py_lib.common.resource.linked_service.base.LinkedServiceInfo]

_datasets: dict[tuple[str, str], ds_resource_plugin_py_lib.common.resource.dataset.base.DatasetInfo]

classmethod get_instance() → ResourceClient

Get the singleton instance of ResourceClient.

property resources: dict[str, dict[str, Any]]

property linked_services: dict[tuple[str, str], ds_resource_plugin_py_lib.common.resource.linked_service.base.LinkedServiceInfo]

property datasets: dict[tuple[str, str], ds_resource_plugin_py_lib.common.resource.dataset.base.DatasetInfo]

_discover_resources(group: str) → None

Discover protocol/provider packages via entry points. Each entry point must target a Python package that contains resource.yaml in its root.

_scan_resource_directory(resource_dir: str) → None

Scan a resource directory for resource.yaml. Checks root first (new behavior), then subdirectories (old behavior). :param resource_dir: Path to the resource directory.

_load_resource_from_path(path: str) → None

Load resource configuration from a directory path.

Parameters:

path – Path to the resource directory.

_parse_linked_services(config: dict[str, Any]) → None

Parse linked services from resource configuration.

Parameters:

config – Resource configuration dictionary.

_parse_datasets(config: dict[str, Any]) → None

Parse datasets from resource configuration.

Parameters:

config – Resource configuration dictionary.

_get_dataset_model_cls(_type: str, version: str) → type[ds_resource_plugin_py_lib.common.resource.dataset.base.Dataset[Any, Any, Any, Any]]

Get a dataset model class by type and optionally version.

Parameters:

• _type – str

• version – str

Returns:

Type[Dataset]

_get_linked_service_model_cls(_type: str, version: str) → type[ds_resource_plugin_py_lib.common.resource.linked_service.base.LinkedService[Any]]

Get a linked service model class by type and version.

Parameters:

• _type – The type of the linked service.

• version – str version of the linked service.

Returns:

Type[LinkedService]

linked_service(config: dict[str, Any]) → ds_resource_plugin_py_lib.common.resource.linked_service.base.LinkedService[Any]

Get a linked service instance by configuration.

Parameters:

config – dict containing at least ‘type’ and ‘version’

Returns:

LinkedService

Raises:

DeserializationError – If the linked service cannot be deserialized.

dataset(config: dict[str, Any]) → ds_resource_plugin_py_lib.common.resource.dataset.base.Dataset[Any, Any, Any, Any]

Get a dataset instance by configuration.

Parameters:

config – dict containing at least ‘type’ and ‘version’

Returns:

Dataset

Raises:

DeserializationError – If the dataset cannot be deserialized.