import logging
from typing import Dict, Any, List, Callable, Union
from qval.framework_integration import load_symbol
from . import framework_integration as fwk
[docs]@fwk._make_request
def make_request(request: Union[Dict[str, str], fwk.Request]) -> fwk.RequestType:
"""
Creates a :class:`qval.framework_integration.DummyRequest`
if :code:`request` is a dictionary, and returns the :code:`request` itself otherwise.
The behavior of this function can be customized with the
:func:`@_make_request() <qval.framework_integration._make_request>` decorator.
Provide the path to your wrapper using :code:`QVAL_MAKE_REQUEST_WRAPPER` in the settings file
or set it as an environment variable. The wrapper function must accept :code:`request` as its first argument and
return an object that implements the request interface.
For example, the following code adds print each incoming request:
::
# settings.py
QVAL_MAKE_REQUEST_WRAPPER = "app.utils.my_wrapper"
# app/utils.py
def my_wrapper(f):
@functools.wraps(f)
def wrapper(request):
print(f"Received new request: {request}")
return f(request)
return wrapper
:param request: dict or request instance
:return: request
"""
if isinstance(request, dict):
return fwk.DummyRequest(request)
return request
[docs]def get_request_params(request: fwk.RequestType):
"""
Returns a dictionary of the query parameters in the given request.
:param request: any supported request
:return: dictionary of parameters
"""
supported_attrs = ("query_params", "GET", "args", "params")
for attr in supported_attrs:
if hasattr(request, attr):
return getattr(request, attr)
raise AttributeError(
"Provided request object does not have any of the following attributes: "
"{}.".format(", ".join(f"`{attr}`" for attr in supported_attrs))
)
[docs]def dummify(request: fwk.Request) -> fwk.DummyRequest:
"""
Constructs a :class:`qval.framework_integration.DummyRequest` with the parameters in the given request.
:param request: any supported request
:return: :code:`DummyRequest(request.<params>)`
"""
return fwk.DummyRequest(get_request_params(request))
[docs]class FrozenBox(object):
"""
A frozen dictionary that allows accessing its elements with :code:`.`
Example:
>>> box = FrozenBox({"num": 10, "s": "string"})
>>> print(box.num, box.s)
10 string
>>> box["num"] = 404
Traceback (most recent call last):
...
TypeError: 'FrozenBox' object does not support item assignment
>>> box.num = 404
Traceback (most recent call last):
...
TypeError: 'FrozenBox' object does not support attribute assignment
>>> box.num
10
"""
[docs] def __init__(self, dct: Dict[Any, Any]):
"""
:param dct: the dict to store
"""
self.__dict__["__dct__"] = dct
def __getitem__(self, item: str) -> Any:
"""
[] operator.
:param item:
:return: value for the key `item`
"""
return self.__dict__["__dct__"][item]
def __getattr__(self, item: str) -> Any:
"""
Returns the value of the stored `item` or attribute of the object
in case the item doesn't exist.
:param item: item key
:return: value
"""
return self[item]
def __setattr__(self, key: str, value: str):
"""
Raises TypeError.
"""
raise TypeError(
f"'{self.__class__.__name__}' object does not support attribute assignment"
)
def __contains__(self, item: str) -> bool:
"""
Determines if the item is stored in the dictionary.
:param item: item to check
"""
return item in self.__dict__["__dct__"]
def __iter__(self):
"""
Returns an iterator over :code:`__dct__.items()`
:return: :code:`iter(__dct__.items())`
"""
return iter(self.__dict__["__dct__"].items())
def __repr__(self) -> str:
"""
Returns an evaluable representation of the :class:`FrozenBox` object.
"""
return f"FrozenBox({self.__dict__['__dct__']})"
def __str__(self) -> str:
"""
Returns a string representation of the :class:`FrozenBox` object.
:return: str(box)
"""
return f"FrozenBox<{self.__dict__['__dct__']}>"
[docs]class ExcLogger(object):
"""
A class used to report critical errors.
>>> from qval.utils import log
>>> log
ExcLogger()
>>> log.is_enabled
True
>>> log.disable()
>>> print(log)
ExcLogger<<Logger qval (WARNING)>>, enabled = false>
"""
[docs] def __init__(self):
"""
Instantiates the logger.
:param logger: a list of loggers
"""
self.logger = logging.getLogger("qval")
self._enabled = True
@property
def is_enabled(self) -> bool:
"""
Returns True if logging is enabled.
"""
return self._enabled
[docs] def disable(self):
"""
Disables logging.
:return: None
"""
self._enabled = False
[docs] def enable(self):
"""
Enables logging.
:return: None
"""
self._enabled = True
[docs] def log(self, level: str, *args, **kwargs):
"""
Logs a new error message on the given level if logging is enabled.
:param args: logger args
:param kwargs: logger kwargs
:return: None
"""
if not self._enabled:
return
try:
self.logger.log(level, *args, **kwargs)
except Exception as e:
self.logger.error(
"Caught an error while logging with the parameters ({}, {}, {}):\n{}".format(
repr(level), args, kwargs, e
)
)
[docs] def error(self, *args, **kwargs):
"""
A shortcut for :meth:`log("error", ...) <qval.utils.ExcLogger.log>`.
:param args: log args
:param kwargs: log kwargs
:return: None
"""
self.log("error", *args, **kwargs)
def __repr__(self) -> str:
return "ExcLogger()"
def __str__(self) -> str:
return (
f"ExcLogger<{self.logger}>, " f"enabled = {str(self.is_enabled).lower()}>"
)
log = ExcLogger()