PKN qval/__init__.py""" This module provides a convenient API for verifying query parameters. The core class is called `QueryParamValidator`. It accepts 4 arguments: - request: Request instance (Any object that has the following attributes: GET, query_params and body) - factories: A dictionary of factories `{param -> factory}`. The value of the parameter will be provided to the factory before validation. Any callable that accepts a string and returns anything is a valid factory. - validators: AD dictionary of validators `{param -> Validator}`. Validator is basically a list of predicates with the __call__() operator. See the `Validator` class for more info. - box_all: If true, adds all request parameters to the output collection. Otherwise, only parameters specified in `factories` will be added. If any parameter fails validation, `InvalidQueryParamException` (HTTP code = 400) will be raised. Also, only `TypeError`, `ValueError` and `KeyError` occurred after an argument was provided to the 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 dictionaries 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 the `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.3" 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 PKN>qval/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) 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,) 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 the 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 the 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 try: from flask import Request RequestType += (Request,) except ImportError: # pragma: no cover pass try: from falcon import Request RequestType += (Request,) except ImportError: # pragma: no cover pass if hasattr(module, "QVAL_MAKE_REQUEST_WRAPPER"): _make_request = load_symbol(module.QVAL_MAKE_REQUEST_WRAPPER) else: def _make_request(f): """ Wraps the default `utils.make_request()` function. Does nothing. """ return f def setup_flask_error_handlers(app: "flask.Flask"): # pragma: no cover """ Setups the 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 the 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] PKN\ي:: 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 the query validator. :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 the resulting dictionary if box_all is true. # Otherwise keep only the 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 the 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 the dictionary of query parameters. """ return self._query_params def add_predicate(self, param: str, predicate: Callable[[Any], bool]): """ Adds a new check for the 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 a reference to self def check( self, param: str, predicate: Callable[[Any], bool] ) -> "QueryParamValidator": """ Adds a new check for the 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 the :code:`greater than zero` comparison check for the 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 the :code:`greater than` comparison check for provided parameter. For example, if value = 10, :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 the `less than` comparison check for the provided parameter. For example, if value = 10, :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 the `equality` check for the provided parameter. For example, if value = 10, :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 the `nonzero` check for the provided parameter. For example, if value = 10, :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 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 the 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 to 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 """ 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 in the output dictionary, even if they're not specified in `factories` :param factories: factories that create python object from string parameters :return: QueryParamValidator instance """ # Convert the request so it will matches the required request interface if needed 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 argument (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 in the output dictionary, even if they're not specified in `factories` :param request_: optional request object will be always provided to the validator :return: wrapped function """ # Check if the 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) elif isinstance(args[0], fwk.RequestType): 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 especially handy in Flask, where `request` 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 PKNDAgN!N! 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 :class:`qval.framework_integration.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 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 a parameter and return an object that implements the 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 a 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 the 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 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__']}>" # XXX: Think about removing this 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 logger. :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 the 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 the built objects. :param name: logger name :param level: logging level. If a 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 a 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 the 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() PKNaGqval/validator.pyfrom typing import Any, Callable, Union Predicate = Callable[[Any], bool] class QvalValidationError(Exception): """ An error raised if validation fails. This exception should be used to provide a custom validation error message to the client. 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 the given value using provided predicates. .. automethod:: __call__ """ # :class:`Validator` implements __call__(Any) -> bool, and # 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 the 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.3.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.3.dist-info/WHEEL HM K-*ϳR03rOK-J,/RH,rzd&Y)r$[)T&UD"PK!H&Ji(qval-0.3.3.dist-info/METADATAZw6[K92G۫9uc'ֱ]iĚ"X>$k )?i#3oGUPrNbB KxB롸DžXJrCU).U'"ҹXgJ|*d}ÙSj!d(#Uy :~n_3+}9lw^DE b8W'0#yUHOݰĠCYal(xCyܼ QoOnN1K>?)ٛ̕\'joP9A2 WddXU‡Po*(`*ÙYK%ئ([ $M"̼^V/U- Z;,W_g21FvFx=gUYz6Ja<1zpBo2@Isř4@!RO]@xR I@U)H44o0}j&L,no"d,ľt8|[0]B8@ĖPEǒq4,N(_|8UTL<^dD^E} pNR#]m|Ԣ.#昸bS$6K7R [b$t6&> إq$8fP L %%?0KTߏ u AmN`#FD,ԡNAH{rxk ZXm"niHLv`Zm|^VpNBOXs#<UBs0i*TfGLc x d˜9Eۼ7#V^khL =ӢmQldL}#/c YSN?_(oѷbHz;mE\K}dHT:Gbwg?%yrt-SiQfsog/>wdZTl wjm:*yxBܷ+OYRA#E `ks 7{@ae S8IbcZRNia+eTɺObrڱBɄ@JJaŖqnjgg}&ጵQͱ8;?P L'baJ(ZtR l%g]-+ 0 =}E|xZs]%U{2|RLQ5#sr<^[D2Q>JQ9i)TZ _LxdAB;E"v@t8al4x1z .D:`ݽ~vxqXvJF7 %Z â5 lXoS2 v-_|)|Lo /0@$ODE}-40+5vT:1x N5[/0$;xXP`S[REٱ{ID%`KP4 7#-LUY?8DO~AΩ6DT0JWS>gtb|CYwuGhJ6c!([t'rx6}*o x7ǖ'1l;5 BW9'@q)ԽR۳Yce d5;xt)x:PٻbQ8\|@ۆ祈: 3ԭl]`Ѱ/fп>ξk˷Xh!@*@q)?wxW_tR-qرE!_2snj''(C)Q sii31q"dTaNa"66;݋٢vk{WP%{i8&U9b?S1|>@S-i\{ndߠ'gPh%}CKJݢvZj5RPS-UkaJRDKydVkTk,+oywO[4LLKg#oI|qoGѱOZޮ/^K.|G0:S7>u9l./MILa}{REW]_2)F[=ל$/RPU70%c&lFBlܪ}(IW8)K cJ _x'6%y''^R 9}ò n7[yM_x'TY!6"8U +-ȫ^gޕԌb/㏇?/t|u=p| X 3kR>T[.+^2栓rO @ǫyœXQ#)@vl}G7wS {1*~]XфWsUsQƣ"u)|%:=SlݫCEJ5swGR* e=R'C |AZTe{)Lw^75&pэ(-sk6/ ܟ+.Ub-Ah\_}߾ڬgm&l0 СUPQcI3?k#]F Mq)7+tbJP ?.]"~[+8H&˹_jx/_ $f6 RelxKSLgcPueSŘj}Y%tY5⎣67^Jto?=YTUBC<"NkTW+vW ^<ZVa aK$TpBI]7)N#mA7q(ۮAdD$&q㚜J^ QP<_Kv}smNYz:M_ ¸`:\d;6pj.a}sv ':ML^:Gbuu0:3T;-ENhn xA.ҁ=[rk(nW[ 7';[_G tk ox]簡Db]h@[d10k2C"#N3|{>Mj| /\q/_ޭM!g !F^wu Qɻӷ`}[st( JX/b{+fgz9^}?Vt^#snNGn<Ξi<Ѻ(&;WRGz'j s_z߾{1?PK!HKqval-0.3.3.dist-info/RECORDmIs@}~ $t/@L6T  2Og ֹ˯i'T~EE+N7ʼnP_Ig*oŴG911,hP:yMݿP<]ezr_l+4m֢\@M<ұCUziӭڐf/ed-]>X\@Z,YimYX+/8I!}@hVI狩 -s *"P&Arp:PތKH|"LFO)qggr Nt/Ҋs M$&K ݞP(H;aPۓXPC"Ov X˪S(7.EԲr]u(6vGJ4hRUI څO՘Y6knuHy@@4UOR$Oz )nw Lv3јfzb㘄w&_duUy*PKN qval/__init__.pyPK0Mw%..qval/exceptions.pyPKN>qval/framework_integration.pyPKN\ي:: 0qval/qval.pyPKNDAgN!N! 6kqval/utils.pyPKNaGqval/validator.pyPKTnMuX ::qval-0.3.3.dist-info/LICENSEPK!Hd BUcZqval-0.3.3.dist-info/WHEELPK!H&Ji(qval-0.3.3.dist-info/METADATAPK!HKqval-0.3.3.dist-info/RECORDPK