try:
from dissolve import replace_me
except ModuleNotFoundError:
import warnings
def replace_me(since=None, remove_in=None, message=None):
def decorator(func):
def wrapper(*args, **kwargs):
msg = f"{func.__name__} has been deprecated"
if since:
msg += f" since {since}"
if remove_in:
msg += f" and will be removed in {remove_in}"
if message:
msg += f". {message}"
else:
msg += ". Consider running 'dissolve migrate' to automatically update your code."
warnings.warn(msg, DeprecationWarning, stacklevel=2)
return func(*args, **kwargs)
return wrapper
return decorator
class ProcessingEngine:
def __init__(self, engine_type="standard"):
self.engine_type = engine_type
self.processed_count = 0
@classmethod
def create_fast_engine(cls):
return cls("fast")
@classmethod
def create_memory_efficient_engine(cls):
return cls("memory_efficient")
@staticmethod
def validate_input(data):
if not isinstance(data, (list, tuple, str)):
raise ValueError("Input must be a list, tuple, or string")
return True
@staticmethod
def normalize_text(text, lowercase=True, strip_whitespace=True):
result = text
if strip_whitespace:
result = result.strip()
if lowercase:
result = result.lower()
return result
def process(self, data):
self.validate_input(data)
self.processed_count += 1
return f"Processed by {self.engine_type} engine: {data}"
@classmethod
@replace_me(since="2.0.0", remove_in="3.0.0")
def old_create_engine(cls, fast=False):
if fast:
return cls.create_fast_engine()
else:
return cls.create_memory_efficient_engine()
@classmethod
@replace_me(since="1.0.0")
def legacy_fast_processor(cls):
return cls.create_fast_engine()
@staticmethod
@replace_me(since="1.5.0", remove_in="2.5.0")
def old_validate_data(input_data):
return ProcessingEngine.validate_input(input_data)
@staticmethod
@replace_me(since="2.1.0")
def legacy_clean_text(text_data):
return ProcessingEngine.normalize_text(text_data, lowercase=True, strip_whitespace=True)
@staticmethod
@replace_me(since="1.0.0")
def old_process_batch(items, uppercase=False, trim=True):
return ProcessingEngine.process_text_batch(
text_items=items,
strip_whitespace=trim,
lowercase=not uppercase
)
@staticmethod
def process_text_batch(text_items, strip_whitespace=True, lowercase=True):
results = []
for item in text_items:
processed = ProcessingEngine.normalize_text(
str(item),
lowercase=lowercase,
strip_whitespace=strip_whitespace
)
results.append(processed)
return results
class DataValidator:
@staticmethod
def is_valid_email(email):
return "@" in email and "." in email.split("@")[-1]
@staticmethod
def is_valid_phone(phone, country_code=None):
cleaned = "".join(c for c in phone if c.isdigit())
if country_code:
return len(cleaned) >= 10 and cleaned.startswith(country_code.lstrip("+"))
return len(cleaned) >= 10
@classmethod
def create_validator(cls, strict=False):
validator = cls()
validator.strict_mode = strict
return validator
@staticmethod
@replace_me(since="1.3.0", remove_in="2.0.0")
def check_email(email_addr):
return DataValidator.is_valid_email(email_addr)
@staticmethod
@replace_me(since="1.6.0")
def validate_phone_number(number):
return DataValidator.is_valid_phone(number, country_code=None)
@classmethod
@replace_me(since="1.0.0")
def create_strict_validator(cls):
return cls.create_validator(strict=True)