How to use _validate_error method in pandera

Best Python code snippet using pandera_python

pipeline_options_validator.py

Source:pipeline_options_validator.py

...117  def is_full_string_match(self, pattern, string):118    """Returns True if the pattern matches the whole string."""119    pattern = '^%s$' % pattern120    return re.search(pattern, string) is not None121  def _validate_error(self, err, *args):122    return [err % args]123  def validate_gcs_path(self, view, arg_name):124    """Validates a GCS path against gs://bucket/object URI format."""125    arg = getattr(view, arg_name, None)126    if arg is None:127      return self._validate_error(self.ERR_MISSING_GCS_PATH, arg_name)128    match = re.match(self.GCS_URI, arg, re.DOTALL)129    if match is None:130      return self._validate_error(self.ERR_INVALID_GCS_PATH, arg, arg_name)131    scheme = match.group('SCHEME')132    bucket = match.group('BUCKET')133    gcs_object = match.group('OBJECT')134    if ((scheme is None) or (scheme.lower() != self.GCS_SCHEME) or135        (bucket is None)):136      return self._validate_error(self.ERR_INVALID_GCS_PATH, arg, arg_name)137    if not self.is_full_string_match(self.GCS_BUCKET, bucket):138      return self._validate_error(self.ERR_INVALID_GCS_BUCKET, arg, arg_name)139    if gcs_object is None or '\n' in gcs_object or '\r' in gcs_object:140      return self._validate_error(self.ERR_INVALID_GCS_OBJECT, arg, arg_name)141    return []142  def validate_cloud_options(self, view):143    """Validates job_name and project arguments."""144    errors = []145    if (view.job_name and146        not self.is_full_string_match(self.JOB_PATTERN, view.job_name)):147      errors.extend(148          self._validate_error(self.ERR_INVALID_JOB_NAME, view.job_name))149    project = view.project150    if project is None:151      errors.extend(self._validate_error(self.ERR_MISSING_OPTION, 'project'))152    else:153      if self.is_full_string_match(self.PROJECT_NUMBER_PATTERN, project):154        errors.extend(155            self._validate_error(self.ERR_INVALID_PROJECT_NUMBER, project))156      elif not self.is_full_string_match(self.PROJECT_ID_PATTERN, project):157        errors.extend(158            self._validate_error(self.ERR_INVALID_PROJECT_ID, project))159    if view.update:160      if not view.job_name:161        errors.extend(162            self._validate_error(163                'Existing job name must be provided when updating a pipeline.'))164    if view.transform_name_mapping:165      if not view.update or not self.options.view_as(StandardOptions).streaming:166        errors.append(167            'Transform name mapping option is only useful when '168            '--update and --streaming is specified')169      for _, (key, value) in enumerate(view.transform_name_mapping.items()):170        if not isinstance(key, (str, unicode)) \171            or not isinstance(value, (str, unicode)):172          errors.extend(173              self._validate_error(174                  self.ERR_INVALID_TRANSFORM_NAME_MAPPING, key, value))175          break176    if view.region is None and self.is_service_runner():177      default_region = self.runner.get_default_gcp_region()178      if default_region is None:179        errors.extend(self._validate_error(self.ERR_MISSING_OPTION, 'region'))180      else:181        view.region = default_region182    return errors183  def validate_worker_region_zone(self, view):184    """Validates Dataflow worker region and zone arguments are consistent."""185    errors = []186    if view.zone and (view.worker_region or view.worker_zone):187      errors.extend(188          self._validate_error(189              'Cannot use deprecated flag --zone along with worker_region or '190              'worker_zone.'))191    if self.options.view_as(DebugOptions).lookup_experiment('worker_region')\192        and (view.worker_region or view.worker_zone):193      errors.extend(194          self._validate_error(195              'Cannot use deprecated experiment worker_region along with '196              'worker_region or worker_zone.'))197    if view.worker_region and view.worker_zone:198      errors.extend(199          self._validate_error(200              'worker_region and worker_zone are mutually exclusive.'))201    if view.zone:202      _LOGGER.warning(203          'Option --zone is deprecated. Please use --worker_zone instead.')204      view.worker_zone = view.zone205      view.zone = None206    return errors207  def validate_optional_argument_positive(self, view, arg_name):208    """Validates that an optional argument (if set) has a positive value."""209    arg = getattr(view, arg_name, None)210    if arg is not None and int(arg) <= 0:211      return self._validate_error(self.ERR_INVALID_NOT_POSITIVE, arg, arg_name)212    return []213  def validate_test_matcher(self, view, arg_name):214    """Validates that on_success_matcher argument if set.215    Validates that on_success_matcher is unpicklable and is instance216    of `hamcrest.core.base_matcher.BaseMatcher`.217    """218    # This is a test only method and requires hamcrest219    from hamcrest.core.base_matcher import BaseMatcher220    pickled_matcher = view.on_success_matcher221    errors = []222    try:223      matcher = pickler.loads(pickled_matcher)224      if not isinstance(matcher, BaseMatcher):225        errors.extend(226            self._validate_error(227                self.ERR_INVALID_TEST_MATCHER_TYPE, matcher, arg_name))228    except:  # pylint: disable=bare-except229      errors.extend(230          self._validate_error(231              self.ERR_INVALID_TEST_MATCHER_UNPICKLABLE,232              pickled_matcher,233              arg_name))...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.