# -*- coding: utf-8 -*-
import functools
import importlib
import inspect
import logging
import types
from collections import deque
from collections.abc import Mapping
from typing import (
Any,
Awaitable,
Callable,
Hashable,
Iterator,
List,
MutableMapping,
Optional,
Tuple,
Type,
)
from . import lang
from .settings import check_override, check_protected
__all__ = ['AttributesDict']
protected = lang.protected(check=check_protected)
override = lang.override(check=check_override)
_LOGGER = logging.getLogger(__name__)
SAVED_STATE_TYPE = MutableMapping[str, Any]
PID_TYPE = Hashable
class Frozendict(Mapping):
"""
An immutable wrapper around dictionaries that implements the complete :py:class:`collections.abc.Mapping`
interface. It can be used as a drop-in replacement for dictionaries where immutability is desired.
Adapted from: slezica/python-frozendict
"""
def __init__(self, *args: Any, **kwargs: Any) -> None:
self._dict = dict(*args, **kwargs)
self._hash: Optional[int] = None
def __getitem__(self, key: str) -> Any:
return self._dict[key]
def __contains__(self, key: Any) -> bool:
return key in self._dict
def copy(self, **add_or_replace: Any) -> 'Frozendict':
return self.__class__(self, **add_or_replace)
def __iter__(self) -> Iterator[str]:
return iter(self._dict)
def __len__(self) -> int:
return len(self._dict)
def __repr__(self) -> str:
return f'<{self.__class__.__name__} {self._dict!r}>'
def __hash__(self) -> int:
if self._hash is None:
hashed = 0
for key, value in self._dict.items():
hashed ^= hash((key, value))
self._hash = hashed
return self._hash
class AttributesFrozendict(Frozendict):
def __init__(self, *args: Any, **kwargs: Any):
super().__init__(*args, **kwargs)
self._initialised: bool = True
def __getattr__(self, attr: str) -> Any:
"""
Read a key as an attribute. Raise AttributeError on missing key.
Called only for attributes that do not exist.
"""
# This attribute is looked for by pickle when deserialising. At this point
# the object is not yet constructed and so accessing any members is
# dangerous and often causes infinite recursion so I have to guard like this.
if attr == '__setstate__':
raise AttributeError()
try:
return self[attr]
except KeyError:
errmsg = f"'{self.__class__.__name__}' object has no attribute '{attr}'"
raise AttributeError(errmsg)
def __dir__(self) -> List[str]:
"""
So we get tab completion.
:return: The keys of the dict
"""
return list(self.keys())
[docs]
class AttributesDict(types.SimpleNamespace):
"""Works like a dictionary, but items can also be added / accessed as attributes.
For example::
dct = AttributeDict()
dct["key1"] = "value"
dct.key2 = "value"
"""
def __setitem__(self, key: str, value: Any) -> None:
setattr(self, key, value)
def __getitem__(self, item: str) -> Any:
try:
return getattr(self, item)
except AttributeError:
raise KeyError(f"No key '{item}'")
def __delitem__(self, item: str) -> None:
return delattr(self, item)
[docs]
def setdefault(self, key: str, value: Any) -> Any:
return self.__dict__.setdefault(key, value)
[docs]
def get(self, *args: Any, **kwargs: Any) -> Any:
return self.__dict__.get(*args, **kwargs)
def load_function(name: str, instance: Optional[Any] = None) -> Callable[..., Any]:
obj = load_object(name)
if inspect.ismethod(obj):
if instance is not None:
return obj.__get__(instance, instance.__class__)
return obj
if inspect.isfunction(obj):
return obj
raise ValueError(f"Invalid function name '{name}'")
def load_object(fullname: str) -> Any:
"""
Load a class from a string
"""
obj, remainder = load_module(fullname)
# Finally, retrieve the object
for name in remainder:
try:
obj = getattr(obj, name)
except AttributeError:
raise ValueError(f"Could not load object corresponding to '{fullname}'")
return obj
def load_module(fullname: str) -> Tuple[types.ModuleType, deque]:
parts = fullname.split('.')
# Try to find the module, working our way from the back
mod = None
remainder: deque = deque()
for _ in range(len(parts)):
try:
mod = importlib.import_module('.'.join(parts))
break
except ImportError:
remainder.appendleft(parts.pop())
if mod is None:
raise ValueError(f"Could not load a module corresponding to '{fullname}'")
return mod, remainder
def type_check(obj: Any, expected_type: Type) -> None:
if not isinstance(obj, expected_type):
raise TypeError(f"Got object of type '{type(obj)}' when expecting '{expected_type}'")
def ensure_coroutine(coro_or_fn: Any) -> Callable[..., Awaitable[Any]]:
"""
Ensure that the given function ``fct`` is a coroutine
If the passed function is not already a coroutine, it will be made to be a coroutine
:param fct: the function
:returns: the coroutine
"""
if inspect.iscoroutinefunction(coro_or_fn):
return coro_or_fn
if inspect.iscoroutinefunction(coro_or_fn.__call__):
return coro_or_fn
if callable(coro_or_fn):
if inspect.isclass(coro_or_fn):
coro_or_fn = coro_or_fn.__call__
from .greenback_bridge import run_with_portal
@functools.wraps(coro_or_fn)
async def wrap(*args: Any, **kwargs: Any) -> Callable[..., Any]:
return await run_with_portal(coro_or_fn, *args, **kwargs)
return wrap
raise TypeError('coro_or_fn must be a callable')
def is_mutable_property(cls: Any, attribute: str) -> bool:
"""
Determine whether the given attribute is a mutable property of cls. That is to say that
the attribute corresponds to a property whose fset attribute is not None.
:param cls: the class
:param attribute: the attribute
:returns: True if the attribute is a mutable property of cls
"""
try:
attr = getattr(cls, attribute)
except AttributeError:
return False
return isinstance(attr, property) and attr.fset is not None
