utils

Utility functions for OTEAPI DLite plugin.

RemoveItem

Singleton class used by update_dict() to indicate items that should be removed in the source dictionary.

Source code in oteapi_dlite/utils/utils.py

class RemoveItem:
    """Singleton class used by update_dict() to indicate items that should
    be removed in the source dictionary."""

TypeMismatchError

Bases: TypeError

Raised by update_dict() if there is a mismatch in value types between the dct and update dictionaries.

Source code in oteapi_dlite/utils/utils.py

class TypeMismatchError(TypeError):
    """Raised by update_dict() if there is a mismatch in value types
    between the `dct` and `update` dictionaries.
    """

add_settings(session, label, settings)

Store settings to the session.

Parameters:

Name	Type	Description	Default
session	Union[dict[str, Any], None]	An OTEAPI session object.	required
label	str	Label of settings to add.	required
settings	Union[dict, list, str, int, float, bool, NoneType]	A JSON-serialisable Python object with the settings to store.	required

Returns:

Type	Description
SessionUpdate	A SessionUpdate instance with the added settings.

Source code in oteapi_dlite/utils/utils.py

def add_settings(
    session: Union[dict[str, Any], None],
    label: str,
    settings: Union[  # noqa: PYI041
        dict, list, str, int, float, bool, NoneType
    ],
) -> SessionUpdate:
    """Store settings to the session.

    Arguments:
        session: An OTEAPI session object.
        label: Label of settings to add.
        settings: A JSON-serialisable Python object with the settings
            to store.

    Returns:
        A SessionUpdate instance with the added settings.
    """
    if session is None:
        session = {}

    # For now we store settings in the session.  This makes "settings"
    # independent of oteapi-dlite.
    d = session.get("settings", {})

    if label in d:
        raise KeyError(f"Setting with this label already exists: '{label}'")
    d[label] = json.dumps(settings)

    # Update the session with new settings
    session["settings"] = d

    return SessionUpdate(settings=d)

get_collection(session=None, collection_id=None)

Retrieve a DLite Collection.

Looks for a Collection UUID in the session. If none exists, a new, empty Collection is created and stored in the session.

If collection_id is provided, that id is used. If there already is a collection_id in the session, that is left untouched. Otherwise collection_id is added to the session.

Parameters:

Name	Type	Description	Default
session	Optional[dict[str, Any]]	An OTEAPI session object.	None
collection_id	Optional[str]	A specific collection ID to retrieve.	None

Return

A DLite Collection to be used throughout the OTEAPI session.

Source code in oteapi_dlite/utils/utils.py

def get_collection(
    session: Optional[dict[str, Any]] = None,
    collection_id: Optional[str] = None,
) -> dlite.Collection:
    """Retrieve a DLite Collection.

    Looks for a Collection UUID in the session.
    If none exists, a new, empty Collection is created and stored in the
    session.

    If `collection_id` is provided, that id is used. If there already is a
    `collection_id` in the session, that is left untouched. Otherwise
    `collection_id` is added to the session.

    Parameters:
        session: An OTEAPI session object.
        collection_id: A specific collection ID to retrieve.

    Return:
        A DLite Collection to be used throughout the OTEAPI session.
    """
    cache = DataCache()

    if session is None:
        session = {}

    id_ = collection_id or session.get("collection_id")

    # Storing the collection in the datacache is not scalable.
    # Do we really want to do that?
    #
    # Currently we check the datacache first and then ask dlite to look
    # up the collection (which is the proper and scalable solution).
    if id_ is None:
        coll = dlite.Collection()
        cache.add(coll.asjson(), key=coll.uuid)
    elif id_ in cache:
        coll = dlite.Instance.from_json(cache.get(id_), id=id_)
    else:
        try:
            coll = dlite.get_instance(id_)
        except dlite.DLiteError as exc:
            raise CollectionNotFound(
                f"Could not find DLite Collection with id {id_}"
            ) from exc

    if coll.meta.uri != dlite.COLLECTION_ENTITY:
        raise CollectionNotFound(f"instance with id {id_} is not a collection")

    if "collection_id" not in session:
        session["collection_id"] = coll.uuid

    return coll

get_driver(mediaType=None, accessService=None)

Return name of DLite driver for the given media type/access service.

Source code in oteapi_dlite/utils/utils.py

def get_driver(
    mediaType: Optional[str] = None,
    accessService: Optional[str] = None,
) -> str:
    """Return name of DLite driver for the given media type/access service."""
    if mediaType:
        if mediaType not in MEDIATYPES:
            raise ValueError("unknown DLite mediaType: {mediaType}")
        return MEDIATYPES[mediaType]

    if accessService:
        if accessService not in ACCESSSERVICES:
            raise ValueError("unknown DLite accessService: {accessService}")
        return ACCESSSERVICES[accessService]

    raise ValueError("either `mediaType` or `accessService` must be provided")

get_instance(meta, session=None, collection=None, routedict=None, instance_id=None, allow_incomplete=False, **kwargs)

Instantiates and returns an instance of meta.

Parameters:

Name	Type	Description	Default
meta	Union[str, Metadata]	Metadata to instantiate. Typically its URI.	required
session	Optional[dict[str, Any]]	An OTEAPI session object.	None
collection	Optional[Collection]	The collection with instances and mappings. The default is to get the collection from session.	None

Some less used optional arguments

routedict: Dict mapping property names to route number to select for the given property. The default is to select the route with lowest cost. instance_id: URI of instance to create. allow_incomplete: Whether to allow not populating all properties of the returned instance. kwargs: Additional arguments passed to dlite.mappings.instantiate().

Source code in oteapi_dlite/utils/utils.py

def get_instance(
    meta: Union[str, dlite.Metadata],
    session: Optional[dict[str, Any]] = None,
    collection: Optional[dlite.Collection] = None,
    routedict: Optional[dict] = None,
    instance_id: Optional[str] = None,
    allow_incomplete: bool = False,
    **kwargs,
) -> dlite.Instance:
    """Instantiates and returns an instance of `meta`.

    Arguments:
        meta: Metadata to instantiate.  Typically its URI.
        session: An OTEAPI session object.
        collection: The collection with instances and mappings.
            The default is to get the collection from `session`.

    Some less used optional arguments:
        routedict: Dict mapping property names to route number to select for
            the given property.  The default is to select the route with
            lowest cost.
        instance_id: URI of instance to create.
        allow_incomplete: Whether to allow not populating all properties
            of the returned instance.
        kwargs: Additional arguments passed to dlite.mappings.instantiate().
    """
    # Import here to avoid a hard dependency on tripper.
    from tripper import Triplestore

    if collection is None:
        if session is None:
            raise TypeError(
                "get_instance() requires that either `session` or "
                "`collection` argument is given."
            )
        collection = get_collection(session)

    if session:
        ts = get_triplestore(session)
    else:
        ts = Triplestore(backend="collection", collection=collection)

    return instantiate(
        meta=meta,
        instances=list(collection.get_instances()),
        triplestore=ts,
        routedict=routedict,
        id=instance_id,
        allow_incomplete=allow_incomplete,
        **kwargs,
    )

get_meta(uri)

Returns metadata corresponding to given uri.

This function may in the future be connected to a database.

Source code in oteapi_dlite/utils/utils.py

def get_meta(uri: str) -> dlite.Instance:
    """Returns metadata corresponding to given uri.

    This function may in the future be connected to a database.
    """
    meta = dlite.get_instance(uri)
    if not meta.is_meta:
        raise ValueError("uri {uri} does not correspond to metadata")
    return meta

get_settings(session, label)

Retrieve settings from the session.

Parameters:

Name	Type	Description	Default
session	Union[dict[str, Any], None]	An OTEAPI session object.	required
label	str	Label of settings to retrieve.	required

Returns:

Type	Description
Any	Python object with the settings or None if no settings exists
Any	with this label.

Source code in oteapi_dlite/utils/utils.py

def get_settings(session: Union[dict[str, Any], None], label: str) -> Any:
    """Retrieve settings from the session.

    Arguments:
        session: An OTEAPI session object.
        label: Label of settings to retrieve.

    Returns:
        Python object with the settings or None if no settings exists
        with this label.
    """
    if session is None:
        session = {}

    d = session.get("settings", {})
    return json.loads(d[label]) if label in d else None

get_triplestore(session)

Return a tripper.Triplestore instance for the current session.

If a 'tripper.triplestore' setting has been added with the SettingsStrategy, it will be used to configure the returned triplestore instance. Otherwise the session collection will be used.

Source code in oteapi_dlite/utils/utils.py

def get_triplestore(
    session: dict[str, Any],
) -> Triplestore:
    """Return a tripper.Triplestore instance for the current session.

    If a 'tripper.triplestore' setting has been added with the
    SettingsStrategy, it will be used to configure the returned
    triplestore instance.  Otherwise the session collection will be
    used.
    """
    # Import here to avoid a hard dependency on tripper.
    from tripper import Triplestore

    kb_settings = get_settings(session, "tripper.triplestore")
    if kb_settings:
        ts = Triplestore(**kb_settings)
    else:
        coll = get_collection(session)
        ts = Triplestore(backend="collection", collection=coll)
    return ts

update_collection(collection)

Update collection in DataCache.

Parameters:

Name	Type	Description	Default
collection	Collection	The DLite Collection to be updated.	required

Source code in oteapi_dlite/utils/utils.py

def update_collection(collection: dlite.Collection) -> None:
    """Update collection in DataCache.

    Parameters:
        collection: The DLite Collection to be updated.
    """
    cache = DataCache()
    cache.add(value=collection.asjson(), key=collection.uuid)

update_dict(dct, update)

Update dictionary dct using dictionary update.

This function differ from dict.update() in that it updates sub-directories recursively, instead of replacing them with the content of update.

If update has RemoveItem as a value, the corresponding item in dct will be removed.

Parameters:

Name	Type	Description	Default
dct	dict	Dict to update.	required
update	Optional[dict]	Dict used to update conf with.	required

Returns:

Type	Description
dict	The updated dict dct.

Raises:

Type	Description
TypeMismatchError	If there is a mismatch in value types between the dct and update dictionaries. Conversion between different number types is accepted.

Source code in oteapi_dlite/utils/utils.py

def update_dict(dct: dict, update: Optional[dict]) -> dict:
    """Update dictionary `dct` using dictionary `update`.

    This function differ from `dict.update()` in that it updates
    sub-directories recursively, instead of replacing them with the
    content of `update`.

    If `update` has `RemoveItem` as a value, the corresponding item
    in `dct` will be removed.

    Arguments:
        dct: Dict to update.
        update: Dict used to update `conf` with.

    Returns:
        The updated dict `dct`.

    Raises:
        TypeMismatchError: If there is a mismatch in value types
            between the `dct` and `update` dictionaries.  Conversion
            between different number types is accepted.

    """
    if not update:
        return dct

    for k, v in dct.items():
        if k in update:

            if update[k] is RemoveItem:
                del dct[k]
                continue

            if not (
                (isinstance(update[k], Number) and isinstance(v, Number))
                or isinstance(update[k], type(v))
            ):
                raise TypeMismatchError(
                    f"type of `update['{k}']` ({type(update[k])!r}) is not a "
                    f"subclass of the type of `dct['{k}']` ({type(v)!r})"
                )

            if isinstance(v, dict):
                update_dict(v, update[k])
            else:
                dct[k] = update[k]

    # Add new items to `dct`
    for k, v in update.items():
        if k not in dct:
            dct[k] = v

    return dct