""" Enterprise-Ready Model Managers Module This module provides custom model managers that extend Django's default manager functionality with enterprise-level features including caching, filtering, searching, and performance optimizations. Features: - Base managers with common query patterns - Caching managers for performance optimization - Search and filtering capabilities - Soft delete management - Audit trail management - Performance monitoring and optimization - Bulk operations with validation - Advanced querying methods Author: Focus Development Team Version: 2.0.0 License: Proprietary """ import logging from datetime import datetime, timedelta from typing import Optional, List, Dict, Any, Union from django.conf import settings from django.utils import timezone from django.core.cache import cache from django.db import models, transaction from django.contrib.auth import get_user_model from django.core.exceptions import ValidationError from django.db.models import Q, Count, Avg, Sum, Max, Min logger = logging.getLogger(__name__) class ActiveManager(models.Manager): """ Custom manager that excludes soft deleted records by default. This manager automatically filters out records where is_deleted=True, providing a clean interface for working with active records only. """ def get_queryset(self): """ Return queryset excluding soft deleted records. Returns: QuerySet: All objects where is_deleted=False """ return super().get_queryset().filter(is_deleted=False) def with_deleted(self): """ Return queryset including soft deleted records. Returns: QuerySet: All objects including soft deleted ones """ return super().get_queryset() def deleted_only(self): """ Return queryset with only soft deleted records. Returns: QuerySet: Only objects where is_deleted=True """ return super().get_queryset().filter(is_deleted=True) class BaseQuerySet(models.QuerySet): """ Base QuerySet with common filtering and optimization methods. This QuerySet provides standard filtering capabilities and performance optimizations that are commonly needed across different models in the application. """ def active(self): """Return only active objects.""" return self.filter(is_active=True) def inactive(self): """Return only inactive objects.""" return self.filter(is_active=False) def published(self): """Return only published objects.""" return self.filter(is_published=True) def unpublished(self): """Return only unpublished objects.""" return self.filter(is_published=False) def recent(self, days: int = 30): """Return objects created in the last N days.""" cutoff_date = timezone.now() - timedelta(days=days) return self.filter(created_at__gte=cutoff_date) def older_than(self, days: int = 30): """Return objects older than N days.""" cutoff_date = timezone.now() - timedelta(days=days) return self.filter(created_at__lt=cutoff_date) def by_user(self, user): """Return objects created by a specific user.""" return self.filter(created_by=user) def by_status(self, status: str): """Return objects with a specific status.""" return self.filter(status=status) def search(self, query: str, fields: List[str] = None): """ Perform text search across specified fields. Args: query: Search query string fields: List of fields to search in Returns: Filtered QuerySet """ if not query or not fields: return self search_filter = Q() for field in fields: search_filter |= Q(**{f"{field}__icontains": query}) return self.filter(search_filter) def with_annotations(self): """Add common annotations to the queryset.""" return self.annotate( total_count=Count('id'), created_date=models.functions.TruncDate('created_at') ) def optimized(self): """Apply common optimizations to the queryset.""" return self.select_related().prefetch_related() class BaseManager(models.Manager): """ Base manager class providing common functionality. This manager provides standard methods and optimizations that are commonly needed across different models. """ def get_queryset(self): """Return the base queryset.""" return BaseQuerySet(self.model, using=self._db) def active(self): """Return only active objects.""" return self.get_queryset().active() def inactive(self): """Return only inactive objects.""" return self.get_queryset().inactive() def published(self): """Return only published objects.""" return self.get_queryset().published() def recent(self, days: int = 30): """Return objects created in the last N days.""" return self.get_queryset().recent(days) def by_user(self, user): """Return objects created by a specific user.""" return self.get_queryset().by_user(user) def search(self, query: str, fields: List[str] = None): """Perform text search across specified fields.""" return self.get_queryset().search(query, fields) def bulk_create_with_validation( self, objs: List[models.Model], batch_size: int = None, ignore_conflicts: bool = False ): """ Bulk create objects with validation. Args: objs: List of model instances to create batch_size: Number of objects to create in each batch ignore_conflicts: Whether to ignore conflicts Returns: List of created objects """ # Validate objects before bulk creation for obj in objs: obj.full_clean() return self.bulk_create( objs, batch_size=batch_size, ignore_conflicts=ignore_conflicts ) def get_or_none(self, **kwargs): """ Get an object or return None if not found. Args: **kwargs: Lookup parameters Returns: Model instance or None """ try: return self.get(**kwargs) except self.model.DoesNotExist: return None def exists_by(self, **kwargs) -> bool: """ Check if an object exists with given parameters. Args: **kwargs: Lookup parameters Returns: Boolean indicating existence """ return self.filter(**kwargs).exists() class SoftDeleteQuerySet(BaseQuerySet): """ QuerySet for models with soft delete functionality. This QuerySet automatically excludes soft-deleted objects from standard queries while providing methods to access deleted objects when needed. """ def delete(self): """Perform soft delete on all objects in queryset.""" return self.update( is_deleted=True, deleted_at=timezone.now() ) def hard_delete(self): """Permanently delete objects from database.""" return super().delete() def restore(self): """Restore soft-deleted objects.""" return self.update( is_deleted=False, deleted_at=None, deleted_by=None ) class SoftDeleteManager(BaseManager): """ Manager for models with soft delete functionality. This manager automatically excludes soft-deleted objects from standard queries while providing methods to access deleted objects. """ def get_queryset(self): """Override to exclude soft-deleted objects by default.""" return SoftDeleteQuerySet(self.model, using=self._db).filter(is_deleted=False) def deleted(self): """Return only soft-deleted objects.""" return SoftDeleteQuerySet(self.model, using=self._db).filter(is_deleted=True) def with_deleted(self): """Return all objects including soft-deleted ones.""" return SoftDeleteQuerySet(self.model, using=self._db) def hard_delete(self): """Permanently delete objects from database.""" return SoftDeleteQuerySet(self.model, using=self._db).hard_delete() class CachedManager(BaseManager): """ Manager with caching capabilities for improved performance. This manager provides automatic caching of frequently accessed objects and query results to improve application performance. """ cache_timeout = 60 * 15 # 15 minutes default cache_prefix = 'cached_manager' def _get_cache_key(self, key: str) -> str: """Generate cache key for the given key.""" return f"{self.cache_prefix}:{self.model._meta.label}:{key}" def get_cached(self, cache_key: str, **kwargs): """ Get an object from cache or database. Args: cache_key: Key to use for caching **kwargs: Lookup parameters Returns: Model instance or None """ full_cache_key = self._get_cache_key(cache_key) obj = cache.get(full_cache_key) if obj is None: try: obj = self.get(**kwargs) cache.set(full_cache_key, obj, self.cache_timeout) except self.model.DoesNotExist: obj = None return obj def invalidate_cache(self, cache_key: str): """Invalidate cached object.""" full_cache_key = self._get_cache_key(cache_key) cache.delete(full_cache_key) def get_cached_list(self, cache_key: str, queryset_method: str = None, **kwargs): """ Get a list of objects from cache or database. Args: cache_key: Key to use for caching queryset_method: Method to call on queryset **kwargs: Filter parameters Returns: List of model instances """ full_cache_key = self._get_cache_key(cache_key) objects = cache.get(full_cache_key) if objects is None: queryset = self.filter(**kwargs) if queryset_method: queryset = getattr(queryset, queryset_method)() objects = list(queryset) cache.set(full_cache_key, objects, self.cache_timeout) return objects class AuditManager(BaseManager): """ Manager for audit log functionality. This manager provides methods for creating and querying audit log entries with proper filtering and optimization. """ def log_action(self, action: str, user, obj: models.Model = None, changes: Dict = None, ip_address: str = None, user_agent: str = None) -> 'AuditLog': """ Create an audit log entry. Args: action: Action performed user: User who performed the action obj: Object affected by the action changes: Dictionary of changes made ip_address: IP address of the user user_agent: User agent string Returns: Created AuditLog instance """ from django.contrib.contenttypes.models import ContentType audit_data = { 'action': action, 'created_by': user, 'changes': changes or {}, 'ip_address': ip_address, 'user_agent': user_agent, } if obj: audit_data.update({ 'object_type': ContentType.objects.get_for_model(obj), 'object_id': str(obj.pk), 'object_repr': str(obj), }) return self.create(**audit_data) def by_action(self, action: str): """Return audit logs for a specific action.""" return self.filter(action=action) def by_user(self, user): """Return audit logs for a specific user.""" return self.filter(created_by=user) def by_object(self, obj: models.Model): """Return audit logs for a specific object.""" from django.contrib.contenttypes.models import ContentType content_type = ContentType.objects.get_for_model(obj) return self.filter(object_type=content_type, object_id=str(obj.pk)) def by_date_range(self, start_date: datetime, end_date: datetime): """Return audit logs within a date range.""" return self.filter(created_at__range=[start_date, end_date]) def user_activity_summary(self, user, days: int = 30): """ Get user activity summary for the last N days. Args: user: User to get activity for days: Number of days to look back Returns: Dictionary with activity statistics """ cutoff_date = timezone.now() - timedelta(days=days) activity = self.filter( created_by=user, created_at__gte=cutoff_date ).values('action').annotate( count=Count('id') ).order_by('-count') return { 'total_actions': sum(item['count'] for item in activity), 'actions_by_type': list(activity), 'period_days': days, }