Skip to content

store_session

MLTE catalog store interface implementation.

CatalogEntryMapper

Bases: ResourceMapper

A interface for mapping CRUD actions to store entries.

Source code in mlte/store/catalog/store_session.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
class CatalogEntryMapper(ResourceMapper):
    """A interface for mapping CRUD actions to store entries."""

    def create(self, entry: CatalogEntry, context: Any = None) -> CatalogEntry:
        raise NotImplementedError(self.NOT_IMPLEMENTED_ERROR_MSG)

    def edit(self, entry: CatalogEntry, context: Any = None) -> CatalogEntry:
        raise NotImplementedError(self.NOT_IMPLEMENTED_ERROR_MSG)

    def read(self, entry_id: str, context: Any = None) -> CatalogEntry:
        raise NotImplementedError(self.NOT_IMPLEMENTED_ERROR_MSG)

    def list(self, context: Any = None) -> typing.List[str]:
        raise NotImplementedError(self.NOT_IMPLEMENTED_ERROR_MSG)

    def delete(self, entry_id: str, context: Any = None) -> CatalogEntry:
        raise NotImplementedError(self.NOT_IMPLEMENTED_ERROR_MSG)

    def create_with_header(
        self,
        entry: CatalogEntry,
        context: Any = None,
        user: typing.Optional[str] = None,
    ) -> CatalogEntry:
        """Create an entry, generating the timestamp and adding creator. Internally calls the appropriate create implementation."""
        entry.header.created = int(time.time())
        entry.header.creator = user
        return self.create(entry, context)

    def edit_with_header(
        self,
        entry: CatalogEntry,
        context: Any = None,
        user: typing.Optional[str] = None,
    ) -> CatalogEntry:
        """Edit an entry, generating the proper timestamp. Internally calls the appropriate create implementation."""
        entry.header.updated = int(time.time())
        entry.header.updater = user
        return self.edit(entry, context)

create_with_header(entry, context=None, user=None)

Create an entry, generating the timestamp and adding creator. Internally calls the appropriate create implementation.

Source code in mlte/store/catalog/store_session.py 
61
62
63
64
65
66
67
68
69
70
def create_with_header(
    self,
    entry: CatalogEntry,
    context: Any = None,
    user: typing.Optional[str] = None,
) -> CatalogEntry:
    """Create an entry, generating the timestamp and adding creator. Internally calls the appropriate create implementation."""
    entry.header.created = int(time.time())
    entry.header.creator = user
    return self.create(entry, context)

edit_with_header(entry, context=None, user=None)

Edit an entry, generating the proper timestamp. Internally calls the appropriate create implementation.

Source code in mlte/store/catalog/store_session.py 
72
73
74
75
76
77
78
79
80
81
def edit_with_header(
    self,
    entry: CatalogEntry,
    context: Any = None,
    user: typing.Optional[str] = None,
) -> CatalogEntry:
    """Edit an entry, generating the proper timestamp. Internally calls the appropriate create implementation."""
    entry.header.updated = int(time.time())
    entry.header.updater = user
    return self.edit(entry, context)

CatalogStoreSession

Bases: StoreSession

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

Source code in mlte/store/catalog/store_session.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
class CatalogStoreSession(StoreSession):
    """The base class for all implementations of the MLTE catalog store session."""

    entry_mapper: CatalogEntryMapper
    """Mapper for the entry resource."""

    read_only: bool = False
    """Whether this session is read only or not."""

    storage: Storage
    """The underlying storage method."""

    def get_uri(self) -> StoreURI:
        """Returns the URI of the store."""
        return self.storage.get_uri()

entry_mapper instance-attribute

Mapper for the entry resource.

read_only = False class-attribute instance-attribute

Whether this session is read only or not.

storage instance-attribute

The underlying storage method.

get_uri()

Returns the URI of the store.

Source code in mlte/store/catalog/store_session.py 
31
32
33
def get_uri(self) -> StoreURI:
    """Returns the URI of the store."""
    return self.storage.get_uri()

ManagedCatalogSession

Bases: ManagedSession

A simple context manager for store sessions.

Source code in mlte/store/catalog/store_session.py 
36
37
38
39
40
class ManagedCatalogSession(ManagedSession):
    """A simple context manager for store sessions."""

    def __enter__(self) -> CatalogStoreSession:
        return typing.cast(CatalogStoreSession, self.session)