A catalog with sample code. This class is used to set that catalog up when needed.
Source code in mlte/store/catalog/sample_catalog.py
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
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 | class SampleCatalog:
"""A catalog with sample code. This class is used to set that catalog up when needed."""
@staticmethod
def setup_sample_catalog(
stores_uri: StoreURI,
validators: CompositeValidator,
) -> CatalogStore:
"""
Sets up the sample catalog.
:param stores_uri: The URI of the stores being used (i.e., base folder, base DB, etc).
:return: The sample catalog store.
"""
# Create the actual sample catalog.
print(f"Creating sample catalog at URI: {stores_uri}")
catalog = create_catalog_store(stores_uri, SAMPLE_CATALOG_STORE_ID)
catalog.set_validators(validators)
# Ensure the catalog is always reset to its initial state, and mark it as read only.
SampleCatalog.reset_catalog(catalog)
catalog.read_only = True
return catalog
@staticmethod
def reset_catalog(catalog_store: CatalogStore) -> None:
"""Ensures the sample catalog is reset to default values."""
with ManagedCatalogSession(catalog_store.session()) as store:
# First remove all existing entries.
entry_ids = store.entry_mapper.list()
for entry_id in entry_ids:
store.entry_mapper.delete(entry_id)
# Now populate it again with the default values.
SampleCatalog._populate_catalog(store)
@staticmethod
def _populate_catalog(catalog_session: CatalogStoreSession) -> None:
"""Load all entry files from entry folder and put them into sample catalog."""
num_entries = 0
resources = importlib.resources.files(sample_entries)
with importlib.resources.as_file(resources) as resources_path:
with os.scandir(resources_path) as files:
print("Loading sample catalog entries.")
for file in files:
if file.is_file() and file.name.endswith("json"):
with open(file.path) as open_file:
entry = CatalogEntry(**json.load(open_file))
entry.header.catalog_id = SAMPLE_CATALOG_STORE_ID
catalog_session.entry_mapper.create(entry)
num_entries += 1
print(f"Loaded {num_entries} entries for sample catalog.")
|
reset_catalog(catalog_store)
staticmethod
Ensures the sample catalog is reset to default values.
Source code in mlte/store/catalog/sample_catalog.py
48
49
50
51
52
53
54
55
56
57
58 | @staticmethod
def reset_catalog(catalog_store: CatalogStore) -> None:
"""Ensures the sample catalog is reset to default values."""
with ManagedCatalogSession(catalog_store.session()) as store:
# First remove all existing entries.
entry_ids = store.entry_mapper.list()
for entry_id in entry_ids:
store.entry_mapper.delete(entry_id)
# Now populate it again with the default values.
SampleCatalog._populate_catalog(store)
|
setup_sample_catalog(stores_uri, validators)
staticmethod
Sets up the sample catalog.
Parameters:
| Name |
Type |
Description |
Default |
stores_uri
|
StoreURI
|
The URI of the stores being used (i.e., base folder, base DB, etc).
|
required
|
Returns:
| Type |
Description |
CatalogStore
|
The sample catalog store.
|
Source code in mlte/store/catalog/sample_catalog.py
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46 | @staticmethod
def setup_sample_catalog(
stores_uri: StoreURI,
validators: CompositeValidator,
) -> CatalogStore:
"""
Sets up the sample catalog.
:param stores_uri: The URI of the stores being used (i.e., base folder, base DB, etc).
:return: The sample catalog store.
"""
# Create the actual sample catalog.
print(f"Creating sample catalog at URI: {stores_uri}")
catalog = create_catalog_store(stores_uri, SAMPLE_CATALOG_STORE_ID)
catalog.set_validators(validators)
# Ensure the catalog is always reset to its initial state, and mark it as read only.
SampleCatalog.reset_catalog(catalog)
catalog.read_only = True
return catalog
|