PK4NC^^^qval/__init__.py""" This module provides convenient API for verifying query parameters. The core class is called `QueryParamValidator`. It accepts 4 arguments: - request: Request instance (Any object that has following attributes: GET, query_params and body) - factories: Dictionary of factories {param -> factory}. The value of the parameter will be provided to factory before validation. Any callable that accepts string and returns anything is a valid factory. - validators: Dictionary if validators {param -> Validator}. Validator is basically a list of predicates with the __call__() operator. See `Validator` class for more info. - box_all: If true, adds all request parameter to the final result. Otherwise, only specified in `factories` parameters will be added. If any parameter fails validation, `InvalidQueryParamException` (HTTP code = 400) will be raised. Also, only `TypeError`, `ValueError` and `KeyError` occurred when param is provided to factory result in the same exception. Any error thrown inside or outside of the context will raise an APIError (HTTP code = 500). Example: >>> from qval.framework_integration import DummyRequest >>> r = DummyRequest({"num": "42"}) >>> with QueryParamValidator(r, {"num": int}) as p: ... print(p.num) 42 The code above is too verbose. That's why you should use `validate()` - this function does all the boilerplate work for you: - `validate()` automatically converts dictionary to Request-like objects - Key-value arguments are used to provide factories - It's easier to type Simple example: >>> r = {"num": "42", "string": "s", "price": "3.14"} # you can use dictionary instead of a Request instance >>> with validate(r, num=int, price=float) as p: ... print(p.num, p.price, p.string, sep=', ') 42, 3.14, s A little bit more complex example, with a custom factory: >>> r = {"price": "2.79$", "tax": "0.5$"} >>> currency2f = lambda x: float(x[:-1]) # factory that converts {num}$ to float >>> with validate(r, price=currency2f, tax=currency2f) as p: ... print(p.price, p.tax, sep=', ') 2.79, 0.5 You can also use `qval()` decorator: >>> factories = {"num": int, "special": float} >>> validators = {"special": Validator(lambda x: x > 0)} >>> @qval(factories, validators) ... def view(request, params): # class-based views are also supported ... print(params.num, params.special, sep=", ") >>> view({"num": "10", "special": "0.7"}) 10, 0.7 If something fails during the validation or inside of the function, an error will be thrown. Consider the following example: >>> factories = {"num": int, "special": int} # now `special` is an integer >>> @qval(factories, validators=None) # no validators for simplicity ... def view(request, params): ... pass >>> view({"num": "10", "special": "0.7"}) # doctest: +IGNORE_EXCEPTION_DETAIL Traceback (most recent call last): ... qval.exceptions.InvalidQueryParamException: {'error': 'Invalid type of the `special` parameter: expected int.'}. The HTTP code of the exception above is 400 (Bad Request). Now the error is raised inside of the context block: >>> factories = {"num": int, "special": float} >>> @qval(factories, validators=None) # no validators for simplicity ... def view(request, params): ... raise IOError # some random exception >>> view({"num": "10", "special": "0.7"}) # doctest: +IGNORE_EXCEPTION_DETAIL Traceback (most recent call last): ... qval.framework_integration.APIException: An error occurred while processing you request. Please contact website administrator. The HTTP code of the exception above is 500 (Internal Server Error). The error is logged to stdout by default. See the Note section for more info. Documentation: Refer to documentation at https://qval.rtfd.io. """ from .utils import log from .qval import QueryParamValidator, validate, qval, qval_curry from .exceptions import InvalidQueryParamException, APIException from .validator import Validator, QvalValidationError __version__ = "0.3.1" PK0Mw%..qval/exceptions.pyfrom typing import Union from . import framework_integration class InvalidQueryParamException(framework_integration.APIException): """ An error thrown when param fails the validation. """ def __init__(self, detail: Union[dict, str], status: int): """ Instantiates the exception. :param detail: dict or string with details :param status: status code """ super().__init__(detail) self.status_code = status # Avoid circular imports APIException = framework_integration.APIException PK4N-KKqval/framework_integration.pyimport os import logging from typing import Union, Dict from importlib import import_module class _EnvironSettings(object): # pragma: no cover """ Lookups attribute calls in os.environ. """ def __getattr__(self, item): item = os.environ.get(item) # Support `hasattr()` if item is None: raise AttributeError return item class DummyRequest(object): """ DummyRequest. Used for compatibility with supported frameworks. """ def __init__(self, params: Dict[str, str]): self.GET = params self.body = "" @property def query_params(self) -> Dict[str, str]: """ More semantically correct name for request.GET. """ return self.GET Request = DummyRequest RequestType = (dict, Request) def get_module() -> Union[_EnvironSettings, "Module"]: # pragma: no cover """ Attempts to load settings module. If none of the supported env variables are defined, returns :class:`_EnvironSettings()` object. """ module = None modules = ["DJANGO_SETTINGS_MODULE", "SETTINGS_MODULE"] for module in map(os.environ.get, modules): if module is not None: module = module.replace(".py", "").replace("/", ".") break return _EnvironSettings() if module is None else import_module(module) module = get_module() def load_symbol(path: Union[object, str]): # pragma: no cover """ Imports object using the given path. :param path: path to an object, e.g. my.module.func_1 :return: loaded symbol """ # Path is already a symbol if not isinstance(path, str): return path _mod, _symbol = path.rsplit(".", maxsplit=1) return getattr(import_module(_mod), _symbol) # Check if DRF is installed try: from rest_framework.request import Request as _Request from rest_framework.exceptions import APIException from rest_framework.status import ( # lgtm [py/unused-import] HTTP_400_BAD_REQUEST, HTTP_500_INTERNAL_SERVER_ERROR, ) Request = _Request RequestType += (_Request,) REST_FRAMEWORK = True except ImportError: # pragma: no cover REST_FRAMEWORK = False # Define missing symbols class APIException(Exception): """ Base class for Qval's exceptions. Used if DRF is not installed. """ def __init__(self, detail: Union[dict, str]): """ Instantiates the exception object. :param detail: dict or string with details """ self.detail = detail self.status_code = HTTP_500_INTERNAL_SERVER_ERROR super().__init__(detail) HTTP_400_BAD_REQUEST = 400 HTTP_500_INTERNAL_SERVER_ERROR = 500 if hasattr(module, "QVAL_REQUEST_CLASS"): Request = load_symbol(module.QVAL_REQUEST_CLASS) RequestType += (Request,) # Check if Django is installed try: # pragma: no cover from django.http import HttpRequest, JsonResponse Request = HttpRequest RequestType += (Request,) class HandleAPIExceptionDjango(object): def __init__(self, get_response): self.get_response = get_response def __call__(self, request): return self.get_response(request) def process_exception(self, _: Request, exception: Exception): if isinstance(exception, APIException): detail = exception.detail if isinstance(detail, str): detail = {"error": detail} return JsonResponse(detail, status=exception.status_code) def setup_django_middleware(module: "Module" = None): """ Setups exception hanlding middleware. :param module: settings module :return: None """ if module is None: module = get_module() if hasattr(module, "MIDDLEWARE"): module.MIDDLEWARE.append( "qval.framework_integration.HandleAPIExceptionDjango" ) else: logging.warning( "Unable to add APIException middleware to the MIDDLEWARE list. " "Django does not support APIException handling without DRF integration. " "Define DJANGO_SETTINGS_MODULE or add 'qval.framework_integration.HandleAPIExceptionDjango' " "to the MIDDLEWARE list." ) # Setup middleware if DRF is not installed if ( hasattr(module, "INSTALLED_APPS") and "rest_framework" not in module.INSTALLED_APPS ): setup_django_middleware() except ImportError: # pragma: no cover pass # Check if flask is installed try: from flask import Request RequestType += (Request,) except ImportError: # pragma: no cover pass # Check if falcon is installed try: from falcon import Request RequestType += (Request,) except ImportError: # pragma: no cover pass # Check if custom wrapper is provided if hasattr(module, "QVAL_MAKE_REQUEST_WRAPPER"): _make_request = load_symbol(module.QVAL_MAKE_REQUEST_WRAPPER) else: def _make_request(f): """ Wraps default `utils.make_request()` function. Does nothing. """ return f def setup_flask_error_handlers(app: "flask.Flask"): # pragma: no cover """ Setups error handler for APIException. :param app: flask app :return: None """ from flask import jsonify @app.errorhandler(APIException) def handle_api_exception(error: APIException): """ Handles APIException in Flask. """ response = error.detail if isinstance(response, str): response = {"error": response} response = jsonify(response) response.status_code = error.status_code return response def setup_falcon_error_handlers(api: "falcon.API"): # pragma: no cover """ Setups error handler for APIException. :param api: falcon.API :return: """ # try to use faster json library try: import ujson as json except ImportError: import json from falcon import HTTP_400, HTTP_500, Response, Request, __version__ major_version = int(__version__.split(".", maxsplit=1)[0]) # Falcon 2.0.0 changed the order of arguments if major_version >= 2: def handle_api_exception( _rq: Request, _rp: Response, exc: "APIException", _p: dict ): """ Handles APIException in Falcon. """ code = HTTP_400 if exc.status_code == 400 else HTTP_500 detail = ( {"error": exc.detail} if isinstance(exc.detail, str) else exc.detail ) _rp.body = json.dumps(detail) _rp.status = code else: def handle_api_exception(exc: "APIException", _rq, _rp: Response, _p): """ Handles APIException in Falcon. """ code = HTTP_400 if exc.status_code == 400 else HTTP_500 detail = ( {"error": exc.detail} if isinstance(exc.detail, str) else exc.detail ) _rp.body = json.dumps(detail) _rp.status = code api.add_error_handler(APIException, handler=handle_api_exception) # RequestType is a tuple that will be used in type checking # Request is a Union of request types and is used in annotations Request = Union[RequestType] PK4N?Y;; qval/qval.pyimport functools from typing import Any, Callable, Dict, Optional, Union from contextlib import contextmanager, AbstractContextManager, ExitStack from . import utils from .validator import Validator, QvalValidationError from . import exceptions from . import framework_integration as fwk class QueryParamValidator(AbstractContextManager): """ Validates query parameters. Examples: >>> r = fwk.DummyRequest({"num": "42", "s": "str", "double": "3.14"}) >>> params = QueryParamValidator(r, dict(num=int, s=None, double=float)) >>> with params as p: ... print(p.num, p.s, p.double, sep=', ') 42, str, 3.14 .. automethod:: __enter__ .. automethod:: __exit__ """ def __init__( self, request: fwk.Request, factories: Dict[str, Optional[type]], validators: Dict[str, Validator.ValidatorType] = None, box_all: bool = True, ): """ Instantiates query validator object. :param request: fwk.Request instance :param factories: mapping of :code:`{param -> factory}`. Providing :code:`None` as a factory is equivalent to :code:`str` or :code:`lambda x: x`, since params are stored as strings. :param validators: dictionary of pre-defined validators :param box_all: include all params, even if they're not specified in :code:`factories` """ self.request = request self._factories = factories self._box_all = box_all self._query_params = utils.get_request_params(self.request) self.result: Dict[str, Any] = { k: self.query_params[k] # Add all parameters to resulting dictionary if box_all is true. # Otherwise keep only specified parameters. for k in (self.query_params if self._box_all else self._factories) } self._params: Dict[str, Validator] = {k: Validator() for k in self.result} self._params.update( { # Convert predicates to validators k: Validator(v) if not isinstance(v, Validator) else v for k, v in (validators or {}).items() } ) def apply_to_request( self, request: Union[Dict[str, str], fwk.Request] ) -> "QueryParamValidator": """ Applies current validation settings to a new request. Example: >>> from qval.utils import make_request >>> request = make_request({"a": "77"}) >>> params = QueryParamValidator(request, {"a": int}, {"a": lambda x: x > 70}) >>> with params as p: ... print(p.a) # Prints 77 77 >>> with params.apply_to_request({"a": "10"}): pass # Error! Traceback (most recent call last): ... qval.exceptions.InvalidQueryParamException: ... :param request: new request instance :return: new :class:`QueryParamValidator` instance """ return self.__class__( utils.make_request(request), self._factories, self._params, self._box_all ) @property def query_params(self) -> Dict[str, str]: """ Returns dictionary of query parameters. """ return self._query_params def add_predicate(self, param: str, predicate: Callable[[Any], bool]): """ Adds new check for provided parameter. :param param: name of the request parameter :param predicate: predicate function :return: None """ self._params.setdefault(param, Validator()) self._params[param].add(predicate) # Alias for add_predicate; returns reference def check( self, param: str, predicate: Callable[[Any], bool] ) -> "QueryParamValidator": """ Adds new check from provided parameter. :param param: name of the request parameter :param predicate: predicate function :return: self """ self.add_predicate(param, predicate) return self def positive( self, param: str, transform: Callable[[Any], Any] = lambda x: x ) -> "QueryParamValidator": """ Adds greater than zero comparison check for provided parameter. Provided :code:`param` will be tested as [:code:`transform(param) > 0`]. :param param: name of the request parameter :param transform: callable that transforms the parameter, default: :code:`lambda x: x` :return: self """ return self.check(param, lambda x: transform(x) >= 0) def gt( self, param: str, value: Any, transform: Callable[[Any], Any] = lambda x: x ) -> "QueryParamValidator": """ Adds greater than comparison check for provided parameter. For example, if value = 10, parameter :code:`param` will be tested as [:code:`transform(param) > 10`]. :param param: name of the request parameter :param value: value to compare with :param transform: callable that transforms the parameter, default: :code:`lambda x: x` :return: self """ return self.check(param, lambda x: transform(x) > value) def lt( self, param: str, value: Any, transform: Callable[[Any], Any] = lambda x: x ) -> "QueryParamValidator": """ Adds less than comparison check for provided parameter. For example, if value = 10, parameter :code:`param` will be tested as [:code:`transform(param) < 10`]. :param param: name of the request parameter :param value: value to compare with :param transform: callable that transforms the parameter, default: :code:`lambda x: x` :return: self """ return self.check(param, lambda x: transform(x) < value) def eq( self, param: str, value: Any, transform: Callable[[Any], Any] = lambda x: x ) -> "QueryParamValidator": """ Adds equality check for provided parameter. For example, if value = 10, parameter :code:`param` will be tested as [:code:`transform(param) == 10`]. :param param: name of the request parameter :param value: value to compare with :param transform: callable that transforms the parameter, default: :code:`lambda x: x` :return: self """ return self.check(param, lambda x: transform(x) == value) def nonzero( self, param: str, transform: Callable[[Any], Any] = lambda x: x ) -> "QueryParamValidator": """ Adds nonzero check for provided parameter. For example, if value = 10, parameter :code:`param` will be tested as [:code:`transform(param) != 0`]. :param param: name of the request parameter :param transform: callable that transforms the parameter, default: :code:`lambda x: x` :return: self """ return self.check(param, lambda x: transform(x) != 0) @contextmanager def _cleanup_on_error(self): """ Unwinds the stack in case of an error. """ with ExitStack() as stack: stack.push(self) yield # The validation checks didn't raise an exception stack.pop_all() def _validate(self): """ Validates the parameters. Only KeyError, ValueError and TypeError are handled as expected errors. :return: None """ # Firstly, cast the parameters to required types for param, cast in self._factories.items(): try: # If cast is None, just leave the parameter a string cast = cast or (lambda x: x) value = cast(self.query_params[param]) self.result[param] = value # Missing a required parameter except KeyError: raise exceptions.InvalidQueryParamException( {"error": f"Missing required parameter `{param}`."}, status=fwk.HTTP_400_BAD_REQUEST, ) # Invalid cast except (ValueError, TypeError): expected = "." # Expose only built-in types if cast in (int, float): expected = f": expected {cast.__name__}." raise exceptions.InvalidQueryParamException( {"error": f"Invalid type of the `{param}` parameter{expected}"}, status=fwk.HTTP_400_BAD_REQUEST, ) # Run validations on each parameter for param, value in self.result.items(): validator = self._params[param] try: if not validator(value): raise QvalValidationError( f"Invalid `{param}` value: {self.result[param]}." ) except QvalValidationError as e: raise exceptions.InvalidQueryParamException( {"error": str(e)}, status=fwk.HTTP_400_BAD_REQUEST ) from e def __enter__(self) -> "utils.FrozenBox": """ Runs validation on provided request. See __exit__() for additional info. :return: box of validated values. """ # This context manager will unwind stack in case of an error. # The __exit__() method will be called with values of the exception raised inside _validate(). # This allows us handle exceptions both inside _validate() and inside of the context. with self._cleanup_on_error(): self._validate() return utils.FrozenBox(self.result) def __exit__(self, exc_type, exc_val, exc_tb): """ If occurred exception is not an :class:`InvalidQueryParamException `, the exception will be re-raised as an APIException, which will result in 500 error on the client side. :param exc_type: exception type :param exc_val: exception instance :param exc_tb: exception traceback :return: None """ # Report unexpected exceptions if exc_type not in (exceptions.InvalidQueryParamException, None): body = getattr(self.request, "body", {}) text = ( f"An error occurred during the validation or inside of the context: exc `{exc_type}` ({exc_val}).\n" f"| Parameters: {self.query_params}\n" f"| Body : {body}\n" f"| Exception:\n" ) utils.log.error( __name__, text, extra={ "stack": True, "traceback": exc_tb, "request_body": body, "parameters": self.query_params, }, exc_info=(exc_type, exc_val, exc_tb), ) raise exceptions.APIException( detail="An error occurred while processing you request. " "Please contact the website administrator." ) from exc_val def validate( request: Union[fwk.Request, Dict[str, str]], validators: Dict[str, Validator.ValidatorType] = None, box_all: bool = True, **factories, ) -> QueryParamValidator: """ Shortcut for QueryParamValidator. Examples: >>> r = {"num": "42", "s": "str", "double": "3.14"} >>> with validate(r, num=int, s=None, double=float) as p: ... print(p.num + p.double, p.s) 45.14 str >>> r = {"price": "43.5$", "n_items": "1"} >>> currency2f = lambda x: float(x[:-1]) >>> params = validate(r, price=currency2f, n_items=int ... ).positive("n_items") # n_items must be greater than 0 >>> with params as p: ... print(p.price, p.n_items) 43.5 1 :param request: request instance :param validators: dictionary of predefined validators :param box_all: include all params that no specified in factories in the param box :param factories: factories that create python object from string parameters :return: QueryParamValidator instance """ # Wrap dictionary with request-like object request = utils.make_request(request) return QueryParamValidator(request, factories, validators, box_all) def qval( factories: Dict[str, Optional[Callable[[str], Any]]], validators: Dict[str, Validator.ValidatorType] = None, box_all: bool = True, request_: fwk.Request = None, ): """ A decorator that validates query parameters. The wrapped function must accept a request as the first parameter (or second if it's a method) and `params` as last. :param factories: mapping (parameter, callable [str -> Any]) :param validators: mapping (parameter, validator) :param box_all: include all params that no specified in fields in the param box :param request_: optional request object that can be provided to validator :return: wrapped function """ # Check if decorator is used improperly, i.e without `factories` if callable(factories): raise TypeError("qval() missing 1 required positional argument: 'factories'") def outer(f): @functools.wraps(f) def inner(*args, **kwargs): args = list(args) # If a default request object is provided, simply use it if request_ is not None: request = utils.make_request(request_) args.insert(0, request) # Otherwise check the arguments elif isinstance(args[0], fwk.RequestType): # And construct a request from the dict request = args[0] = utils.make_request(args[0]) elif len(args) > 1 and isinstance(args[1], fwk.RequestType): request = args[1] = utils.make_request(args[1]) else: raise ValueError( "The first argument of the view must be a request-like." ) with validate(request, validators, box_all, **factories) as params: return f(*args, params, **kwargs) return inner return outer def qval_curry(request: fwk.Request): """ Curries :func:`qval() ` decorator and provides given :code:`request` object on each call. This is handy especially in Flask, where request object is global. Example: .. code-block:: python >>> r = {"num": "42", "s": "str", "double": "3.14"} >>> qval = qval_curry(r) >>> @qval({"num": int, "double": float}, None) ... def view(request, extra_param, params): ... print(params.num, params.double, params.s, extra_param, sep=', ') >>> view("test") 42, 3.14, str, test :param request: request instance :return: wrapped :code:`qval(..., request_=request)` """ @functools.wraps(qval) def outer(*args, **kwargs): kwargs.setdefault("request_", request) return qval(*args, **kwargs) return outer PK~N#3 qval/utils.pyimport logging from typing import Dict, Any, List, Callable, Union from qval.framework_integration import load_symbol from . import framework_integration as fwk @fwk._make_request def make_request(request: Union[Dict[str, str], fwk.Request]) -> fwk.RequestType: """ Creates DummyRequest if :code:`request` is a dictionary, and returns the :code:`request` itself otherwise. Behavior of this function can be customized with the :func:`@_make_request() ` decorator. Provide 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 parameter and return object that implements request interface. For example, the following code adds print to each function call: :: # app/utils.py def my_wrapper(f): @functools.wraps(f) def wrapper(request): print(f"Received new request: {request}") return f(request) return wrapper Then execute :code:`export QVAL_MAKE_REQUEST_WRAPPER=app.utils.my_wrapper` in your console or simply add it to the config file. :param request: dict or request instance :return: request """ if isinstance(request, dict): return fwk.DummyRequest(request) return request def get_request_params(request: fwk.RequestType): """ Returns dictionary of query parameters of 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 has no any of the following attributes: " "{}.".format(", ".join(f"`{attr}`" for attr in supported_attrs)) ) def dummyfy(request: fwk.Request) -> fwk.DummyRequest: """ Constructs :class:`qval.framework_integration.DummyRequest` with params of the given request. :param request: any supported request :return: :code:`DummyRequest(request.)` """ return fwk.DummyRequest(get_request_params(request)) class FrozenBox(object): """ Frozen dictionary that allows accessing elements by :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 """ def __init__(self, dct: Dict[Any, Any]): """ :param dct: dict to store """ self.__dict__["__dct__"] = dct def __getitem__(self, item: str) -> Any: """ [] operator. :param item: item key :return: value for key `item` """ return self.__dict__["__dct__"][item] def __getattr__(self, item: str) -> Any: """ Returns value of the stored `item` or attribute of the object. :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 item is inside of the dictionary. :param item: item to check """ return item in self.__dict__["__dct__"] def __iter__(self): """ Returns iterator over :code:`__dct__.items()` :return: :code:`iter(__dct__.items())` """ return iter(self.__dict__["__dct__"].items()) def __repr__(self) -> str: """ Returns evaluable representation of the :class:`FrozenBox` object. """ return f"FrozenBox({self.__dict__['__dct__']})" def __str__(self) -> str: """ Returns string representation of the :class:`FrozenBox` object. :return: str(box) """ return f"FrozenBox<{self.__dict__['__dct__']}>" class ExcLogger(object): """ A class used to report critical errors. >>> from qval.utils import log >>> log ExcLogger([getLogger]) >>> log.is_enabled True >>> log.disable() >>> print(log) ExcLogger<[getLogger], enabled = false> """ def __init__(self, logger_factories: List[Callable[[str], Any]]): """ Instantiates the ExcLogger. :param logger_factories: list of logger factories """ self.factories = logger_factories self._enabled = True @property def is_enabled(self) -> bool: """ Returns True if logging is enabled. """ return self._enabled def disable(self): """ Disables logging. :return: None """ self._enabled = False def enable(self): """ Enables logging. :return: None """ self._enabled = True def add_logger(self, log_factory: Callable[[str], Any]): """ Adds new logger factory to list. :param log_factory: logger :return: None """ self.factories.append(log_factory) def clear(self): """ Removes all saved factories. :return: None """ self.factories.clear() def dump(self, name: str, level: str, *args, **kwargs): """ Instantiates new loggers using configured factories and provides :code:`*args` and :code:`**kwargs` to built objects. :param name: logger name :param level: logging level. If built object has no attribute :code:`level`, it will be treated as callable. :param args: logger args :param kwargs: logger kwargs :return: None """ if not self._enabled: return for build in self.factories: try: logger = build(name) if hasattr(logger, level): getattr(logger, level)(*args, **kwargs) else: logger(*args, **kwargs) except TypeError: raise except: # lgtm [py/catch-base-exception] pass def error(self, name: str, *args, **kwargs): """ Shortcut for :meth:`dump(name, "error", ...) `. :param name: logger name :param args: logger args :param kwargs: logger kwargs :return: None """ self.dump(name, "error", *args, **kwargs) @staticmethod def collect_loggers() -> list: """ Looks for configuration and returns list of detected loggers or :func:`logging.getLogger`. :return: list of collected loggers """ module = fwk.get_module() if hasattr(module, "QVAL_LOGGERS"): _loggers = module.QVAL_LOGGERS if not isinstance(_loggers, (tuple, list)): _loggers = [_loggers] loggers = [load_symbol(log) for log in _loggers] else: loggers = [logging.getLogger] return loggers @classmethod def detect_loggers(cls, silent: bool = False) -> "ExcLogger": """ Looks for configuration and instantiates ExcLogger with detected loggers or default :func:`logging.getLogger`. :param silent: omit logging test message :return: ExcLogger object """ logger = cls(cls.collect_loggers()) if not silent: logger.dump(__file__, "info", "test message") return logger def __repr__(self) -> str: return f"ExcLogger([{', '.join(map(lambda x: x.__name__, self.factories))}])" def __str__(self) -> str: return ( f"ExcLogger<[{', '.join(map(lambda x: x.__name__, self.factories))}], " f"enabled = {str(self.is_enabled).lower()}>" ) log = ExcLogger.detect_loggers() PK/Nzjgqval/validator.pyfrom typing import Any, Callable, Union Predicate = Callable[[Any], bool] class QvalValidationError(Exception): """ An error raised on if validation fails. This exception should be used to provide custom validation error message to the frontend. Example: >>> from qval import validate >>> def f(v: str) -> bool: ... if not v.isnumeric(): ... raise QvalValidationError(f"Expected a number, got \'{v}\'") ... return True >>> params = validate({"number": "42"}, {"number": f}) >>> with params: pass # OK >>> with params.apply_to_request({"number": "a string"}): pass Traceback (most recent call last): ... qval.exceptions.InvalidQueryParamException: ... """ class Validator(object): """ Validates given value using provided predicates. .. automethod:: __call__ """ # :class:`Validator` implements __call__(Any) -> bool, # therefore can be treated in the same way as :type:`ValidatorType`. # For the sake of clarity, it is reflected in the class attributes below. ValidatorType = Union["Validator", Predicate] Predicate = ValidatorType def __init__(self, *predicates: Predicate): """ Instantiates the validator. :param predicates: predefined predicates :type predicates: Callable[[Any], bool] """ self.predicates = list(predicates) def add(self, predicate: Predicate) -> "Validator": """ Adds new predicate to the list. :param predicate: predicate function :return: self """ self.predicates.append(predicate) return self def __call__(self, value: Any) -> bool: """ Applies all stored predicates to the given value. :param value: value to validate :return: True if all checks have passed, False otherwise """ for p in self.predicates: if not p(value): return False return True PKTnMuX ::qval-0.3.1.dist-info/LICENSEThe MIT License (MIT) Copyright (c) 2018 OptimalStrategy Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. PK!Hd BUcqval-0.3.1.dist-info/WHEEL HM K-*ϳR03rOK-J,/RH,rzd&Y)r$[)T&UD"PK!H͍h(qval-0.3.1.dist-info/METADATAZmw6_RLN9uc'6q\iĆ"XH֦ )F" f3ϼ@U%#Y_TQ&: āɥ +z=F.R@\-R,uTJ^%*EUTĺ+U$&Z+Pzvs9^`4'բ^>UeUJ7wI >t3郞KhLZ$ wʰH@WgtVݫMrFSי*eĉaT$'[9KW E λWIM0aUR`r5YVTm?:Pauْ,tXP MU9"d4W+c'۔U!wI$z7z*%ԑ #yA…h4/d0|w󧤘 .6|c$9L|`2.Js\$*X9F]^wv8\V#:T\fe%ӔO$<hGY&Kx gO<uOe1(uH`O{3v3sZ\EC"3Sd@N?ᐴ&ON!L,Gy]85f;=0oU7Nȓ\XuE(3<) 3]-,q" ,ԢBʫR<.EYqK&/27 |ˀJlƋT,z, @¢d3E %KƓe*IZ, -8k_ 3ZN{{(>uAlr#e!̿cq397I&Hb%4+pmP"nT)1`q[fvUcSlxn2 baX'u"ړdQu7[sVVjqREbBC;j79otGy0k*XU\04S D3(8bKt$d%ͩ/e)CK7Md^9k,]C d[&SԠfz{lAduߞ `Mz=|sr%:Fߒ v־LgV WȸCl Jrl|-SYQO)P|ɬ\l wj: kyxRf·+ORI#-E-dk( 7Ca/dS8͘bk'ZRA)a\(UR;bHbrƱBɄ@N+a%qn|og}7&lQ8;?P)ΦbiJ(;tR l%窂]e-+K0 =C|xF ]U{2|V'LQ # r+9Y "L>e%RT8jG9$dk1[*w(^PYҮȐӝ9/&TOӥ@R/ 1Kr܎A2]+dXp"#aܗb4ˀ!V4GDT#glA \nHÍ߰4kZE὇f&cςڊ jɞ{1+O6*[E9h(пC9la!~ DM> 4('⠂Trl&9{M ٨@8BSҰD!Ⱗ8ӻ16ق~by/`A@A(ӄU.8J N+[L8 {lKÁM-3u8չߴ<.E;g]84feC1u_;B zEzM NNjּTj9I-,J~iɴKtm  1dvb> e<((P!cEZSmmw0(E#퐑*mGP;8qLhJ&6s3l~"c~S|Zjx '^L QC6. YA';@3IF6 |d©[rMC[ygt6nBgozֿzn{xYD( vc΢6W4cn;0)L@Z?(MxtAE|Ky⃡N o+.a]j9kk/`-;'@1>:~t{kح!=z-7"$,Ͱ@벜Ϙ\ZpX^k>S7>t&T+.+^,*%rO @EœXQ#)@vlCG7wSK{1*~]XцWsUsƽ")~%:ͅ=Slkx#EJ5soGR* eS'T>!*IKpj,JimK5D*T C4n.>o_:mVJΈt  \P*n1ٮGΦ:5%.\T.x%$¯o</`mNl32_R6o<¥)&1ҹ貍)g!bL,QҚy qQ%| ~ܐ·-!;EKJQI[\}J:o\r_Xz^jY% ;" ʸuzZ NIk 8{vs2tM,% V0i3ć7T@6Ŵ\[P&0kKwjlz5ïb`@%%+q&}ؠ⫹lFu]a .(؞h@0A TϞZ߁mq("pG{cXHȝ@\܀܌_<3?<۹r?hc'i *`C3ľlwKc4| `dȇGDG !*gg m,Sx|銋xӚ }һnx\[1}+ w:@CNzaޛ^{7:ӫb޳`ktzR͹};={vj|vG\0 ^KR