Source code for rudra.utils.validation

"""Validation Utility module."""

import urllib.request as request
import xmlrpc.client as xmlrpclib
from rudra import logger
import re
_canonicalize_regex = re.compile(r"[-_.]+")


[docs]def nn(name): """Return a normalized name.""" # This is taken from PEP 503. return _canonicalize_regex.sub("-", name).lower()
[docs]def check_field_exists(input_data, fields): """Check field exist in the input data.""" if isinstance(input_data, dict): for field in fields: if not input_data.get(field): logger.error( "Please provide the valid value for the field {}".format(field)) if isinstance(input_data, (list, dict, set, frozenset)): return list(set(fields).difference(set(input_data))) raise ValueError
[docs]def check_url_alive(url, accept_codes=[401]): """Validate github repo exist or not.""" try: logger.info("checking url is alive", extra={"url": url}) response = request.urlopen(url) status_code = response.getcode() if status_code in accept_codes or status_code // 100 in (2, 3): return True except Exception as exc: logger.debug("Unable to reach url", extra={"exception": str(exc)}) return False
[docs]class BQValidation: """Add validation for ecosystems.""" def __init__(self): """Initialize the BQValidation object.""" pypi_org = xmlrpclib.ServerProxy('https://pypi.python.org/pypi') self.pypi_org_packages = {nn(p) for p in pypi_org.list_packages()}
[docs] def validate_pypi(self, content): """Validate python packages. Attributes: content (:obj:`str` or [:obj:`str`] or {:obj:`str`}): list/set of packages or package str Returns: [:obj:`str`]: list of valid packages. Raises: ValueError: if content is not a type of :obj:`str` or :obj:`list` """ if not isinstance(content, (str, list, set, frozenset)): raise ValueError("content type should be string or set/list of string") content = [content] if isinstance(content, str) else content return list(self.pypi_org_packages.intersection(content))