main

mlte/backend/main.py

Entry point for MLTE artifact store server.

run(host, port, store_uri, catalog_uris, allowed_origins, jwt_secret)

Run the artifact store application.

Parameters:

Name	Type	Description	Default
host	str	The application host	required
port	int	The application port	required
store_uri	str	The store URI string	required
catalog_uris	Dict[str, str]	A dict of URIs for catalog stores	required
allowed_origins	List[str]	A list of allowed CORS origins	required
jwt_secret	str	A secret random string key used to sign tokens	required

Returns:

Type	Description
int	Return code

Source code in mlte/backend/main.py

def run(
    host: str,
    port: int,
    store_uri: str,
    catalog_uris: Dict[str, str],
    allowed_origins: List[str],
    jwt_secret: str,
) -> int:
    """
    Run the artifact store application.
    :param host: The application host
    :param port: The application port
    :param store_uri: The store URI string
    :param catalog_uris: A dict of URIs for catalog stores
    :param allowed_origins: A list of allowed CORS origins
    :param jwt_secret: A secret random string key used to sign tokens
    :return: Return code
    """
    # TODO(Kyle): use log level from arguments.
    logging.basicConfig(level=logging.INFO)

    # Resolve hosts and validate resolved origins.
    allowed_origins = util.resolve_hosts(allowed_origins)
    _ = _validate_origins(allowed_origins)
    logging.info(f"Allowed origins: {allowed_origins}")

    # The global FastAPI application
    app = app_factory.create(allowed_origins)

    logging.info(
        f"Backend using artifact and user store URI of type: {StoreURI.from_string(store_uri).type}"
    )

    # Initialize the backing artifact store instance
    artifact_store = artifact_store_factory.create_artifact_store(store_uri)
    if artifact_store.uri.type == StoreType.REMOTE_HTTP:
        raise RuntimeError("Cannot run backend with remote HTTP store.")
    state.set_artifact_store(artifact_store)

    # Initialize the backing user store instance. Assume same store as artifact one for now.
    # TODO: allow for separate config of uri here
    user_store = user_store_factory.create_user_store(store_uri)
    state.set_user_store(user_store)

    # First add the sample catalog store.
    sample_catalog = SampleCatalog.setup_sample_catalog(
        stores_uri=artifact_store.uri
    )
    state.add_catalog_store(
        store=sample_catalog, id=SampleCatalog.SAMPLE_CATALOG_ID
    )

    # Add all configured catalog stores.
    for id, uri in catalog_uris.items():
        logging.info(
            f"Adding catalog with id '{id}' and URI of type: {StoreURI.from_string(uri).type}"
        )
        state.add_catalog_store_from_uri(uri, id)

    # Initialize the backing custom list store instance. Assume same store as artifact one for now.
    # TODO: allow for separate config of uri here
    # TODO: Remove this check once RDBS and HTTP are implemented
    parsed_uri = StoreURI.from_string(artifact_store.uri.uri)
    if (
        parsed_uri.type == StoreType.LOCAL_MEMORY
        or parsed_uri.type == StoreType.LOCAL_FILESYSTEM
    ):
        custom_list_store = InitialCustomLists.setup_custom_list_store(
            stores_uri=artifact_store.uri
        )
        state.set_custom_list_store(custom_list_store)

    # Set the token signing key.
    state.set_token_key(jwt_secret)

    # Run the server
    uvicorn.run(app, host=host, port=port)
    return EXIT_SUCCESS