base

MLTE general store interface.

ManagedSession

A simple context manager for store sessions.

Source code in mlte/store/base.py

class ManagedSession:
    """A simple context manager for store sessions."""

    def __init__(self, session: StoreSession) -> None:
        self.session = session
        """The wrapped session."""

    def __enter__(self) -> StoreSession:
        return self.session

    def __exit__(self, exc_type, exc_value, exc_tb) -> None:
        self.session.close()

session = session instance-attribute

The wrapped session.

ResourceMapper

Bases: ABC

A generic interface for mapping CRUD actions to store specific resources.

Source code in mlte/store/base.py

class ResourceMapper(ABC):
    """A generic interface for mapping CRUD actions to store specific resources."""

    NOT_IMPLEMENTED_ERROR_MSG = (
        "Cannot invoke method that has not been implemented for this mapper."
    )
    """Default error message for this abstract class."""

    DEFAULT_LIST_LIMIT = 100
    """Default limit for lists."""

    def __init__(
        self, *, validators: CompositeValidator = CompositeValidator()
    ) -> None:
        self.validators: CompositeValidator = validators
        """A reference to the store validators."""

    @abstractmethod
    def create(self, new_resource: Any, context: Any) -> Any:
        """
        Create a new resource.
        :param new_resource: The data to create the resource
        :param context: Any additional context needed for this resource.
        :return: The created resource
        :throws: ErrorAlreadyExists if found.
        """
        raise NotImplementedError(self.NOT_IMPLEMENTED_ERROR_MSG)

    @abstractmethod
    def edit(self, updated_resource: Any, context: Any) -> Any:
        """
        Edit an existing resource.
        :param updated_resource: The data to edit the resource
        :param context: Any additional context needed for this resource.
        :return: The edited resource
        :throws: ErrorNotFound if not found.
        """
        raise NotImplementedError(self.NOT_IMPLEMENTED_ERROR_MSG)

    @abstractmethod
    def read(self, resource_identifier: str, context: Any) -> Any:
        """
        Read a resource.
        :param resource_identifier: The identifier for the resource
        :param context: Any additional context needed for this resource.
        :return: The resource
        :throws: ErrorNotFound if not found.
        """
        raise NotImplementedError(self.NOT_IMPLEMENTED_ERROR_MSG)

    @abstractmethod
    def list(self, context: Any) -> List[str]:
        """
        List all resources of this type in the store.
        :param context: Any additional context needed for this resource.
        :return: A collection of identifiers for all resources of this type
        """
        raise NotImplementedError(self.NOT_IMPLEMENTED_ERROR_MSG)

    @abstractmethod
    def delete(self, resource_identifier: str, context: Any) -> Any:
        """
        Delete a resource.
        :param resource_identifier: The identifier for the resource
        :param context: Any additional context needed for this resource.
        :return: The deleted resource
        :throws: ErrorNotFound if not found.
        """
        raise NotImplementedError(self.NOT_IMPLEMENTED_ERROR_MSG)

    def list_details(
        self,
        context: Any = None,
        limit: int = DEFAULT_LIST_LIMIT,
        offset: int = 0,
    ) -> List[Any]:
        """
        Read details of resources within limit and offset.
        :param context: Any additional context needed for this resource.
        :param limit: The limit on resources to read
        :param offset: The offset on resources to read
        :return: The read resources
        """
        entry_ids = self.list(context)
        return [self.read(entry_id, context) for entry_id in entry_ids][
            offset : offset + limit
        ]

    def search(self, query: Query, context: Any = None) -> List[Any]:
        """
        Read a collection of resources, optionally filtered.
        :param query: The resource query to apply
        :param context: Any additional context needed for this resource.
        :return: A collection of resources that satisfy the filter
        """
        # TODO: not the most efficient way, since it loads all items first, before filtering.
        entries = self.list_details(context)
        return [entry for entry in entries if query.filter.match(entry)]

DEFAULT_LIST_LIMIT = 100 class-attribute instance-attribute

Default limit for lists.

NOT_IMPLEMENTED_ERROR_MSG = 'Cannot invoke method that has not been implemented for this mapper.' class-attribute instance-attribute

Default error message for this abstract class.

validators = validators instance-attribute

A reference to the store validators.

create(new_resource, context) abstractmethod

Create a new resource. :throws: ErrorAlreadyExists if found.

Parameters:

Name	Type	Description	Default
new_resource	Any	The data to create the resource	required
context	Any	Any additional context needed for this resource.	required

Returns:

Type	Description
Any	The created resource

Source code in mlte/store/base.py

@abstractmethod
def create(self, new_resource: Any, context: Any) -> Any:
    """
    Create a new resource.
    :param new_resource: The data to create the resource
    :param context: Any additional context needed for this resource.
    :return: The created resource
    :throws: ErrorAlreadyExists if found.
    """
    raise NotImplementedError(self.NOT_IMPLEMENTED_ERROR_MSG)

delete(resource_identifier, context) abstractmethod

Delete a resource. :throws: ErrorNotFound if not found.

Parameters:

Name	Type	Description	Default
resource_identifier	str	The identifier for the resource	required
context	Any	Any additional context needed for this resource.	required

Returns:

Type	Description
Any	The deleted resource

Source code in mlte/store/base.py

@abstractmethod
def delete(self, resource_identifier: str, context: Any) -> Any:
    """
    Delete a resource.
    :param resource_identifier: The identifier for the resource
    :param context: Any additional context needed for this resource.
    :return: The deleted resource
    :throws: ErrorNotFound if not found.
    """
    raise NotImplementedError(self.NOT_IMPLEMENTED_ERROR_MSG)

edit(updated_resource, context) abstractmethod

Edit an existing resource. :throws: ErrorNotFound if not found.

Parameters:

Name	Type	Description	Default
updated_resource	Any	The data to edit the resource	required
context	Any	Any additional context needed for this resource.	required

Returns:

Type	Description
Any	The edited resource

Source code in mlte/store/base.py

@abstractmethod
def edit(self, updated_resource: Any, context: Any) -> Any:
    """
    Edit an existing resource.
    :param updated_resource: The data to edit the resource
    :param context: Any additional context needed for this resource.
    :return: The edited resource
    :throws: ErrorNotFound if not found.
    """
    raise NotImplementedError(self.NOT_IMPLEMENTED_ERROR_MSG)

list(context) abstractmethod

List all resources of this type in the store.

Parameters:

Name	Type	Description	Default
context	Any	Any additional context needed for this resource.	required

Returns:

Type	Description
List[str]	A collection of identifiers for all resources of this type

Source code in mlte/store/base.py

@abstractmethod
def list(self, context: Any) -> List[str]:
    """
    List all resources of this type in the store.
    :param context: Any additional context needed for this resource.
    :return: A collection of identifiers for all resources of this type
    """
    raise NotImplementedError(self.NOT_IMPLEMENTED_ERROR_MSG)

list_details(context=None, limit=DEFAULT_LIST_LIMIT, offset=0)

Read details of resources within limit and offset.

Parameters:

Name	Type	Description	Default
context	Any	Any additional context needed for this resource.	None
limit	int	The limit on resources to read	DEFAULT_LIST_LIMIT
offset	int	The offset on resources to read	0

Returns:

Type	Description
List[Any]	The read resources

Source code in mlte/store/base.py

def list_details(
    self,
    context: Any = None,
    limit: int = DEFAULT_LIST_LIMIT,
    offset: int = 0,
) -> List[Any]:
    """
    Read details of resources within limit and offset.
    :param context: Any additional context needed for this resource.
    :param limit: The limit on resources to read
    :param offset: The offset on resources to read
    :return: The read resources
    """
    entry_ids = self.list(context)
    return [self.read(entry_id, context) for entry_id in entry_ids][
        offset : offset + limit
    ]

read(resource_identifier, context) abstractmethod

Read a resource. :throws: ErrorNotFound if not found.

Parameters:

Name	Type	Description	Default
resource_identifier	str	The identifier for the resource	required
context	Any	Any additional context needed for this resource.	required

Returns:

Type	Description
Any	The resource

Source code in mlte/store/base.py

@abstractmethod
def read(self, resource_identifier: str, context: Any) -> Any:
    """
    Read a resource.
    :param resource_identifier: The identifier for the resource
    :param context: Any additional context needed for this resource.
    :return: The resource
    :throws: ErrorNotFound if not found.
    """
    raise NotImplementedError(self.NOT_IMPLEMENTED_ERROR_MSG)

search(query, context=None)

Read a collection of resources, optionally filtered.

Parameters:

Name	Type	Description	Default
query	Query	The resource query to apply	required
context	Any	Any additional context needed for this resource.	None

Returns:

Type	Description
List[Any]	A collection of resources that satisfy the filter

Source code in mlte/store/base.py

def search(self, query: Query, context: Any = None) -> List[Any]:
    """
    Read a collection of resources, optionally filtered.
    :param query: The resource query to apply
    :param context: Any additional context needed for this resource.
    :return: A collection of resources that satisfy the filter
    """
    # TODO: not the most efficient way, since it loads all items first, before filtering.
    entries = self.list_details(context)
    return [entry for entry in entries if query.filter.match(entry)]

Store

An abstract store. A Store instance is the "static" part of a store configuration. In contrast, a StoreSession represents an active session with the store.

Source code in mlte/store/base.py

class Store:
    """
    An abstract store. A Store instance is the "static" part of a store configuration.
    In contrast, a StoreSession represents an active session with the store.
    """

    def __init__(self, *, uri: StoreURI):
        """
        Initialize a Store instance.
        :param uri: The parsed store URI
        """

        self.uri = uri
        """The parsed store URI."""

        self.validators: CompositeValidator = CompositeValidator()
        """The inter store validators for this store."""

    def set_validators(self, validators: CompositeValidator):
        self.validators = validators

    def session(self) -> StoreSession:
        """
        Return a session handle for the store instance.
        :return: The session handle
        """
        raise NotImplementedError("Can't call session on a base Store.")

uri = uri instance-attribute

The parsed store URI.

validators = CompositeValidator() instance-attribute

The inter store validators for this store.

__init__(*, uri)

Initialize a Store instance.

Parameters:

Name	Type	Description	Default
uri	StoreURI	The parsed store URI	required

Source code in mlte/store/base.py

def __init__(self, *, uri: StoreURI):
    """
    Initialize a Store instance.
    :param uri: The parsed store URI
    """

    self.uri = uri
    """The parsed store URI."""

    self.validators: CompositeValidator = CompositeValidator()
    """The inter store validators for this store."""

session()

Return a session handle for the store instance.

Returns:

Type	Description
StoreSession	The session handle

Source code in mlte/store/base.py

def session(self) -> StoreSession:
    """
    Return a session handle for the store instance.
    :return: The session handle
    """
    raise NotImplementedError("Can't call session on a base Store.")

StoreSession

Bases: Protocol

The base class for all implementations of the MLTE store session.

Source code in mlte/store/base.py

class StoreSession(Protocol):
    """The base class for all implementations of the MLTE store session."""

    def close(self) -> None:
        """Close the session."""
        ...

close()

Close the session.

Source code in mlte/store/base.py

def close(self) -> None:
    """Close the session."""
    ...

StoreType

Bases: Enum

Represents the type of an MLTE store.

Source code in mlte/store/base.py

class StoreType(Enum):
    """Represents the type of an MLTE store."""

    LOCAL_MEMORY = "local_memory"
    """An in-memory store implementation."""

    LOCAL_FILESYSTEM = "local_filesystem"
    """A local filesystem implementation."""

    REMOTE_HTTP = "remote_http"
    """A remote HTTP implementation."""

    RELATIONAL_DB = "database"
    """A relational database system implementation."""

LOCAL_FILESYSTEM = 'local_filesystem' class-attribute instance-attribute

A local filesystem implementation.

LOCAL_MEMORY = 'local_memory' class-attribute instance-attribute

An in-memory store implementation.

RELATIONAL_DB = 'database' class-attribute instance-attribute

A relational database system implementation.

REMOTE_HTTP = 'remote_http' class-attribute instance-attribute

A remote HTTP implementation.

StoreURI

Represents the URI for an store instance.

Source code in mlte/store/base.py

class StoreURI:
    """Represents the URI for an store instance."""

    PREFIXES = {
        StoreType.LOCAL_MEMORY: ["memory"],
        StoreType.LOCAL_FILESYSTEM: ["fs", "local"],
        StoreType.REMOTE_HTTP: ["http"],
        StoreType.RELATIONAL_DB: [
            "sqlite",
            "mysql",
            "postgresql",
            "oracle",
            "mssql",
        ],
    }
    """Valid prefixes by store type."""

    DELIMITER = "://"
    """Delimiter used to separate prefixes from rest of the path."""

    DEFAULT_STORES_FOLDER = "default_store"
    """Default root folder for all built-in file-based stores."""

    def __init__(self, uri: str, type: StoreType, path: str, prefix: str):
        """
        Initialize a StoreURI instance.
        :param uri: The URI
        :param type: The type of the backend store
        :param path: The rest of the URI's path with the prefix removed
        """
        self.uri = uri
        """The string that represents the URI."""

        self.type = type
        """The type identifier for the URI."""

        self.path = path
        """The rest of the path, with the prefix removed."""

        self.prefix = prefix
        """The actual prefix used in the type."""

    @staticmethod
    def from_string(uri: str) -> StoreURI:
        """
        Parse a StoreURI from a string.
        :param uri: The URI
        :return: The parsed StoreURI
        """
        prefix, path = StoreURI._parse_uri(uri)

        try:
            type = StoreURI.get_type(prefix)
            return StoreURI(uri, type, path, prefix)
        except Exception as e:
            # If we got here, the stucture of the URI is fine, but the prefix is unknown.
            raise RuntimeError(f"Unsupported store URI: {uri}") from e

    @staticmethod
    def from_type(type: StoreType, path: str = "") -> StoreURI:
        """Creates a parsed StoreURI from the given type, and optional path."""
        return StoreURI.from_string(StoreURI.create_uri_string(type, path))

    @staticmethod
    def get_type(prefix: str) -> StoreType:
        """Returns the type given a prefix."""
        for type in StoreType:
            if prefix.startswith(tuple(StoreURI.PREFIXES[type])):
                return type
        raise RuntimeError(
            f"Prefix {prefix} is not associated to any known type."
        )

    @staticmethod
    def _parse_uri(uri: str) -> tuple[str, str]:
        """Split an URI into its prefix and the rest of the path."""
        parts = uri.split(StoreURI.DELIMITER)
        if len(parts) != 2:
            # Even if path is an empty string, we'll only get 2 parts if the delimiter is present.
            raise RuntimeError(f"Invalid store URI: {uri}")
        else:
            prefix = parts[0]
            path = parts[1]
            return prefix, path

    @staticmethod
    def create_uri_string(type: StoreType, path: str = "") -> str:
        """Creates a URI using the default (first) prefix for the given type."""
        prefix = StoreURI.PREFIXES[type][0]
        return f"{prefix}{StoreURI.DELIMITER}{path}"

    @staticmethod
    def create_default_fs_uri() -> StoreURI:
        """Creates the default StoreURI for file system stores."""
        uri_string = StoreURI.create_uri_string(
            type=StoreType.LOCAL_FILESYSTEM,
            path=StoreURI.DEFAULT_STORES_FOLDER,
        )
        return StoreURI.from_string(uri_string)

    def __str__(self) -> str:
        return f"{self.type}:{self.uri}"

DEFAULT_STORES_FOLDER = 'default_store' class-attribute instance-attribute

Default root folder for all built-in file-based stores.

DELIMITER = '://' class-attribute instance-attribute

Delimiter used to separate prefixes from rest of the path.

PREFIXES = {StoreType.LOCAL_MEMORY: ['memory'], StoreType.LOCAL_FILESYSTEM: ['fs', 'local'], StoreType.REMOTE_HTTP: ['http'], StoreType.RELATIONAL_DB: ['sqlite', 'mysql', 'postgresql', 'oracle', 'mssql']} class-attribute instance-attribute

Valid prefixes by store type.

path = path instance-attribute

The rest of the path, with the prefix removed.

prefix = prefix instance-attribute

The actual prefix used in the type.

type = type instance-attribute

The type identifier for the URI.

uri = uri instance-attribute

The string that represents the URI.

__init__(uri, type, path, prefix)

Initialize a StoreURI instance.

Parameters:

Name	Type	Description	Default
uri	str	The URI	required
type	StoreType	The type of the backend store	required
path	str	The rest of the URI's path with the prefix removed	required

Source code in mlte/store/base.py

def __init__(self, uri: str, type: StoreType, path: str, prefix: str):
    """
    Initialize a StoreURI instance.
    :param uri: The URI
    :param type: The type of the backend store
    :param path: The rest of the URI's path with the prefix removed
    """
    self.uri = uri
    """The string that represents the URI."""

    self.type = type
    """The type identifier for the URI."""

    self.path = path
    """The rest of the path, with the prefix removed."""

    self.prefix = prefix
    """The actual prefix used in the type."""

create_default_fs_uri() staticmethod

Creates the default StoreURI for file system stores.

Source code in mlte/store/base.py

@staticmethod
def create_default_fs_uri() -> StoreURI:
    """Creates the default StoreURI for file system stores."""
    uri_string = StoreURI.create_uri_string(
        type=StoreType.LOCAL_FILESYSTEM,
        path=StoreURI.DEFAULT_STORES_FOLDER,
    )
    return StoreURI.from_string(uri_string)

create_uri_string(type, path='') staticmethod

Creates a URI using the default (first) prefix for the given type.

Source code in mlte/store/base.py

@staticmethod
def create_uri_string(type: StoreType, path: str = "") -> str:
    """Creates a URI using the default (first) prefix for the given type."""
    prefix = StoreURI.PREFIXES[type][0]
    return f"{prefix}{StoreURI.DELIMITER}{path}"

from_string(uri) staticmethod

Parse a StoreURI from a string.

Parameters:

Name	Type	Description	Default
uri	str	The URI	required

Returns:

Type	Description
StoreURI	The parsed StoreURI

Source code in mlte/store/base.py

@staticmethod
def from_string(uri: str) -> StoreURI:
    """
    Parse a StoreURI from a string.
    :param uri: The URI
    :return: The parsed StoreURI
    """
    prefix, path = StoreURI._parse_uri(uri)

    try:
        type = StoreURI.get_type(prefix)
        return StoreURI(uri, type, path, prefix)
    except Exception as e:
        # If we got here, the stucture of the URI is fine, but the prefix is unknown.
        raise RuntimeError(f"Unsupported store URI: {uri}") from e

from_type(type, path='') staticmethod

Creates a parsed StoreURI from the given type, and optional path.

Source code in mlte/store/base.py

@staticmethod
def from_type(type: StoreType, path: str = "") -> StoreURI:
    """Creates a parsed StoreURI from the given type, and optional path."""
    return StoreURI.from_string(StoreURI.create_uri_string(type, path))

get_type(prefix) staticmethod

Returns the type given a prefix.

Source code in mlte/store/base.py

@staticmethod
def get_type(prefix: str) -> StoreType:
    """Returns the type given a prefix."""
    for type in StoreType:
        if prefix.startswith(tuple(StoreURI.PREFIXES[type])):
            return type
    raise RuntimeError(
        f"Prefix {prefix} is not associated to any known type."
    )