base

mlte/store/base.py

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

    @abstractmethod
    def create(self, new_resource: Any, context: Any = None) -> 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
        """
        raise NotImplementedError(self.NOT_IMPLEMENTED_ERROR_MSG)

    @abstractmethod
    def edit(self, updated_resource: Any, context: Any = None) -> 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
        """
        raise NotImplementedError(self.NOT_IMPLEMENTED_ERROR_MSG)

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

    @abstractmethod
    def list(self, context: Any = None) -> 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 = None) -> 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
        """
        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 = 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.

create(new_resource, context=None) abstractmethod

Create a new resource.

Parameters:

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

Returns:

Type	Description
Any	The created resource

Source code in mlte/store/base.py

@abstractmethod
def create(self, new_resource: Any, context: Any = None) -> 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
    """
    raise NotImplementedError(self.NOT_IMPLEMENTED_ERROR_MSG)

delete(resource_identifier, context=None) abstractmethod

Delete a resource.

Parameters:

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

Returns:

Type	Description
Any	The deleted resource

Source code in mlte/store/base.py

@abstractmethod
def delete(self, resource_identifier: str, context: Any = None) -> 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
    """
    raise NotImplementedError(self.NOT_IMPLEMENTED_ERROR_MSG)

edit(updated_resource, context=None) abstractmethod

Edit an existing resource.

Parameters:

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

Returns:

Type	Description
Any	The edited resource

Source code in mlte/store/base.py

@abstractmethod
def edit(self, updated_resource: Any, context: Any = None) -> 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
    """
    raise NotImplementedError(self.NOT_IMPLEMENTED_ERROR_MSG)

list(context=None) abstractmethod

List all resources of this type in the store.

Parameters:

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

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 = None) -> 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=None) abstractmethod

Read a resource.

Parameters:

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

Returns:

Type	Description
Any	The resource

Source code in mlte/store/base.py

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

search(query=Query(), context=None)

Read a collection of resources, optionally filtered.

Parameters:

Name	Type	Description	Default
query	Query	The resource query to apply	Query()
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 = 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."""

    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.

__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."""

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

    def __init__(self, uri: str, type: StoreType, path: 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."""

    @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)
        for type in StoreType:
            if prefix.startswith(tuple(StoreURI.PREFIXES[type])):
                return StoreURI(uri, type, path)

        # If we got here, the stucture of the URI is fine, but the prefix is unknown.
        raise RuntimeError(f"Unsupported store URI: {uri}")

    @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:
            raise RuntimeError(f"Invalid store URI: {uri}")
        else:
            prefix = parts[0]
            path = parts[1]
            return prefix, path

    @staticmethod
    def get_default_prefix(type: StoreType) -> str:
        """Returns the default prefix for the given type, which will be the first one."""
        return f"{StoreURI.PREFIXES[type][0]}{StoreURI.DELIMITER}"

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

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.

type = type instance-attribute

The type identifier for the URI.

uri = uri instance-attribute

The string that represents the URI.

__init__(uri, type, path)

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):
    """
    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."""

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)
    for type in StoreType:
        if prefix.startswith(tuple(StoreURI.PREFIXES[type])):
            return StoreURI(uri, type, path)

    # If we got here, the stucture of the URI is fine, but the prefix is unknown.
    raise RuntimeError(f"Unsupported store URI: {uri}")

get_default_prefix(type) staticmethod

Returns the default prefix for the given type, which will be the first one.

Source code in mlte/store/base.py

@staticmethod
def get_default_prefix(type: StoreType) -> str:
    """Returns the default prefix for the given type, which will be the first one."""
    return f"{StoreURI.PREFIXES[type][0]}{StoreURI.DELIMITER}"