factory

Top-level functions for catalog store creation.

create_catalog_store(parsed_uri, catalog_id=None)

Create a MLTE catalog store instance.

Parameters:

Name	Type	Description	Default
parsed_uri	StoreURI	The URI for the store instance	required
catalog_id	Optional[str]	The id used to identify the catalog.	None

Returns:

Type	Description
CatalogStore	The store instance

Source code in mlte/store/catalog/factory.py

def create_catalog_store(
    parsed_uri: StoreURI, catalog_id: Optional[str] = None
) -> CatalogStore:
    """
    Create a MLTE catalog store instance.
    :param parsed_uri: The URI for the store instance
    :param catalog_id: The id used to identify the catalog.
    :return: The store instance
    """
    if parsed_uri.type == StoreType.LOCAL_MEMORY:
        return InMemoryCatalogStore(uri=parsed_uri)
    if parsed_uri.type == StoreType.LOCAL_FILESYSTEM:
        return FileSystemCatalogStore(uri=parsed_uri, catalog_folder=catalog_id)
    if parsed_uri.type == StoreType.RELATIONAL_DB:
        # Import is here to avoid importing SQL libraries if they have not been installed.
        from mlte.store.catalog.underlying.rdbs.store import (
            RelationalDBCatalogStore,
        )

        return RelationalDBCatalogStore(uri=parsed_uri)
    if parsed_uri.type == StoreType.REMOTE_HTTP:
        return HttpCatalogGroupStore(uri=parsed_uri)

    raise Exception(
        f"Catalog store can't be created, unknown or unsupported URI type received for uri: {parsed_uri}"
    )