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 DataProcessor:
def __init__(self, data_source, batch_size=100, enable_caching=True):
self.data_source = data_source
self.batch_size = batch_size
self.enable_caching = enable_caching
self._cache = {} if enable_caching else None
def process(self, data):
if self._cache is not None and data in self._cache:
return self._cache[data]
result = f"Processed: {data} (batch_size={self.batch_size})"
if self._cache is not None:
self._cache[data] = result
return result
def batch_process(self, data_list):
return [self.process(item) for item in data_list]
def get_stats(self):
cache_size = len(self._cache) if self._cache else 0
return {
"data_source": self.data_source,
"batch_size": self.batch_size,
"caching_enabled": self.enable_caching,
"cache_size": cache_size
}
class User:
def __init__(self, username, email, full_name=None, is_active=True):
self.username = username
self.email = email
self.full_name = full_name or username
self.is_active = is_active
self._profile = {}
def get_display_name(self):
return self.full_name or self.username
def update_profile(self, **kwargs):
self._profile.update(kwargs)
return self._profile
def get_profile(self):
return self._profile.copy()
@property
def status(self):
return "active" if self.is_active else "inactive"
@replace_me(since="2.0.0", remove_in="3.0.0")
class OldDataProcessor:
def __init__(self, data_source, cache_size=50):
self._processor = DataProcessor(
data_source=data_source,
batch_size=cache_size * 2, enable_caching=True
)
def process(self, data):
return self._processor.process(data)
def get_info(self):
return self._processor.get_stats()
class LegacyUser:
def __init__(self, name, email_address, active=True):
self._user = User(username=name, email=email_address, is_active=active)
@replace_me(since="1.5.0", remove_in="2.5.0")
def get_name(self):
return self._user.get_display_name()
@replace_me(since="1.8.0")
def set_profile_data(self, data):
return self._user.update_profile(**data)
def get_display_name(self):
return self._user.get_display_name()
def update_profile(self, **kwargs):
return self._user.update_profile(**kwargs)
@property
@replace_me(since="1.0.0")
def is_enabled(self):
return self._user.status == "active"
@property
def status(self):
return self._user.status
@replace_me(since="1.0.0")
class SimpleProcessor:
def __init__(self, source):
self._processor = DataProcessor(source, batch_size=1, enable_caching=False)
def process_item(self, item):
return self._processor.process(item)