209 lines
5.1 KiB
Python
209 lines
5.1 KiB
Python
"""
|
|
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) |