Documentation for ToMeDa

tomeda.repository.repository_connector

This module defines the TomedaRepositoryConnector base class.

The TomedaRepositoryConnector is an abstract base class that is used to establish a connection to a specific repository and perform various operations like fetching metadata, uploading data, etc. It provides the basic structure and methods for the connection and leaves the specific implementation to its subclasses.

Each repository that needs to be connected will have a specific subclass of TomedaRepositoryConnector, and those subclasses will be placed in their own separate modules.

Classes:

Name	Description
TomedaRepositoryConnector : ABC	An abstract base class that provides a common interface for all repository connectors. Defines a set of methods that subclasses should implement.

logger module-attribute

logger: TraceLogger = getLogger(__name__)

TomedaRepositoryConnector

TomedaRepositoryConnector(
    repository_type: str,
    validator: TomedaRepositoryValidator,
    url: str,
    api_token: str,
)

Bases: ABC

Abstract base class for connecting to a specific repository.

This class provides the basic structure and methods for interacting with a repository, including setting up the connection, fetching metadata, uploading data, etc. The specific details of how these operations are performed should be implemented in the subclasses.

Parameters:

Name	Type	Description	Default
url	str	The base URL of the repository to connect to.	required
api_token	str	The API token to use for authenticating with the repository.	required

Attributes:

Name	Type	Description
repository_type	str	The type of the repository e.g. 'dataverse', 'coscine', etc.
base_url	str	The base URL of the repository.
api_token	str	The API token used for authentication.
base_url_api	str	The base URL of the repository's API.
USER_AGENT	str	The user agent string to use when making requests to the repository.
API_VERSION	str	The version of the API to use.

Source code in tomeda/repository/repository_connector.py

def __init__(
    self,
    repository_type: str,
    validator: TomedaRepositoryValidator,
    url: str,
    api_token: str,
):
    self.repository_type = repository_type
    self.validator = validator
    self.base_url = url
    self.base_url_api = self._set_base_url_api()
    self.validator.base_url = self.base_url
    self._validate_url(self.base_url)
    self.api_token = api_token
    self.http = TomedaHttpHandler()

API_VERSION class-attribute instance-attribute

API_VERSION = ''

USER_AGENT class-attribute instance-attribute

USER_AGENT = 'ToMeDa'

api_token instance-attribute

api_token = api_token

base_url class-attribute instance-attribute

base_url: str = url

base_url_api class-attribute instance-attribute

base_url_api: str = _set_base_url_api()

http instance-attribute

http = TomedaHttpHandler()

repository_type instance-attribute

repository_type = repository_type

validator instance-attribute

validator = validator

get_collection abstractmethod

get_collection(
    collection_id: str,
) -> RepositoryDataCollection

Source code in tomeda/repository/repository_connector.py

@abstractmethod
def get_collection(self, collection_id: str) -> RepositoryDataCollection:
    pass

get_collection_contents abstractmethod

get_collection_contents(
    collection_id: str,
) -> list[RepositoryDataCollectionContent]

Source code in tomeda/repository/repository_connector.py

@abstractmethod
def get_collection_contents(
    self, collection_id: str
) -> list[RepositoryDataCollectionContent]:
    pass

get_collection_metadata_blocks abstractmethod

get_collection_metadata_blocks(
    collection_id: str,
) -> list[RepositoryDataMetadataBlock]

Source code in tomeda/repository/repository_connector.py

@abstractmethod
def get_collection_metadata_blocks(
    self, collection_id: str
) -> list[RepositoryDataMetadataBlock]:
    pass

get_dataset abstractmethod

get_dataset(
    dataset_id: int, dataset_name: str
) -> RepositoryDataDataset

Source code in tomeda/repository/repository_connector.py

@abstractmethod
def get_dataset(
    self, dataset_id: int, dataset_name: str
) -> RepositoryDataDataset:
    pass

get_metadata_block_definition abstractmethod

get_metadata_block_definition(
    block_id: int | str,
) -> list[RepositoryDataMetadataBlockFieldDefinition]

Source code in tomeda/repository/repository_connector.py

@abstractmethod
def get_metadata_block_definition(
    self, block_id: int | str
) -> list[RepositoryDataMetadataBlockFieldDefinition]:
    pass

get_metadata_blocks abstractmethod

get_metadata_blocks() -> list[RepositoryDataMetadataBlock]

Source code in tomeda/repository/repository_connector.py

@abstractmethod
def get_metadata_blocks(self) -> list[RepositoryDataMetadataBlock]:
    pass

get_server_info abstractmethod

get_server_info() -> RepositoryDataServer

Source code in tomeda/repository/repository_connector.py

@abstractmethod
def get_server_info(self) -> RepositoryDataServer:
    pass

get_user_info abstractmethod

get_user_info() -> RepositoryDataUser

Source code in tomeda/repository/repository_connector.py

@abstractmethod
def get_user_info(self) -> RepositoryDataUser:
    pass

TomedaRepositoryFactory

A factory class to dynamically create instances of TomedaRepositoryConnector.

The factory uses the provided repository name to determine which specific repository connector to instantiate. It ensures that the specified module and class names are valid, imports the required module, and then instantiates the specified class, passing any additional arguments and keyword arguments to the class constructor.

The factory ensures that the specified module and class names are valid, imports the required module, and then instantiates the specified class, passing any additional arguments and keyword arguments to the class constructor.

This factory was chosen as the design pattern to facilitate the dynamic importation of classes. This dynamic import capability enables the system to seamlessly integrate with plugins, as specific repository connectors can be determined, imported, and instantiated at runtime based on the provided repository name.

Methods:

Name	Description
__new__	Creates and returns an instance of a TomedaRepositoryConnector subclass.

TActualClass class-attribute instance-attribute

TActualClass = TypeVar('TActualClass')

TBaseClass class-attribute instance-attribute

TBaseClass = TypeVar('TBaseClass')

__new__

__new__(
    repository_name: str, *args: Any, **kwargs: Any
) -> TomedaRepositoryConnector

Create and return an instance of a TomedaRepositoryConnector subclass.

The specific subclass is determined based on the provided repository name. Any additional arguments or keyword arguments are passed to the class constructor.

Parameters:

Name	Type	Description	Default
repository_name	str	The name of the repository to determine which connector to instantiate.	required
*args	tuple[Any, ...]	Positional arguments to be passed to the class constructor.	()
**kwargs	dict[str, Any]	Keyword arguments to be passed to the class constructor.	{}

Returns:

Type	Description
TomedaRepositoryConnector	An instance of the specified TomedaRepositoryConnector subclass.

Raises:

Type	Description
TomedaFactoryTypeError	If the provided names (repository, module, class) are not strings.
TomedaRepositoryInvalidRepositoryNameError	If the provided repository name is not supported.
TomedaFactoryImportError	If there's an error importing the specified module or class.
TomedaFactoryInvalidClassError	If the imported class is not a subclass of TomedaRepositoryConnector.
TomedaFactoryInstantiationError	If there's an error instantiating the specified class.

Source code in tomeda/repository/repository_connector.py

def __new__(  # type: ignore[misc]
    cls,
    repository_name: str,
    *args: Any,
    **kwargs: Any,
) -> TomedaRepositoryConnector:
    """
    Create and return an instance of a TomedaRepositoryConnector subclass.

    The specific subclass is determined based on the provided repository
    name. Any additional arguments or keyword arguments are passed
    to the class constructor.

    Parameters
    ----------
    repository_name : str
        The name of the repository to determine which connector to
        instantiate.
    *args : tuple[Any, ...]
        Positional arguments to be passed to the class constructor.
    **kwargs : dict[str, Any]
        Keyword arguments to be passed to the class constructor.

    Returns
    -------
    TomedaRepositoryConnector
        An instance of the specified TomedaRepositoryConnector subclass.

    Raises
    ------
    TomedaFactoryTypeError
        If the provided names (repository, module, class) are not strings.
    TomedaRepositoryInvalidRepositoryNameError
        If the provided repository name is not supported.
    TomedaFactoryImportError
        If there's an error importing the specified module or class.
    TomedaFactoryInvalidClassError
        If the imported class is not a subclass of
        TomedaRepositoryConnector.
    TomedaFactoryInstantiationError
        If there's an error instantiating the specified class.
    """
    cls._validate_repository_name(repository_name)
    repository_data = cls._get_repository_data(repository_name)

    validator = cls._create_instance_from_data(
        TomedaRepositoryValidator,
        repository_data["validator"]["module"],
        repository_data["validator"]["class"],
        repository_name,
        (repository_name, "url"),
        {},
    )

    args = (repository_name, validator) + args
    # TODO: Check type of base_class
    return cls._create_instance_from_data(
        TomedaRepositoryConnector,  # type: ignore[type-abstract]
        repository_data["connector"]["module"],
        repository_data["connector"]["class"],
        repository_name,
        args,
        kwargs,
    )