Source code for boss.helper

import base64
import hashlib
import importlib
import json
import time
import urllib
import urllib.parse
import urllib.request
from collections.abc import Callable
from io import IOBase
from typing import Any

from heros import RemoteHERO
from heros.helper import get_logger

log = get_logger("boss")




def get_module(path, relative_path=None):
    module_name = f"{relative_path}.{path}" if relative_path is not None else path
    module = importlib.import_module(module_name)
    return module





def get_class_by_name(name, relative_path=None):
    """
    Resolve class by name

    :param name: (str) "%s.%s" % (module.name, class.name)
    :return: (class)
    """
    assert name is not None

    module_path, class_name = name.rsplit(".", 1)

    module__ = get_module(module_path, relative_path=relative_path)
    class_ = getattr(module__, class_name)
    return class_





def file_from_url(url: str) -> IOBase:
    """
    Load content from a file specified by a URL.

    This can be every type of URL supported by pythons urllib
    (e.g. http://, file://, etc ). Giving the basic auth credentials in the URL in the form
    http://user:password@hostname:port/path is supported.

    Returns:
        file handle on the file
    """
    parsed = urllib.parse.urlparse(url)

    if parsed.username and parsed.password:
        request = urllib.request.Request(parsed._replace(netloc=parsed.netloc.split("@")[1]).geturl())
        base64string = base64.b64encode(bytes(f"{parsed.username}:{parsed.password}", "ascii"))
        request.add_header("Authorization", f"Basic {base64string.decode('utf-8')}")
        f_handle = urllib.request.urlopen(request)
    else:
        f_handle = urllib.request.urlopen(url)

    return f_handle





def extend_none_allowed_list(list1: list | None, list2: list | None) -> list | None:
    """Extend a list with another list which both can be None.

    If one of the lists is None, the other list is returned. If both lists are None, None is returned.

    Args:
        list1: list to extend
        list2: list to extend with
    """
    if list1 is None:
        return list2
    if list2 is None:
        return list1
    return list1 + list2





def get_remote_hero(name: str, realm: str, trials: int = 10, delay: float = 1.0):
    """Try multiple times to get RemoteHERO.

    Args:
        name: name of the remote HERO.
        realm: realm of the remote HERO.
        trials: number of trials before giving up
        delays: time delay between trials in seconds.
    """

    for i in range(trials):
        try:
            return RemoteHERO(name, realm)
        except Exception:  # noqa: BLE001
            log.warning(f"Could not get RemoteHERO {name} in trial {i + 1}/{trials}.")
        time.sleep(delay)

    raise NameError(f"Could not get HERO {name} after {trials} trials! Giving up.")





def add_class_descriptor(cls: type, attr_name: str, descriptor) -> None:
    """
    Add a descriptor to a class.

    This is a simple helper function which uses `setattr` to add an attribute to the class and then also calls
    `__set_name__` on the attribute.

    Args:
        cls: Class to add the descriptor to
        attr_name: Name of the attribute the descriptor will be added to
        descriptor: The descriptor to be added
    """
    setattr(cls, attr_name, descriptor)
    getattr(cls, attr_name).__set_name__(cls, attr_name)





def get_or_create_dynamic_subclass(base_cls: Any, *args: Any, **kwargs: Any) -> type:
    """Return a cached dynamic subclass of ``base_cls`` based on the input arguments.

    This helper generates a subclass of ``base_cls`` whose identity is determined by ``*args`` and ``**kwargs``.
    The argument signature is serialized into a hash, which is then used as both a cache key and the dynamic subclass
    name. If the subclass for a given argument combination already exists, it is returned from cache.

    The generated subclass replaces ``__new__`` with a dummy implementation to prevent recursive invocation of
    ``base_cls.__new__``.

    Args:
        base_cls: The base class to derive from. Must be passed positionally.
        *args: Positional values that should influence subclass identity.
        **kwargs: Keyword values that should influence subclass identity.

    Returns:
        A dynamically generated subclass of ``base_cls``.

    Raises:
        TypeError: If arguments cannot be serialized for hashing.
    """
    # normalize args+kwargs
    signature = json.dumps(
        {"args": args, "kwargs": kwargs},
        sort_keys=True,
        default=str,
    )
    arg_hash = hashlib.blake2s(signature.encode()).hexdigest()
    new_cls_name = f"{base_cls.__name__}_{arg_hash}"
    # generate cache
    generated = getattr(base_cls, "_generated_classes", {})
    if new_cls_name in generated:
        return generated[new_cls_name]

    # dummy __new__ method for subclasses
    def _no_new(subcls: type, *_a, **_kw) -> Callable:
        return object.__new__(subcls)

    # generate new class
    new_cls = type(
        new_cls_name,
        (base_cls,),
        {"__new__": _no_new},
    )
    # cache it
    if not hasattr(base_cls, "_generated_classes"):
        base_cls._generated_classes = {}
    base_cls._generated_classes[new_cls_name] = new_cls
    return new_cls