session

Session info when using the MLTE library: context (model and version), stores and credentials.

Session

The Session data structure encapsulates package-wide state.

The primary function of the Session data structure is to provide convenient access to the MLTE context for application developers.

Source code in mlte/session/session.py

class Session:
    """
    The Session data structure encapsulates package-wide state.

    The primary function of the Session data structure is to provide
    convenient access to the MLTE context for application developers.
    """

    ENV_CONTEXT_MODEL_VAR = "MLTE_CONTEXT_MODEL"
    ENV_CONTEXT_VERSION_VAR = "MLTE_CONTEXT_VERSION"
    ENV_STORE_URI_VAR = "MLTE_STORE_URI"
    ENV_CURRENT_USER_VAR = "MLTE_CURRENT_USER"
    ENV_CURRENT_PASS_VAR = "MLTE_CURRENT_PASS"
    """Environment variables to get model, version, store_uri and credentials from, if needed."""

    def __init__(self):
        """Constructors, just resets all vars."""
        self.reset()

    def reset(self):
        """Resets all internal state to defaults."""
        self._context: Context | None = None
        """The MLTE context for the session."""

        self._stores: UnifiedStore | None = None
        """All stores in this session."""

        self._credentials: Credentials | None = None
        """Current user and password for auditing and connections."""

    @property
    def context(self) -> Context:
        if self._context is None:
            # If the context has not been manually set, get it from environment.
            model_context = self._get_env_var(self.ENV_CONTEXT_MODEL_VAR)
            version_context = self._get_env_var(self.ENV_CONTEXT_VERSION_VAR)
            if model_context and version_context:
                self._context = Context(
                    model=model_context, version=version_context
                )
            else:
                raise RuntimeError(
                    "Must initialize MLTE context for session, either manually or through environment variables."
                )

        return self._context

    @property
    def stores(self) -> UnifiedStore:
        if self._stores is None:
            # If the stores have not been manually set, get URI from environment.
            stores_uri = self._get_env_var(self.ENV_STORE_URI_VAR)
            if stores_uri:
                self._set_stores(stores_uri)
            else:
                raise RuntimeError(
                    "Must initialize store URI, either manually or through environment variables."
                )

        assert self._stores is not None
        return self._stores

    @property
    def credentials(self) -> Credentials | None:
        if self._credentials is None:
            # If the stores have not been manually set, get URI from environment.
            user = self._get_env_var(self.ENV_CURRENT_USER_VAR)
            password = self._get_env_var(self.ENV_CURRENT_PASS_VAR)
            if user:
                self._credentials = Credentials(user, password)

        return self._credentials

    def _prepare_uri(self, store_uri: str) -> StoreURI:
        """Prepares the uri, returning a parsed version that has everything needed."""
        parsed_uri = StoreURI.from_string(store_uri)

        if parsed_uri.type == StoreType.REMOTE_HTTP:
            # For remote HTTP, if we have set proper credentials, inject them into the URI and/or override what was there.
            username = self.credentials.user if self.credentials else None
            password = self.credentials.password if self.credentials else None
            if username and password:
                # We only set this if we have both credentials. If either is missing, we just don't inject.
                updated_uri = url.set_url_username_password(
                    store_uri, username, password
                )
                parsed_uri = StoreURI.from_string(updated_uri)

        return parsed_uri

    def _set_context(self, context: Context) -> None:
        """Set the session context."""
        self._context = context

    def _set_stores(self, store_uri: str) -> None:
        """Set the session stores from the given URI."""
        parsed_uri = self._prepare_uri(store_uri)
        self._stores = setup_stores(parsed_uri)

    def _set_credentials(self, credentials: Credentials) -> None:
        """Set the session stores."""
        self._credentials = credentials

    def _get_env_var(self, env_var: str) -> str | None:
        """Get env var or return none if does not exist."""
        return os.environ.get(env_var, None)

    def create_context(self):
        """Creates the currently configured context in the currently configured session. Fails if either is not set. Does nothing if already created."""
        artifact_store = self.stores.artifact_store
        context = self.context
        artifact_store.session().create_parents(context.model, context.version)

ENV_CURRENT_PASS_VAR = 'MLTE_CURRENT_PASS' class-attribute instance-attribute

Environment variables to get model, version, store_uri and credentials from, if needed.

__init__()

Constructors, just resets all vars.

Source code in mlte/session/session.py

def __init__(self):
    """Constructors, just resets all vars."""
    self.reset()

create_context()

Creates the currently configured context in the currently configured session. Fails if either is not set. Does nothing if already created.

Source code in mlte/session/session.py

def create_context(self):
    """Creates the currently configured context in the currently configured session. Fails if either is not set. Does nothing if already created."""
    artifact_store = self.stores.artifact_store
    context = self.context
    artifact_store.session().create_parents(context.model, context.version)

reset()

Resets all internal state to defaults.

Source code in mlte/session/session.py

def reset(self):
    """Resets all internal state to defaults."""
    self._context: Context | None = None
    """The MLTE context for the session."""

    self._stores: UnifiedStore | None = None
    """All stores in this session."""

    self._credentials: Credentials | None = None
    """Current user and password for auditing and connections."""

add_catalog_store(catalog_store_uri, id)

Adds a global MLTE catalog store URI.

Parameters:

Name	Type	Description	Default
catalog_store_uri	str	The catalog store URI string	required

Source code in mlte/session/session.py

def add_catalog_store(catalog_store_uri: str, id: str):
    """
    Adds a global MLTE catalog store URI.
    :param catalog_store_uri: The catalog store URI string
    """
    g_session.stores.add_catalog_store_from_uri(
        StoreURI.from_string(catalog_store_uri), id
    )

get_session()

Return the package global session.

Source code in mlte/session/session.py

def get_session() -> Session:
    """Return the package global session."""
    return g_session

print_custom_list_entries(list_name)

Prints custom list entries in a user-friendly way.

Source code in mlte/session/session.py

def print_custom_list_entries(list_name: CustomListName) -> None:
    """Prints custom list entries in a user-friendly way."""
    entry_list = g_session.stores.custom_list_store.session().custom_list_entry_mapper.list_details(
        list_name
    )
    for entry in entry_list:
        print(str(entry))

reset_session()

Used to reset session if needed.

Source code in mlte/session/session.py

def reset_session() -> None:
    """Used to reset session if needed."""
    g_session.reset()

set_context(model_id, version_id, lazy=True)

Set the global MLTE context.

Parameters:

Name	Type	Description	Default
model_id	str	The model identifier	required
version_id	str	The version identifier	required
lazy	bool	Whether to wait to create the context until an artifact is written (True), or to eagerly create it immediately (False).	True

Source code in mlte/session/session.py

def set_context(model_id: str, version_id: str, lazy: bool = True):
    """
    Set the global MLTE context.
    :param model_id: The model identifier
    :param version_id: The version identifier
    :param lazy: Whether to wait to create the context until an artifact is written (True), or to eagerly create it immediately (False).
    """
    g_session._set_context(Context(model_id, version_id))
    if not lazy:
        g_session.create_context()

set_credentials(user, password=None)

Set the global MLTE credentials.

Parameters:

Name	Type	Description	Default
user	str	The user	required
password	str | None	The password (can be ommitted if only user is being set)	None

Source code in mlte/session/session.py

def set_credentials(user: str, password: str | None = None):
    """
    Set the global MLTE credentials.
    :param user: The user
    :param password: The password (can be ommitted if only user is being set)
    """
    g_session._set_credentials(Credentials(user, password))

set_store(store_uri)

Set the global MLTE stores from the given URI.

Parameters:

Name	Type	Description	Default
store_uri	str	The store URI string	required

Source code in mlte/session/session.py

def set_store(store_uri: str):
    """
    Set the global MLTE stores from the given URI.
    :param store_uri: The store URI string
    """
    g_session._set_stores(store_uri)