""" Base repository class for data access. """ from typing import TypeVar, Generic, List, Dict, Any, Optional, Type from django.db import models from django.db.models import Q, QuerySet T = TypeVar('T', bound=models.Model) class BaseRepository(Generic[T]): """ Base repository class with common methods for data access. """ model: Type[T] = None @classmethod def get_by_id(cls, entity_id: str) -> Optional[T]: """ Get an entity by ID. Args: entity_id: The entity ID Returns: The entity or None if not found """ try: return cls.model.objects.get(pk=entity_id) except cls.model.DoesNotExist: return None @classmethod def get_all(cls, **filters) -> QuerySet[T]: """ Get all entities with optional filtering. Args: **filters: Filter parameters Returns: QuerySet of matching entities """ queryset = cls.model.objects.all() # Apply filters for key, value in filters.items(): if value is not None: # Handle special filter keys if key.endswith('__in') and not value: # Empty list for __in lookup should return empty queryset return cls.model.objects.none() # Apply the filter queryset = queryset.filter(**{key: value}) return queryset @classmethod def create(cls, data: Dict[str, Any]) -> T: """ Create a new entity. Args: data: Entity data Returns: The created entity """ return cls.model.objects.create(**data) @classmethod def update(cls, entity_id: str, data: Dict[str, Any]) -> Optional[T]: """ Update an existing entity. Args: entity_id: The entity ID data: Updated data Returns: The updated entity or None if not found """ obj = cls.get_by_id(entity_id) if not obj: return None for key, value in data.items(): setattr(obj, key, value) obj.save() return obj @classmethod def delete(cls, entity_id: str) -> bool: """ Delete an entity. Args: entity_id: The entity ID Returns: True if deleted, False if not found """ obj = cls.get_by_id(entity_id) if not obj: return False obj.delete() return True @classmethod def bulk_create(cls, data_list: List[Dict[str, Any]]) -> List[T]: """ Create multiple entities. Args: data_list: List of entity data Returns: List of created entities """ objects = [cls.model(**data) for data in data_list] return cls.model.objects.bulk_create(objects) @classmethod def bulk_update(cls, objects: List[T], fields: List[str]) -> int: """ Update multiple entities. Args: objects: List of entity objects fields: List of fields to update Returns: Number of updated entities """ # Cast the objects to Any to bypass type checking for this call return cls.model.objects.bulk_update(objects, fields) # type: ignore @classmethod def count(cls, **filters) -> int: """ Count entities with optional filtering. Args: **filters: Filter parameters Returns: Count of matching entities """ return cls.get_all(**filters).count() @classmethod def exists(cls, **filters) -> bool: """ Check if any entities exist with optional filtering. Args: **filters: Filter parameters Returns: True if entities exist, False otherwise """ return cls.get_all(**filters).exists() @classmethod def filter_by_date_range(cls, start_date=None, end_date=None, date_field='date') -> QuerySet[T]: """ Filter entities by date range. Args: start_date: Start date (inclusive) end_date: End date (inclusive) date_field: Name of the date field to filter on Returns: QuerySet of matching entities """ filters = {} if start_date: filters[f'{date_field}__gte'] = start_date if end_date: filters[f'{date_field}__lte'] = end_date return cls.get_all(**filters) @classmethod def search(cls, search_term: str, search_fields: List[str]) -> QuerySet[T]: """ Search entities by term across multiple fields. Args: search_term: The search term search_fields: List of fields to search in Returns: QuerySet of matching entities """ if not search_term: return cls.model.objects.all() q_objects = Q() for field in search_fields: q_objects |= Q(**{f'{field}__icontains': search_term}) return cls.model.objects.filter(q_objects)