"""Validator base class.""" import re class BaseValidator: """Validator base class. It contains methods to validate SQL identifies. We have to introduce custom validation, because we can't bind db identifiers (db, schema, column names, etc.) within cursor.execute() method. """ def __init__(self): """Create a validator entity.""" # e.g., load some config with valid names to validate params against pass def if_alnum_or_underscore_only(self, s): """A simple check to validate SQL identifiers. Args: s (str): An SQL identifier to validate (e.g., db name, etc). Returns: s (str): A passed string if valid. Raises: Exception: If string contains unacceptable symbols. """ if not re.match(r'^[a-zA-Z_0-9_]*$', s): raise Exception( 'There is a bad symbol in a SQL parameter {}, abort ' 'operation'.format(s)) else: return s # TODO (borisuvarov): Check if escaping of quotes works properly def validate_file_format(self, file_format): """Validate FILE_FORMAT Snowflake statement. Args: file_format (list): Parts of a FILE_FORMAT statement. Raises: ValueError: If validation fails. """ for option in file_format: if re.match(r'^SKIP_HEADER=\d+$', option): continue elif (re.match(r'^DATE_FORMAT="\w{4}-\w{2}-\w{2}"$', option) or option == 'DATE_FORMAT="AUTO"'): continue elif (re.match( r'^TIMESTAMP_FORMAT="\w{4}-\w{2}-\w{2}\s\w{2}:\w{2}:' r'\w{2}\.\w{6}"$', option) or re.match( r'^TIMESTAMP_FORMAT="\w{4}-\w{2}-\w{2}T\w{2}:\w{2}:' r'\w{2}"$', option) or option == 'TIMESTAMP_FORMAT="AUTO"'): continue elif re.match(r'^TYPE="(CSV|AVRO|XML|JSON)"$', option): continue elif re.match( r'^COMPRESSION="(AUTO|GZIP|BZ2|DEFLATE|RAW_DEFLATE|' r'NONE)"$', option): continue elif re.match(r'^ESCAPE="\\134"$', option): continue elif re.match(r'^ESCAPE_UNENCLOSED_FIELD="\\134"$', option): continue elif re.match(r'^RECORD_DELIMITER="(\\n|\\r|NONE)"$', option): continue elif re.match(r'^FIELD_DELIMITER="(,|;|\||\\t|\s|NONE)"$', option): continue elif re.match( r'^NULL_IF=(\("__NULL__"\)|\("\w{4}-\w{2}-\w{2}"\)|' r'\("NULL"\)|\(\))$', option): continue elif re.match( r'^FIELD_OPTIONALLY_ENCLOSED_BY="(\'|\"|NONE)"$', option): continue elif re.match(r"^FIELD_OPTIONALLY_ENCLOSED_BY='(\"|NONE)'$", option): continue elif re.match(r'^TRIM_SPACE=(TRUE|FALSE)$', option): continue elif re.match( r'^ERROR_ON_COLUMN_COUNT_MISMATCH=(TRUE|FALSE)$', option): continue elif re.match(r'^ENABLE_OCTAL=(TRUE|FALSE)$', option): continue elif re.match(r'^ALLOW_DUPLICATE=(TRUE|FALSE)$', option): continue elif re.match(r'^STRIP_OUTER_ARRAY=(TRUE|FALSE)$', option): continue elif re.match(r'^STRIP_NULL_VALUES=(TRUE|FALSE)$', option): continue elif re.match(r'^IGNORE_UTF8_ERRORS=(TRUE|FALSE)$', option): continue elif re.match(r'^PRESERVE_SPACE=(TRUE|FALSE)$', option): continue elif re.match(r'^STRIP_OUTER_ELEMENT=(TRUE|FALSE)$', option): continue elif re.match(r'^DISABLE_SNOWFLAKE_DATA=(TRUE|FALSE)$', option): continue elif re.match(r'^DISABLE_AUTO_CONVERT=(TRUE|FALSE)$', option): continue else: raise ValueError( 'FILE_FORMAT is not valid, {} is not correct'.format( option))