Tools for accessing CMS Metadata
# coding: utf-8

import sys
import os


thisdir = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.join(thisdir, "_extensions"))
sys.path.insert(0, os.path.dirname(thisdir))

import order as od


project = od.__name__
author = od.__author__
copyright = od.__copyright__
copyright = copyright[10:] if copyright.startswith("Copyright ") else copyright
copyright = copyright.split(",", 1)[0]
version = od.__version__[:od.__version__.index(".", 2)]
release = od.__version__
language = "en"

templates_path = ["_templates"]
html_static_path = ["_static"]
master_doc = "index"
source_suffix = ".rst"
exclude_patterns = []
pygments_style = "sphinx"
add_module_names = False

html_title = "{} v{}".format(project, version)
html_logo = "../assets/logo240.png"
html_favicon = "../assets/favicon.ico"
html_theme = "sphinx_book_theme"
html_theme_options = {
    "logo_only": True,
    "home_page_in_toc": True,
    "show_navbar_depth": 2,
    "show_toc_level": 2,
    "repository_url": "https://github.com/cms-cat/order",
    "use_repository_button": True,
    "use_issues_button": True,
    "use_edit_page_button": True,
}

extensions = [
    "sphinx.ext.autodoc",
    "sphinx.ext.intersphinx",
    "sphinx.ext.viewcode",
    "sphinx.ext.autosectionlabel",
    "autodocsumm",
    "myst_parser",
    "sphinx_lfs_content",
]

autodoc_member_order = "bysource"


def setup(app):
    app.add_css_file("styles_sphinx_book_theme.css")

docs/conf.py
cms-cat/order
# coding: utf-8
# flake8: noqa

"""
Pythonic class collection to structure and access CMS metadata.
"""


__all__ = [
    "Settings",
    "Lazy", "Model",
    "AdapterModel", "Adapter", "Materialized", "DataProvider",
    "UniqueObject", "LazyUniqueObject", "UniqueObjectIndex",
    "DuplicateObjectException", "DuplicateNameException", "DuplicateIdException",
    "GenOrder", "DatasetInfo", "Dataset", "LazyDataset", "DatasetIndex",
    "Campaign",
]


# package infos
from order.__meta__ import (
    __doc__, __author__, __email__, __copyright__, __credits__, __contact__, __license__,
    __status__, __version__,
)

# provisioning imports
from order.settings import Settings
from order.types import Lazy
from order.adapters.base import AdapterModel, Adapter, Materialized, DataProvider
from order.models.base import Model
from order.models.unique import (
    UniqueObject, LazyUniqueObject, UniqueObjectIndex, DuplicateObjectException,
    DuplicateNameException, DuplicateIdException,
)
from order.models.dataset import GenOrder, DatasetInfo, Dataset, LazyDataset, DatasetIndex
from order.models.campaign import Campaign

# import adapters to trigger their registration
import order.adapters.order
import order.adapters.das
# import order.adapters.dbs
# import order.adapters.xsdb

order/__init__.py
cms-cat/order
# coding: utf-8

"""
Pythonic class collection to structure and access CMS metadata.
"""

__author__ = "CMS Collaboration"
__email__ = "marcel.rieger@cern.ch"
__copyright__ = "Copyright 2023, CMS Collaboration"
__credits__ = ["Marcel Rieger", "Davide Valsecchi"]
__contact__ = "https://github.com/cms-cat/order"
__license__ = "BSD-3-Clause"
__status__ = "Development"
__version__ = "3.0.0"

order/__meta__.py
cms-cat/order
# coding: utf-8

"""
Base class definitions for adapters and the overarching data provider for cached retrieval.
"""

from __future__ import annotations


__all__ = ["AdapterModel", "Adapter", "Materialized", "DataProvider"]


import os
import re
import json
import shutil
from contextlib import contextmanager
from abc import ABCMeta, abstractmethod, abstractproperty

from pydantic import BaseModel

from order.types import Any, Sequence, Dict, GeneratorType, NonEmptyStrictStr, StrictStr, Field
from order.settings import Settings
from order.util import create_hash


class AdapterModel(BaseModel):

    adapter: NonEmptyStrictStr
    key: StrictStr
    arguments: Dict[NonEmptyStrictStr, Any] = Field(default_factory=dict)

    def compare_signature(self, other: "AdapterModel") -> bool:
        return (
            isinstance(other, AdapterModel) and
            other.adapter == self.adapter and
            other.arguments == self.arguments
        )


class AdapterMeta(ABCMeta):
    """
    Metaclass for :py:class:`Adapter` classes providing fast subclass lookup and collision handling.
    """

    adapters: dict[str, "AdapterMeta"] = {}

    def __new__(
        meta_cls,
        class_name: str,
        bases: tuple[type],
        class_dict: dict[str, Any],
    ) -> "AdapterMeta":
        cls = type.__new__(meta_cls, class_name, bases, class_dict)

        # check if the class was registered previously
        if meta_cls.has_cls(cls.name):
            raise ValueError(
                f"cannot register adapter {cls}, name '{cls.name}' previously used by "
                f"{meta_cls.adapters[cls.name]}",
            )

        # store class by name
        if getattr(cls, "_name", "") != "base":
            meta_cls.adapters[cls.name] = cls

        return cls

    @classmethod
    def has_cls(cls, name: str) -> bool:
        return name in cls.adapters

    @classmethod
    def get_cls(cls, name: str) -> "AdapterMeta":
        if not cls.has_cls(name):
            raise KeyError(f"unknown adapter '{name}'")
        return cls.adapters[name]


class Materialized(dict):
    """
    Container for materialized values returned by :py:meth:`Adapter.retrieve_data`.
    """


class Adapter(object, metaclass=AdapterMeta):
    """
    Abstract base class for all adapters.
    """

    # temporary attribute used during class registration
    _name = "base"

    # whether the retrieve_data method needs the data_location as its first positional argument
    needs_data_location = False

    @abstractproperty
    def name(self) -> str:
        # must be implemented by subclasses
        return ""

    @abstractmethod
    def retrieve_data(self) -> Materialized:
        # must be implemented by subclasses
        return

    def get_cache_key(self, **kwargs) -> tuple:
        return tuple(
            (
                key,
                self.get_cache_key(**kwargs[key]) if isinstance(kwargs[key], dict) else kwargs[key],
            )
            for key in sorted(kwargs)
        )

    @classmethod
    def location_is_local(cls, data_location: str) -> bool:
        return data_location.startswith("file://")

    @classmethod
    def location_is_remote(cls, data_location: str) -> bool:
        return data_location.startswith(("http://", "https://"))

    @classmethod
    def remove_scheme(cls, data_location: str) -> str:
        return re.sub(r"^(\w+)\:\/\/", "", data_location)


# remove temporary _name attribute
del Adapter._name


class DataProvider(object):
    """
    Interface between data locations plus caches and :py:class:`Adapter` instances.
    """

    __instance = None

    class SkipCaching(Exception):
        """
        Special exception type that can be thrown within the :py:meth:`DataProvider.get_data`
        context to instruct it to skip caching.
        """

    @classmethod
    def instance(cls) -> "DataProvider":
        """
        Singleton constructor and getter.
        """
        if cls.__instance is None:
            settings = Settings.instance()
            cls.__instance = cls(
                data_location=settings.data_location,
                cache_directory=settings.cache_directory,
                readonly_cache_directories=settings.readonly_cache_directories,
                clear_cache=settings.clear_cache,
            )

        return cls.__instance

    def __init__(
        self,
        *,
        data_location: str,
        cache_directory: str,
        readonly_cache_directories: Sequence[str] = (),
        clear_cache: bool = False,
    ) -> None:
        super().__init__()

        # expansion helper
        expand = lambda path: os.path.expandvars(os.path.expanduser(str(path)))

        # store attributes
        self.data_location: str = expand(data_location)
        self.cache_directory: str = expand(cache_directory)
        self.readonly_cache_directories: list[str] = list(map(expand, readonly_cache_directories))

        # cache_directory must not be in read_cache
        if self.cache_directory in self.readonly_cache_directories:
            raise ValueError(
                f"cache_directory '{self.cache_directory}' must not be in "
                "readonly_cache_directories",
            )

        # clear the cache initially
        if clear_cache and os.path.exists(self.cache_directory):
            shutil.rmtree(self.cache_directory)

    @contextmanager
    def materialize(
        self,
        adapter_model: AdapterModel | dict[str, Any],
        adapter_kwargs: dict[str, Any] | None = None,
    ) -> GeneratorType:
        if not isinstance(adapter_model, AdapterModel):
            adapter_model = AdapterModel(**adapter_model)

        # get the adapter class and instantiate it
        adapter = AdapterMeta.get_cls(adapter_model.adapter)()

        # merge kwargs
        adapter_kwargs = {**adapter_model.arguments, **(adapter_kwargs or {})}

        # determine the basename of the cache file (if existing)
        h = (
            os.path.realpath(self.data_location),
            adapter.get_cache_key(**adapter_kwargs),
        )
        cache_name = f"{create_hash(h)}.json"

        # when cached, read the cached object instead
        readable_path, writable_path, cached = self.check_cache(cache_name)
        if cached:
            yield self.read_cache(readable_path)
            return

        # in cache-only mode, this point should not be reached
        if Settings.instance().cache_only:
            raise Exception(f"adapter '{adapter.name}' cannot be evaluated in cache-only mode")

        # invoke the adapter
        args = (self.data_location,) if adapter.needs_data_location else ()
        materialized = adapter.retrieve_data(*args, **adapter_kwargs)

        # complain when the return value is not a materialized container
        if not isinstance(materialized, Materialized):
            raise TypeError(
                f"retrieve_data of adapter '{adapter_model.adapter}' must return a Materialized "
                f"instance, but got '{materialized}'",
            )

        # yield the materialized data and cache it if the receiving context did not raise
        try:
            yield materialized
        except self.SkipCaching:
            return

        # cache it
        if writable_path:
            self.write_cache(writable_path, materialized)

    def check_cache(self, cache_name: str) -> [str, str, bool]:
        # check the writable (default) cache directory
        writable_path = os.path.join(self.cache_directory, cache_name)
        if os.path.exists(writable_path):
            return writable_path, writable_path, True

        for readonly_cache_directory in self.readonly_cache_directories:
            readable_path = os.path.join(readonly_cache_directory, cache_name)
            if os.path.exists(readable_path):
                return readable_path, writable_path, True

        return writable_path, writable_path, False

    def write_cache(self, path: str, materialized: Materialized) -> None:
        dirname = os.path.dirname(path)
        if not os.path.exists(dirname):
            os.makedirs(dirname)

        with open(path, "w") as f:
            json.dump(materialized, f)

    def read_cache(self, path: str) -> Materialized:
        with open(path, "r") as f:
            return Materialized(json.load(f))

order/adapters/base.py
cms-cat/order
# coding: utf-8

from __future__ import annotations


__all__ = ["DASDatasetAdapter"]


from order.adapters.base import Adapter, Materialized


class DASDatasetAdapter(Adapter):

    name = "das_dataset"

    def retrieve_data(self, *, keys: list[str]) -> Materialized:
        if keys[0].startswith("/SCALE"):
            return Materialized(n_events=1, n_files=1)
        return Materialized(n_events=5_000_000, n_files=12)


class DASLFNsAdapter(Adapter):

    name = "das_lfns"

    def retrieve_data(self, *, keys: list[str]) -> Materialized:
        if keys[0].startswith("/SCALE"):
            return Materialized(lfns=["/SCALE/b/NANOAODSIM"])
        return Materialized(lfns=["/a/b/NANOAODSIM"])

order/adapters/das.py
cms-cat/order
# coding: utf-8

from __future__ import annotations


__all__ = ["DBSFilesAdapter"]


import requests

from order.settings import Settings
from order.adapters.base import Adapter, Materialized


class DBSFilesAdapter(Adapter):

    name = "dbs_files"

    def retrieve_data(self, *, dataset_key: str, dbs_instance: str = "prod/global") -> Materialized:
        resource = f"https://cmsweb.cern.ch:8443/dbs/{dbs_instance}/DBSReader/files?dataset={dataset_key}&detail=True"  # noqa
        r = requests.get(
            resource,
            cert=Settings.instance().user_proxy,
            verify=False,
        )
        return Materialized(data=r.json())

order/adapters/dbs.py
cms-cat/order
# coding: utf-8

"""
Order-internal adapters, mainly used to avoid redundancies inside order-data.
"""

from __future__ import annotations


__all__ = ["OrderAdapter"]


import os
import glob

import yaml

from order.adapters.base import Adapter, Materialized


class OrderAdapter(Adapter):

    # order adapters need to DataProvider's data_location in retrieve_data
    needs_data_location = True


class DatasetsAdapter(OrderAdapter):

    name = "order_datasets"

    def retrieve_data(
        self,
        data_location: str,
        *,
        campaign_name: str,
    ) -> Materialized:
        # only supporting local evaluation for now
        if not self.location_is_local(data_location):
            raise NotImplementedError(f"non-local location {data_location} not handled by {self}")

        # build the directory in which to look for dataset files
        dataset_dir = os.path.join(self.remove_scheme(data_location), "datasets", campaign_name)

        # read yaml files in the datasets directory
        datasets = []
        for path in glob.glob(os.path.join(dataset_dir, "*.yaml")):
            with open(path, "r") as f:
                # allow multiple documents per file
                stream = yaml.load_all(f, Loader=yaml.SafeLoader)
                for i, entry in enumerate(stream):
                    if "name" not in entry:
                        raise AttributeError(
                            f"no field 'name' defined in enty {i} of dataset yaml file {path}",
                        )
                    if "id" not in entry:
                        raise AttributeError(
                            f"no field 'id' defined in enty {i} of dataset yaml file {path}",
                        )
                    datasets.append(
                        self.create_lazy_dataset_dict(campaign_name, entry["name"], entry["id"]),
                    )

        return Materialized(datasets=datasets)

    @classmethod
    def create_lazy_dataset_dict(cls, campaign_name: str, name: str, id: int) -> dict:
        return {
            "name": name,
            "id": id,
            "class_name": "Dataset",
            "adapter": {
                "adapter": "order_dataset",
                "arguments": {
                    "campaign_name": campaign_name,
                    "dataset_name": name,
                },
                "key": "dataset",
            },
        }


class DatasetAdapter(OrderAdapter):

    name = "order_dataset"

    def retrieve_data(
        self,
        data_location: str,
        *,
        campaign_name: str,
        dataset_name: str,
    ) -> Materialized:
        # only supporting local evaluation for now
        if not self.location_is_local(data_location):
            raise NotImplementedError(f"non-local location {data_location} not handled by {self}")

        # build the yaml file path
        path = os.path.join(
            self.remove_scheme(data_location),
            "datasets",
            campaign_name,
            f"{dataset_name}.yaml",
        )
        if not os.path.exists(path):
            raise Exception(f"dataset file {path} not existing")

        # open the file and look for the dataset
        with open(path, "r") as f:
            stream = yaml.load_all(f, Loader=yaml.SafeLoader)
            for entry in stream:
                if entry.get("name") == dataset_name:
                    return Materialized(dataset=entry)

        raise Exception(f"no dataset entry with name '{dataset_name}' found in {path}")

order/adapters/order.py
cms-cat/order
# coding: utf-8

"""
Cross section data base (XSDB) adapters.
"""

from __future__ import annotations


__all__ = ["XSDBAdapter"]


from order.adapters.base import Adapter, Materialized


class XSDBAdapter(Adapter):

    name = "xsdb"

    def retrieve_data(self, *, foo: int) -> Materialized:
        return Materialized(ecm=foo * 2, data_tier="nano")

order/adapters/xsdb.py
cms-cat/order
# coding: utf-8

"""
Custom base models and types for lazy evaluation.
"""

from __future__ import annotations


__all__ = ["Model"]


from pydantic import BaseModel, ConfigDict

from order.types import Any, GeneratorType, Field, FieldInfo, Lazy
from order.adapters.base import AdapterModel, DataProvider
from order.util import no_value, colored


class ModelMeta(type(BaseModel)):

    def __new__(meta_cls, class_name: str, bases: tuple, class_dict: dict[str, Any]) -> "ModelMeta":
        # convert "Lazy" annotations to proper fields and add access properties
        lazy_attrs = []
        for attr, type_str in list(class_dict.get("__annotations__", {}).items()):
            type_names = Lazy.parse_annotation(type_str)
            if type_names:
                meta_cls.register_lazy_attr(attr, type_names, class_name, bases, class_dict)
                lazy_attrs.append(attr)

        # store names of lazy attributes, considering also bases
        lazy_attrs_dict = {}
        for base in reversed(bases):
            if getattr(base, "_lazy_attrs", None) is None:
                continue
            lazy_attrs_dict.update({
                attr: lazy_attr
                for attr, lazy_attr in base._lazy_attrs.default.items()
                if lazy_attr in base.__fields__
            })
        lazy_attrs_dict.update({attr: meta_cls.get_lazy_attr(attr) for attr in lazy_attrs})
        class_dict["_lazy_attrs"] = lazy_attrs_dict

        # check the model_config
        class_dict["model_config"] = model_config = class_dict.get("model_config") or ConfigDict()
        if not isinstance(model_config, dict):
            raise TypeError(
                "class attribute 'model_config' should be empty or a ConfigDict, "
                f"but got {model_config}",
            )

        # enable default value validation
        model_config["validate_default"] = True

        # enable assignment validation
        model_config["validate_assignment"] = True

        # create the class
        cls = super().__new__(meta_cls, class_name, bases, class_dict)

        # remove non-existing lazy attributes from above added dict after class was created
        for attr, lazy_attr in list(cls._lazy_attrs.default.items()):
            if lazy_attr not in cls.__fields__:
                del cls._lazy_attrs.default[attr]

        return cls

    @classmethod
    def get_lazy_attr(meta_cls, attr: str) -> str:
        return f"lazy_{attr}"

    @classmethod
    def register_lazy_attr(
        meta_cls,
        attr: str,
        type_names: list[str],
        class_name: str,
        bases: tuple,
        class_dict: dict[str, Any],
    ) -> None:
        # if a field already exist, get it
        field = class_dict.get(attr)
        if field is not None and not isinstance(field, FieldInfo):
            raise TypeError(
                f"class attribute '{attr}' should be empty or a Field, but got {field}",
            )
        class_dict.pop(attr, None)

        # store existing fields
        class_dict.setdefault("__orig_fields__", {})[attr] = field

        # exchange the annotation with the lazy one
        lazy_attr = meta_cls.get_lazy_attr(attr)
        class_dict["__annotations__"][lazy_attr] = class_dict["__annotations__"].pop(attr)

        # make sure the field has an alias set and is skipped in repr
        _field = Field(alias=attr, repr=False)
        field = FieldInfo.merge_field_infos(field, _field) if field else _field
        class_dict[lazy_attr] = field

        # add a property for the original attribute
        def fget(self):
            value = getattr(self, lazy_attr)

            # when the value is (already) materialized, just return it
            if not isinstance(value, AdapterModel):
                return value

            # at this point, we must materialize the value through the adapter
            # and assign all resulting lazy attributes
            adapter_model = value
            value = no_value
            with DataProvider.instance().materialize(adapter_model) as materialized:
                # loop through known lazy attributes and check which of them is assigned a
                # materialized value
                for attr_, lazy_attr_ in self._lazy_attrs.items():
                    # the adapter model must be compatible that the called one
                    adapter_model_ = getattr(self, lazy_attr_)
                    if not adapter_model.compare_signature(adapter_model_):
                        continue

                    # complain when the adapter did not provide a value for this attribute
                    if adapter_model_.key not in materialized:
                        raise KeyError(
                            f"adapter '{adapter_model.adapter}' did not provide field "
                            f"'{adapter_model_.key}' as required by attribute '{attr_}'",
                        )

                    # set the value
                    setattr(self, lazy_attr_, materialized[adapter_model_.key])

                    # assign it to the return value for the requested attribute
                    if attr_ == attr:
                        value = getattr(self, lazy_attr_)

            # complain if the return value was not set
            if value == no_value:
                raise RuntimeError(
                    f"adapter referred to by '{adapter_model}' did not materialize value "
                    f"for field '{attr}'",
                )

            return value

        # we need a valid type hint for the setter so create the function dynamically
        fset_type = f"Lazy[{', '.join(type_names)}]"
        locals_ = {}
        exec(
            f"def fset(self, value: {fset_type}) -> None: setattr(self, '{lazy_attr}', value)",
            globals(),
            locals_,
        )
        fset = locals_["fset"]

        class_dict[attr] = property(fget=fget, fset=fset)


class Model(BaseModel, metaclass=ModelMeta):
    """
    Base model for all order entities.
    """

    def __repr_name__(self) -> str:
        return colored(super().__repr_name__(), color="light_green")

    def __repr_args__(self) -> GeneratorType:
        """
        Yields all key-values pairs to be injected into the representation.
        """
        for attr, value in super().__repr_args__():
            yield colored(attr, color="light_blue"), value

        for attr, lazy_attr in self._lazy_attrs.items():
            # skip when field was originally skipped
            orig_field = self.__orig_fields__.get(attr)
            if orig_field and not orig_field.repr:
                continue

            value = getattr(self, lazy_attr)
            yield (
                colored(attr, color="light_blue"),
                f"lazy({value.adapter})" if isinstance(value, AdapterModel) else value,
            )

order/models/base.py
cms-cat/order
# coding: utf-8

"""
Campaign definition.
"""

from __future__ import annotations


__all__ = ["Campaign"]


from order.types import Lazy, NonEmptyStrictStr, StrictFloat, Field
from order.models.unique import UniqueObject
from order.models.dataset import DatasetIndex


class Campaign(UniqueObject):

    tier: Lazy[NonEmptyStrictStr]
    ecm: Lazy[StrictFloat]
    recommended_global_tag: NonEmptyStrictStr
    datasets: DatasetIndex = Field(default_factory=DatasetIndex)

order/models/campaign.py
cms-cat/order
# coding: utf-8

from __future__ import annotations


__all__ = ["GenOrder", "DatasetInfo", "Dataset", "LazyDataset", "DatasetIndex"]


import enum

from pydantic import field_validator

from order.types import Union, List, Dict, Field, NonEmptyStrictStr, PositiveStrictInt, Lazy
from order.models.base import Model
from order.models.unique import UniqueObject, LazyUniqueObject, UniqueObjectIndex


# class File(Model):
#     logical_file_name: str
#     block_name: str
#     check_sum: int
#     last_modification_date: int
#     file_type: str


class GenOrder(enum.Enum):

    unknown: str = "unknown"
    lo: str = "lo"
    nlo: str = "nlo"
    nnlo: str = "nnlo"
    n3lo: str = "n3lo"

    def __str__(self) -> str:
        return self.value


class DatasetInfo(Model):

    keys: List[NonEmptyStrictStr]
    gen_order: NonEmptyStrictStr = Field(default=str(GenOrder.unknown))
    n_files: Lazy[PositiveStrictInt]
    n_events: Lazy[PositiveStrictInt]
    lfns: Lazy[List[NonEmptyStrictStr]]

    @field_validator("gen_order", mode="after")
    @classmethod
    def validate_gen_order(cls, gen_order: str) -> str:
        try:
            return str(GenOrder[gen_order])
        except KeyError:
            raise ValueError(f"unknown gen_order '{gen_order}'")


class Dataset(UniqueObject):

    info: Dict[str, DatasetInfo]

    def __getitem__(self, name: str) -> DatasetInfo:
        return self.get_info(name)

    def get_info(self, name: str) -> DatasetInfo:
        return self.info[name]

    def set_info(self, name: str, info: DatasetInfo) -> None:
        if not isinstance(info, DatasetInfo):
            raise TypeError(f"expected info to be DatasetInfo object, but got '{info}'")
        self.info[name] = info

    @property
    def keys(self) -> list[NonEmptyStrictStr]:
        return self.info["nominal"].keys

    @property
    def gen_order(self) -> GenOrder:
        return self.info["nominal"].gen_order

    @property
    def n_files(self) -> int:
        return self.info["nominal"].n_files

    @property
    def n_events(self) -> int:
        return self.info["nominal"].n_events

    @property
    def lfns(self) -> list[NonEmptyStrictStr]:
        return self.info["nominal"].lfns


class LazyDataset(LazyUniqueObject):

    class_name: NonEmptyStrictStr = Field(default=Dataset)


class DatasetIndex(UniqueObjectIndex):

    class_name: NonEmptyStrictStr = Field(default=Dataset)
    objects: Lazy[List[Union[Dataset, LazyDataset]]] = Field(default_factory=list, repr=False)

order/models/dataset.py
cms-cat/order
# coding: utf-8

"""
Models definitions of unique objects and an index container to access them.
"""

from __future__ import annotations


__all__ = [
    "UniqueObject", "LazyUniqueObject", "UniqueObjectIndex",
    "DuplicateObjectException", "DuplicateNameException", "DuplicateIdException",
]


from contextlib import contextmanager

from pydantic import field_validator

from order.types import (
    ClassVar, Any, T, List, Union, GeneratorType, Field, PositiveStrictInt, NonEmptyStrictStr,
    KeysView, Lazy,
)
from order.models.base import Model
from order.adapters.base import AdapterModel, DataProvider
from order.util import no_value, DotAccessProxy


class UniqueObjectMeta(type(Model)):
    """
    Meta class definition for :py:class:`UniqueObject` that ammends the class dict for newly created
    classes.
    """

    # registered classes mapped to their names
    __unique_classes = {}

    def __new__(
        meta_cls,
        class_name: str,
        bases: tuple,
        class_dict: dict[str, Any],
    ) -> "UniqueObjectMeta":
        # define a separate integer to remember the maximum id
        class_dict.setdefault("_max_id", 0)
        class_dict.setdefault("__annotations__", {})["_max_id"] = "ClassVar[int]"

        # create the class
        cls = super().__new__(meta_cls, class_name, bases, class_dict)

        # remember it
        meta_cls.__unique_classes[class_name] = cls

        return cls

    @classmethod
    def has_unique_cls(meta_cls, name: str) -> bool:
        """
        Returns whether a class named *name* was registered previously.
        """
        return name in meta_cls.__unique_classes

    @classmethod
    def get_unique_cls(meta_cls, name: str) -> "UniqueObjectMeta":
        """
        Returns a previously created class named *name*.
        """
        return meta_cls.__unique_classes[name]


class UniqueObjectBase(Model):

    id: PositiveStrictInt
    name: NonEmptyStrictStr

    def __hash__(self) -> int:
        """
        Returns the unique hash of the unique object.
        """
        return hash((hex(id(self)), self.__repr__()))

    def __eq__(self, other: Any) -> bool:
        """
        Compares *other* to this instance. When *other* is a string (integer), the comparison is
        *True* when it matches the *name* (*id*) if this instance. When *other* is a unique object
        as well, the comparison is *True* when *__class__*, *name* and *id* match. All other cases
        evaluate to *False*.
        """
        if isinstance(other, str):
            return self.name == other

        if isinstance(other, int):
            return self.id == other

        if isinstance(other, self.__class__):
            return self.name == other.name and self.id == other.id

        # TODO: not particularly clean to use a subclass of _this_ class, solve by inheritance
        if (
            (isinstance(other, LazyUniqueObject) and other.cls == self.__class__) or
            (isinstance(self, LazyUniqueObject) and self.cls == other.__class__)
        ):
            return self.name == other.name and self.id == other.id

        return False

    def __ne__(self, other: Any) -> bool:
        """
        Opposite of :py:meth:`__eq__`.
        """
        return not self.__eq__(other)

    def __lt__(self, other: Any) -> bool:
        """
        Returns *True* when the id of this instance is lower than an *other* one. *other* can either
        be an integer or a unique object of the same class.
        """
        if isinstance(other, int):
            return self.id < other

        if isinstance(other, UniqueObjectBase):
            return self.id < other.id

        return False

    def __le__(self, other: Any) -> bool:
        """
        Returns *True* when the id of this instance is lower than or equal to an *other* one.
        *other* can either be an integer or a unique object of the same class.
        """
        if isinstance(other, int):
            return self.id <= other

        if isinstance(other, UniqueObjectBase):
            return self.id <= other.id

        return False

    def __gt__(self, other: Any) -> bool:
        """
        Returns *True* when the id of this instance is greater than an *other* one. *other* can
        either be an integer or a unique object of the same class.
        """
        if isinstance(other, int):
            return self.id > other

        if isinstance(other, UniqueObjectBase):
            return self.id > other.id

        return False

    def __ge__(self, other: Any) -> bool:
        """
        Returns *True* when the id of this instance is greater than or qual to an *other* one.
        *other* can either be an integer or a unique object of the same class.
        """
        if isinstance(other, int):
            return self.id >= other

        if isinstance(other, UniqueObjectBase):
            return self.id >= other.id

        return False


class WrapsUniqueClcass(Model):

    class_name: NonEmptyStrictStr

    @field_validator("class_name", mode="before")
    @classmethod
    def convert_class_to_name(cls, class_name: str | UniqueObjectMeta) -> str:
        if isinstance(class_name, UniqueObjectMeta):
            class_name = class_name.__name__
        return class_name

    @field_validator("class_name", mode="after")
    @classmethod
    def validate_class_name(cls, class_name: str) -> str:
        # check that the model class is existing
        if not UniqueObjectMeta.has_unique_cls(class_name):
            raise ValueError(f"class '{class_name}' is not a subclass of UniqueObject")
        return class_name

    def __init__(self, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)

        # store a reference to the wrapped class
        self._cls = UniqueObjectMeta.get_unique_cls(self.class_name)

    @property
    def cls(self) -> UniqueObjectMeta:
        return self._cls


class UniqueObject(UniqueObjectBase, metaclass=UniqueObjectMeta):

    AUTO_ID: ClassVar[str] = "+"

    @field_validator("id", mode="before")
    @classmethod
    def evaluate_auto_id(cls, id: str | int) -> int:
        if id == cls.AUTO_ID:
            cls._max_id += 1
            id = cls._max_id
        return id

    def __init__(self, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)

        # adjust max id class attribute
        if self.id > self.__class__._max_id:
            self.__class__._max_id = self.id


class LazyUniqueObject(UniqueObjectBase, WrapsUniqueClcass):

    adapter: AdapterModel

    @contextmanager
    def materialize(self, index: "UniqueObjectIndex") -> GeneratorType:
        with DataProvider.instance().materialize(self.adapter) as materialized:
            # complain when the adapter did not provide a value for this attribute
            if self.adapter.key not in materialized:
                raise KeyError(
                    f"adapter '{self.adapter.name}' did not provide field "
                    f"'{self.adapter.key}' required to materialize '{self}'",
                )

            # create the materialized instance
            inst = self.cls(**materialized[self.adapter.key])

            yield inst


class UniqueObjectIndex(WrapsUniqueClcass):
    """
    Index of :py:class:`UniqueObject` instances which are - as the name suggests - unique within
    this index, enabling fast lookups by either name or id.

    **Example**

    .. code-block:: python

        idx = UniqueObjectIndex(class_name="UniqueObject")
        foo = idx.add(UniqueObject(name="foo", id=1))
        bar = idx.add(UniqueObject(name="bar", id=2))

        len(idx)
        # -> 2

        idx.get(1) == foo
        # -> True

        idx.add(UniqueObject(name="foo", id=3))
        # -> DuplicateNameException

        idx.add(UniqueObject(name="test", id=1))
        # -> DuplicateIdException

        idx.names()
        # -> ["foo", "bar"]

        idx.ids()
        # -> [1, 2]

    **Members**

    .. py:attribute:: cls

        type: :py:class:`UniqueObjectMeta` (read-only)

        Class of objects hold by this index.

    .. py:attribute:: n

        type: :py:class:`DotAccessProxy` (read-only)

        An object that provides simple attribute access to contained objects via name.
    """

    objects: Lazy[List[Union[UniqueObject, LazyUniqueObject]]] = Field(
        default_factory=list,
        repr=False,
    )

    @field_validator("lazy_objects", mode="after")
    @classmethod
    def detect_duplicate_objects(
        cls,
        objects: Lazy[list[UniqueObject | LazyUniqueObject]],
    ) -> Lazy[list[UniqueObject | LazyUniqueObject]]:
        # skip adapters
        if isinstance(objects, AdapterModel):
            return objects

        # detect duplicate ids and names
        seen_names, seen_ids = set(), set()
        for obj in objects:
            if obj.name in seen_names:
                raise DuplicateNameException(type(obj), obj.name, cls)
            if obj.id in seen_ids:
                raise DuplicateIdException(type(obj), obj.id, cls)
            seen_ids.add(obj.id)
            seen_names.add(obj.name)

        return objects

    def __init__(self, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)

        # name-based DotAccessProxy
        self._n = DotAccessProxy(self.get)

        # hashmap indices for faster lookups
        self._name_index = {}
        self._id_index = {}

        # sync indices initially
        self._sync_indices()

    def __len__(self) -> int:
        """
        Returns the number of objects in the index.
        """
        return len(self.objects)

    def __contains__(self, obj: Any) -> None:
        """
        Checks if an object is contained in the index. :py:meth:`has` is used internally.
        """
        return self.has(obj)

    def __iter__(self) -> GeneratorType:
        """
        Iterates through the index and yields the contained objects (i.e. the *values*).
        """
        for obj in self.objects:
            yield self.get(obj.name)

    def __nonzero__(self) -> bool:
        """
        Boolish conversion that depends on the number of objects in the index.
        """
        return bool(self.objects)

    def __getitem__(self, obj: Any) -> UniqueObject:
        """
        Shorthand for :py:func:`get` without a default value.
        """
        return self.get(obj)

    def __repr_args__(self) -> GeneratorType:
        """
        Yields all key-values pairs to be injected into the representation.
        """
        yield from super().__repr_args__()
        yield "len", len(self)

    @property
    def n(self) -> DotAccessProxy:
        return self._n

    def _sync_indices(self, force: bool = False) -> None:
        """
        Synchronizes the two hashmap indices with the actual objects. Unless *force* is *True*, the
        synchronization stops in case the index lengths match the number of objects.
        """
        # when not forcing and all lengths are all identical, do nothing
        if not force and (len(self.objects) == len(self._name_index) == len(self._id_index)):
            return

        # clear indices
        self._name_index.clear()
        self._id_index.clear()

        # populate them
        for obj in self.objects:
            self._name_index[obj.name] = obj
            self._id_index[obj.id] = obj

    def names(self) -> KeysView:
        """
        Returns the names of the contained objects in the index.
        """
        self._sync_indices()
        return self._name_index.keys()

    def ids(self) -> KeysView:
        """
        Returns the ids of the contained objects in the index.
        """
        self._sync_indices()
        return self._id_index.keys()

    def keys(self) -> GeneratorType:
        """
        Returns the (name, id) pairs of all objects contained in the index.
        """
        self._sync_indices()
        return (tpl for tpl in zip(self.names(), self.ids()))

    def values(self) -> GeneratorType:
        """
        Returns all objects contained in the index.
        """
        self._sync_indices()
        return (obj for obj in self)

    def items(self) -> GeneratorType:
        """
        Returns (name, id, object) 3-tuples of all objects contained in the index
        """
        return (
            ((obj.name, obj.id), obj)
            for obj in self
        )

    def has(self, obj: Any) -> bool:
        """
        Returns whether an object *obj* is contained in the index. *obj* can be an :py:attr:`id`, a
        :py:attr:`name`, a :py:class:`UniqueObject` of type or a :py:class:`LazyUniqueObject` that
        wraps a type :py:attr:`cls`.
        """
        self._sync_indices()

        if isinstance(obj, self.cls) or (isinstance(obj, LazyUniqueObject) and obj.cls == self.cls):
            obj = obj.name

        if isinstance(obj, str):
            return obj in self._name_index

        if isinstance(obj, id):
            return obj in self._id_index

        return False

    def get(self, obj: Any, default: T = no_value) -> UniqueObject | T:
        """
        Returns an object *obj* contained in this index. *obj* can be an :py:attr:`id`, a
        :py:attr:`name`, a :py:class:`UniqueObject` of type or a :py:class:`LazyUniqueObject` that
        wraps a type :py:attr:`cls`. If no object could be found, *default* is returned if set. An
        exception is raised otherwise.
        """
        self._sync_indices()

        name_or_id = obj
        inst_passed = False
        if isinstance(obj, self.cls) or isinstance(obj, LazyUniqueObject) and obj.cls == self.cls:
            name_or_id = obj.name
            inst_passed = True

        _obj = None
        if isinstance(name_or_id, str):
            if name_or_id in self._name_index:
                _obj = self._name_index[name_or_id]

        elif isinstance(name_or_id, int):
            if name_or_id in self._id_index:
                _obj = self._id_index[name_or_id]

        # when an obj was an instance, but the found one is not equal to it, reset the found one
        if _obj is not None and inst_passed and _obj != obj:
            _obj = None

        # prepare and return the found object
        if _obj is not None:
            # materialize when the found object is lazy
            if isinstance(_obj, LazyUniqueObject):
                # remember the position of the object
                idx = self.objects.index(_obj)

                # materialize
                with _obj.materialize(self) as _obj:
                    # add back the materialized object
                    self.objects[idx] = _obj
                    self._name_index[_obj.name] = _obj
                    self._id_index[_obj.id] = _obj

            return _obj

        if default != no_value:
            return default

        raise ValueError(f"object '{obj}' not known to index '{self}'")

    def get_first(self, default: T = no_value) -> UniqueObject | T:
        """
        Returns the first object of this index. If no object could be found, *default* is returned
        if set. An exception is raised otherwise.
        """
        if self.objects:
            return self.get(self.objects[0].name)

        if default != no_value:
            return default

        raise Exception(f"cannot return first object, '{self}' is empty")

    def get_last(self, default: T = no_value) -> UniqueObject | T:
        """
        Returns the last object of this index. If no object could be found, *default* is returned if
        set. An exception is raised otherwise.
        """
        if self.objects:
            return self.get(self.objects[-1].name)

        if default != no_value:
            return default

        raise Exception(f"cannot return last object, '{self}' is empty")

    def add(
        self,
        obj: UniqueObject | LazyUniqueObject,
        overwrite: bool = False,
    ) -> UniqueObject | LazyUniqueObject:
        """
        Adds *obj*, a :py:class:`UniqueObject` of type or a :py:class:`LazyUniqueObject` that
        wraps a type :py:attr:`cls`, to the index. When an object with the same :py:attr:`name` or
        :py:attr:`id` already exists and *overwrite* is *False*, an exception is raised. Otherwise,
        the object is overwritten. The added object is returned.
        """
        if isinstance(obj, LazyUniqueObject):
            # unique object type of the lazy object and this index must match
            if self.cls != obj.cls:
                raise TypeError(
                    f"LazyUniqueObject '{obj}' must materialize into '{self.cls}' instead of "
                    f"'{obj.cls}'",
                )
        elif not isinstance(obj, self.cls):
            # type of the object must match that of the index
            raise TypeError(f"object '{obj}' to add must be of type '{self.cls}'")

        self._sync_indices()

        # handle duplicates
        if obj.name in self._name_index:
            if not overwrite:
                raise DuplicateNameException(self.cls, obj.name, self)
            self.remove(obj.name)
        if obj.id in self._id_index:
            if not overwrite:
                raise DuplicateIdException(self.cls, obj.id, self)
            self.remove(obj.id)

        # add to objects and indices
        self.objects.append(obj)
        self._name_index[obj.name] = obj
        self._id_index[obj.id] = obj

        return obj

    def extend(
        self,
        objects: "UniqueObjectIndex" | list[UniqueObject | LazyUniqueObject],
        overwrite: bool = False,
    ) -> None:
        """
        Adds multiple new *objects* of type :py:attr:`cls` to this index. See :py:meth:`add` for
        more info.
        """
        # when objects is an index, do not materialize its objects via the normal iterator
        gen = objects.objects if isinstance(objects, UniqueObjectIndex) else objects
        for obj in gen:
            self.add(obj, overwrite=overwrite)

    def index(self, obj: Any) -> int:
        """
        Returns the position of an object *obj* in this index. *obj* can be an :py:attr:`id`, a
        :py:attr:`name`, a :py:class:`UniqueObject` of type or a :py:class:`LazyUniqueObject` that
        wraps a type :py:attr:`cls`.
        """
        return self.objects.index(self.get(obj))

    def remove(self, obj: Any) -> bool:
        """
        Remove an object *obj* from the index. *obj* can be an :py:attr:`id`, a :py:attr:`name`, a
        :py:class:`UniqueObject` of type or a :py:class:`LazyUniqueObject` that wraps a type
        :py:attr:`cls`. *True* is returned in case an object could be removed, and *False*
        otherwise.
        """
        self._sync_indices()

        name_or_id = obj
        inst_passed = False
        if isinstance(obj, self.cls) or isinstance(obj, LazyUniqueObject) and obj.cls == self.cls:
            name_or_id = obj.name
            inst_passed = True

        _obj = None
        if isinstance(name_or_id, str):
            if name_or_id in self._name_index:
                _obj = self._name_index[name_or_id]

        elif isinstance(name_or_id, int):
            if name_or_id in self._id_index:
                _obj = self._id_index[name_or_id]

        # when an obj was an instance, but the found one is not equal to it, reset the found one
        if _obj is not None and inst_passed and _obj != obj:
            _obj = None

        # do nothing if no object was found, or if it does not exactly match the passed one
        if _obj is None:
            return False

        # remove from indices and objects
        self._name_index.pop(_obj.name)
        self._id_index.pop(_obj.id)
        self.objects.remove(_obj)

        return True

    def clear(self) -> None:
        """
        Removes all objects from the index.
        """
        del self.objects[:]
        self._sync_indices()


class DuplicateObjectException(Exception):
    """
    Base class for exceptions that are raised when a duplicate of a unique object is encountered.
    """


class DuplicateNameException(DuplicateObjectException):
    """
    An exception which is raised when a duplicate object, identified by its name, is encountered.
    """

    def __init__(
        self,
        cls: UniqueObjectMeta,
        name: str,
        index: UniqueObjectIndex | None = None,
    ) -> None:
        # create the message
        msg = f"duplicate '{cls.__module__}.{cls.__name__}' object with name '{name}' encountered"
        if index is not None:
            msg += f" in {index!r}"

        super().__init__(msg)


class DuplicateIdException(DuplicateObjectException):
    """
    An exception which is raised when a duplicate object, identified by its id, is encountered.
    """

    def __init__(
        self,
        cls: UniqueObjectMeta,
        id: int,
        index: UniqueObjectIndex | None = None,
    ) -> None:
        # create the message
        msg = f"duplicate '{cls.__module__}.{cls.__name__}' object with id '{id}' encountered"
        if index is not None:
            msg += f" in {index!r}"

        super().__init__(msg)

order/models/unique.py
cms-cat/order
# coding: utf-8

"""
Container for global settings.
"""

from __future__ import annotations


__all__ = ["Settings"]


import os
import re

from order.types import T
from order.util import no_value


class Settings(object):

    __instance = None

    @classmethod
    def instance(cls):
        if cls.__instance is None:
            cls.__instance = cls()
        return cls.__instance

    @classmethod
    def get_env(cls, name: str, default: T = no_value) -> T:
        if name not in os.environ:
            if default != no_value:
                return default
            raise Exception(f"undefined environment variable '{name}'")
        return os.environ[name]

    @classmethod
    def flag_to_bool(cls, flag: bool | int | str) -> bool:
        if isinstance(flag, int):
            return bool(flag)
        if isinstance(flag, bool):
            return flag

        _flag = str(flag).lower()
        if _flag in ("false", "no", "0"):
            return False
        if _flag in ("true", "yes", "1"):
            return True

        raise ValueError(f"cannot interpret '{flag}' as bool")

    @classmethod
    def get_data_location(cls) -> str:
        loc = cls.get_env("ORDER_DATA_LOCATION")
        scheme = re.match(r"^(\w+)\:\/\/(.*)$", loc)
        if not scheme:
            loc = f"file://{loc}"
        return loc

    @classmethod
    def get_cache_directory(cls) -> str:
        return cls.get_env("ORDER_CACHE_DIRECTORY", os.path.join(os.getcwd(), ".order_cache"))

    @classmethod
    def get_readonly_cache_directories(cls) -> list[str]:
        dirs = cls.get_env("ORDER_READONLY_CACHE_DIRECTORIES", [])
        if dirs:
            dirs = [d.strip() for d in dirs.split(",")]
        return dirs

    @classmethod
    def get_clear_cache(cls) -> bool:
        return cls.flag_to_bool(cls.get_env("ORDER_CLEAR_CACHE", False))

    @classmethod
    def get_cache_only(cls) -> bool:
        return cls.flag_to_bool(cls.get_env("ORDER_CACHE_ONLY", False))

    @classmethod
    def get_user_proxy(cls) -> str:
        return cls.get_env("X509_USER_PROXY", f"/tmp/x509up_u{os.getuid()}")

    def __init__(self):
        super().__init__()

        # get all settings
        self.data_location: str = self.get_data_location()
        self.cache_directory: str = self.get_cache_directory()
        self.readonly_cache_directories: list[str] = self.get_readonly_cache_directories()
        self.clear_cache: bool = self.get_clear_cache()
        self.cache_only: bool = self.get_cache_only()
        self.user_proxy: str = self.get_user_proxy()

order/settings.py
cms-cat/order
# coding: utf-8

"""
Custom type definitions and shorthands to simplify imports of types that are spread across multiple
packages.
"""

from __future__ import annotations


__all__ = []


import re
from collections.abc import KeysView, ValuesView  # noqa
from typing import Any, Union, TypeVar, ClassVar, List, Tuple, Sequence, Set, Dict, Callable  # noqa
from types import GeneratorType  # noqa

from typing_extensions import Annotated, _AnnotatedAlias as AnnotatedType  # noqa
from annotated_types import Ge, Len  # noqa
from pydantic import Field, Strict, StrictInt, StrictFloat, StrictStr  # noqa
from pydantic.fields import FieldInfo  # noqa


#: Strict positive integer.
PositiveStrictInt = Annotated[StrictInt, Ge(0)]

#: Strict non-empty string.
NonEmptyStrictStr = Annotated[StrictStr, Len(min_length=1)]

#: Generic type variable, more stringent than Any.
T = TypeVar("T")


class Lazy(object):
    """
    Annotation factory that adds :py:class:`AdapterModel` to the metadata of the returned annotated
    type.
    """

    @classmethod
    def __class_getitem__(cls, types: tuple[type]) -> type:
        from order.adapters.base import AdapterModel

        if not isinstance(types, tuple):
            types = (types,)
        return Union[tuple(map(cls.make_strict, types)) + (AdapterModel,)]

    @classmethod
    def parse_annotation(cls, type_str: str) -> list[str] | None:
        m = re.match(r"^Lazy\[(.+)\]$", type_str)
        return m and [s.strip() for s in m.group(1).split(",")]

    @classmethod
    def make_strict(cls, type_: type) -> AnnotatedType:
        # some types cannot be strict
        if not cls.can_make_strict(type_):
            return type_

        # when not decorated with strict meta data, just create a new strict tyoe
        if (
            not isinstance(type_, AnnotatedType) or
            not any(isinstance(m, Strict) for m in getattr(type_, "__metadata__", []))
        ):
            return Annotated[type_, Strict()]

        # when already strict, return as is
        metadata = type_.__metadata__
        if all(m.strict for m in metadata if isinstance(m, Strict)):
            return type_

        # at this point, strict metadata exists but it is actually disabled,
        # so replace it in metadata and return a new annotated type
        metadata = [
            (Strict() if isinstance(m, Strict) else m)
            for m in metadata
        ]
        return Annotated[(*type_.__args__, *metadata)]

    @classmethod
    def can_make_strict(cls, type_: type) -> bool:
        if type_.__dict__.get("_name") in ("Dict", "List"):
            return False

        return True

order/types.py
cms-cat/order
# coding: utf-8

"""
Helpful utilities.
"""

from __future__ import annotations


__all__ = ["no_value", "colored", "uncolored", "create_hash", "DotAccessProxy"]


import os
import sys
import re
import random
import hashlib

try:
    import ipykernel
    import ipykernel.iostream
except ImportError:
    ipykernel = None

from order.types import Any, Callable


#: Unique object denoting *no value*.
no_value = object()


# terminal codes for colors
colors = {
    "default": 39,
    "black": 30,
    "red": 31,
    "green": 32,
    "yellow": 33,
    "blue": 34,
    "magenta": 35,
    "cyan": 36,
    "light_gray": 37,
    "dark_gray": 90,
    "light_red": 91,
    "light_green": 92,
    "light_yellow": 93,
    "light_blue": 94,
    "light_magenta": 95,
    "light_cyan": 96,
    "white": 97,
}

# terminal codes for backgrounds
backgrounds = {
    "default": 49,
    "black": 40,
    "red": 41,
    "green": 42,
    "yellow": 43,
    "blue": 44,
    "magenta": 45,
    "cyan": 46,
    "light_gray": 47,
    "dark_gray": 100,
    "light_red": 101,
    "light_green": 102,
    "light_yellow": 103,
    "light_blue": 104,
    "light_magenta": 105,
    "light_cyan": 106,
    "white": 107,
}

# terminal codes for styles
styles = {
    "default": 0,
    "bright": 1,
    "dim": 2,
    "underlined": 4,
    "blink": 5,
    "inverted": 7,
    "hidden": 8,
}


def colored(msg, color=None, background=None, style=None, force=False):
    """
    Return the colored version of a string *msg*. For *color*, *background* and *style* options, see
    https://misc.flogisoft.com/bash/tip_colors_and_formatting. They can also be explicitely set to
    ``"random"`` to get a random value. Unless *force* is *True*, the *msg* string is returned
    unchanged in case the output is neither a tty nor an IPython output stream.
    """
    if not force:
        tty = False
        ipy = False

        try:
            tty = os.isatty(sys.stdout.fileno())
        except:
            pass

        if not tty and ipykernel is not None:
            ipy = isinstance(sys.stdout, ipykernel.iostream.OutStream)

        if not tty and not ipy:
            return msg

    # get the color
    if color == "random":
        color = random.choice(list(colors.values()))
    else:
        color = colors.get(color, colors["default"])

    # get the background
    if background == "random":
        background = random.choice(list(backgrounds.values()))
    else:
        background = backgrounds.get(background, backgrounds["default"])

    # get multiple styles
    if not isinstance(style, (tuple, list, set)):
        style = (style,)
    style_values = list(styles.values())
    style = ";".join(
        str(random.choice(style_values) if s == "random" else styles.get(s, styles["default"]))
        for s in style
    )

    return f"\033[{style};{background};{color}m{msg}\033[0m"


# compiled regular expression for removing all terminal style codes
uncolor_cre = re.compile(r"(\x1B\[[0-?]*[ -/]*[@-~])")


def uncolored(s):
    """
    Removes all terminal style codes from a string *s* and returns it.
    """
    return uncolor_cre.sub("", s)


def create_hash(inp: Any, l: int = 10, algo: str = "sha256", to_int: bool = False) -> str | int:
    """
    Takes an arbitrary input *inp* and creates a hexadecimal string hash based on an algorithm
    *algo*. For valid algorithms, see python's hashlib. *l* corresponds to the maximum length of the
    returned hash and is limited by the length of the hexadecimal representation produced by the
    hashing algorithm. When *to_int* is *True*, the decimal integer representation is returned.
    """
    h = getattr(hashlib, algo)(str(inp).encode("utf-8")).hexdigest()[:l]
    return int(h, 16) if to_int else h


class DotAccessProxy(object):
    """
    Proxy object that provides simple attribute access to values that are retrieved by a *getter*
    and optionally set through a *setter*. Example:

    .. code-block:: python

        my_dict = {"foo": 123}

        proxy = DotAccessProxy(my_dict.__getattr__)
        proxy.foo
        # -> 123
        proxy.bar
        # -> AttributeError

        proxy = DotAccessProxy(my_dict.get)
        proxy.foo
        # -> 123
        proxy.bar
        # -> None

        proxy = DotAccessProxy(my_dict.get, my_dict.__setitem__)
        proxy.foo
        # -> 123
        proxy.bar
        # -> None
        proxy.bar = 99
        proxy.bar
        # -> 99
    """

    def __init__(
        self,
        getter: Callable[[str], Any],
        setter: Callable[[str, Any], None] | None = None,
    ) -> None:
        super().__init__()

        self._getter = getter
        self._setter = setter

    def __call__(self, *args, **kwargs) -> Any:
        return self._getter(*args, **kwargs)

    def __getattr__(self, attr: str) -> Any:
        if attr.startswith("__") or attr in ("_getter", "_setter"):
            return super().__getattr__(attr)

        try:
            return self._getter(attr)
        except KeyError as e:
            raise AttributeError(*e.args)

    def __setattr__(self, attr: str, value: Any) -> None:
        if attr.startswith("__") or attr in ("_getter", "_setter"):
            super().__setattr__(attr, value)
            return

        if self._setter is None:
            raise Exception(
                f"cannot set attribute '{attr}', not setter defined on {self.__class__.__name__}",
            )
        self._setter(attr, value)

order/util.py
cms-cat/order
# coding: utf-8
# flake8: noqa


__all__ = []


# adjust the path to import order
import os
import sys
base = os.path.normpath(os.path.join(os.path.abspath(__file__), "../.."))
sys.path.append(base)
from order import *

# import all tests
from .test_util import *

tests/__init__.py
cms-cat/order
# coding: utf-8


__all__ = ["UtilTest"]


import unittest

from order.util import create_hash


class UtilTest(unittest.TestCase):

    def test_create_hash(self):
        data = (1, "2", True, (42,))

        # test different variations
        self.assertEqual(create_hash(data), "6c98c78722")
        self.assertEqual(create_hash(data, algo="sha512"), "75bd8c75c8")
        self.assertEqual(create_hash(data, l=12), "6c98c7872207")
        self.assertEqual(create_hash(data, to_int=True), 466419681058)

tests/test_util.py
cms-cat/order