mirror of
https://github.com/pacnpal/thrillwiki_django_no_react.git
synced 2025-12-20 12:11:13 -05:00
274 lines
8.8 KiB
Python
274 lines
8.8 KiB
Python
"""
|
|
Custom managers and QuerySets for optimized database patterns.
|
|
Following Django styleguide best practices for database access.
|
|
"""
|
|
|
|
from typing import Optional, List, Union
|
|
from django.db import models
|
|
from django.db.models import Q, Count, Avg, Max
|
|
# from django.contrib.gis.geos import Point # Disabled temporarily for setup
|
|
# from django.contrib.gis.measure import Distance # Disabled temporarily for setup
|
|
from django.utils import timezone
|
|
from datetime import timedelta
|
|
|
|
|
|
class BaseQuerySet(models.QuerySet):
|
|
"""Base QuerySet with common optimizations and patterns."""
|
|
|
|
def active(self):
|
|
"""Filter for active/enabled records."""
|
|
if hasattr(self.model, "is_active"):
|
|
return self.filter(is_active=True)
|
|
return self
|
|
|
|
def published(self):
|
|
"""Filter for published records."""
|
|
if hasattr(self.model, "is_published"):
|
|
return self.filter(is_published=True)
|
|
return self
|
|
|
|
def recent(self, *, days: int = 30):
|
|
"""Filter for recently created records."""
|
|
cutoff_date = timezone.now() - timedelta(days=days)
|
|
return self.filter(created_at__gte=cutoff_date)
|
|
|
|
def search(self, *, query: str, fields: Optional[List[str]] = None):
|
|
"""
|
|
Full-text search across specified fields.
|
|
|
|
Args:
|
|
query: Search query string
|
|
fields: List of field names to search (defaults to name, description)
|
|
"""
|
|
if not query:
|
|
return self
|
|
|
|
if fields is None:
|
|
fields = ["name", "description"] if hasattr(self.model, "name") else []
|
|
|
|
q_objects = Q()
|
|
for field in fields:
|
|
if hasattr(self.model, field):
|
|
q_objects |= Q(**{f"{field}__icontains": query})
|
|
|
|
return self.filter(q_objects) if q_objects else self
|
|
|
|
def with_stats(self):
|
|
"""Add basic statistics annotations."""
|
|
return self
|
|
|
|
def optimized_for_list(self):
|
|
"""Optimize queryset for list display."""
|
|
return self.select_related().prefetch_related()
|
|
|
|
def optimized_for_detail(self):
|
|
"""Optimize queryset for detail display."""
|
|
return self.select_related().prefetch_related()
|
|
|
|
|
|
class BaseManager(models.Manager):
|
|
"""Base manager with common patterns."""
|
|
|
|
def get_queryset(self):
|
|
return BaseQuerySet(self.model, using=self._db)
|
|
|
|
def active(self):
|
|
return self.get_queryset().active()
|
|
|
|
def published(self):
|
|
return self.get_queryset().published()
|
|
|
|
def recent(self, *, days: int = 30):
|
|
return self.get_queryset().recent(days=days)
|
|
|
|
def search(self, *, query: str, fields: Optional[List[str]] = None):
|
|
return self.get_queryset().search(query=query, fields=fields)
|
|
|
|
|
|
class LocationQuerySet(BaseQuerySet):
|
|
"""QuerySet for location-based models with geographic functionality."""
|
|
|
|
def near_point(self, *, point, distance_km: float = 50): # Point type disabled for setup
|
|
"""Filter locations near a geographic point."""
|
|
if hasattr(self.model, "point"):
|
|
return (
|
|
self.filter(point__distance_lte=(point, Distance(km=distance_km)))
|
|
.distance(point)
|
|
.order_by("distance")
|
|
)
|
|
return self
|
|
|
|
def within_bounds(self, *, north: float, south: float, east: float, west: float):
|
|
"""Filter locations within geographic bounds."""
|
|
if hasattr(self.model, "point"):
|
|
return self.filter(
|
|
point__latitude__gte=south,
|
|
point__latitude__lte=north,
|
|
point__longitude__gte=west,
|
|
point__longitude__lte=east,
|
|
)
|
|
return self
|
|
|
|
def by_country(self, *, country: str):
|
|
"""Filter by country."""
|
|
if hasattr(self.model, "country"):
|
|
return self.filter(country__iexact=country)
|
|
return self
|
|
|
|
def by_region(self, *, state: str):
|
|
"""Filter by state/region."""
|
|
if hasattr(self.model, "state"):
|
|
return self.filter(state__iexact=state)
|
|
return self
|
|
|
|
def by_city(self, *, city: str):
|
|
"""Filter by city."""
|
|
if hasattr(self.model, "city"):
|
|
return self.filter(city__iexact=city)
|
|
return self
|
|
|
|
|
|
class LocationManager(BaseManager):
|
|
"""Manager for location-based models."""
|
|
|
|
def get_queryset(self):
|
|
return LocationQuerySet(self.model, using=self._db)
|
|
|
|
def near_point(self, *, point, distance_km: float = 50): # Point type disabled for setup
|
|
return self.get_queryset().near_point(point=point, distance_km=distance_km)
|
|
|
|
def within_bounds(self, *, north: float, south: float, east: float, west: float):
|
|
return self.get_queryset().within_bounds(
|
|
north=north, south=south, east=east, west=west
|
|
)
|
|
|
|
|
|
class ReviewableQuerySet(BaseQuerySet):
|
|
"""QuerySet for models that can be reviewed."""
|
|
|
|
def with_review_stats(self):
|
|
"""Add review statistics annotations."""
|
|
return self.annotate(
|
|
review_count=Count("reviews", filter=Q(reviews__is_published=True)),
|
|
average_rating=Avg("reviews__rating", filter=Q(reviews__is_published=True)),
|
|
latest_review_date=Max(
|
|
"reviews__created_at", filter=Q(reviews__is_published=True)
|
|
),
|
|
)
|
|
|
|
def highly_rated(self, *, min_rating: float = 8.0):
|
|
"""Filter for highly rated items."""
|
|
return self.with_review_stats().filter(average_rating__gte=min_rating)
|
|
|
|
def recently_reviewed(self, *, days: int = 30):
|
|
"""Filter for items with recent reviews."""
|
|
cutoff_date = timezone.now() - timedelta(days=days)
|
|
return self.filter(
|
|
reviews__created_at__gte=cutoff_date, reviews__is_published=True
|
|
).distinct()
|
|
|
|
|
|
class ReviewableManager(BaseManager):
|
|
"""Manager for reviewable models."""
|
|
|
|
def get_queryset(self):
|
|
return ReviewableQuerySet(self.model, using=self._db)
|
|
|
|
def with_review_stats(self):
|
|
return self.get_queryset().with_review_stats()
|
|
|
|
def highly_rated(self, *, min_rating: float = 8.0):
|
|
return self.get_queryset().highly_rated(min_rating=min_rating)
|
|
|
|
|
|
class HierarchicalQuerySet(BaseQuerySet):
|
|
"""QuerySet for hierarchical models (with parent/child relationships)."""
|
|
|
|
def root_level(self):
|
|
"""Filter for root-level items (no parent)."""
|
|
if hasattr(self.model, "parent"):
|
|
return self.filter(parent__isnull=True)
|
|
return self
|
|
|
|
def children_of(self, *, parent_id: int):
|
|
"""Get children of a specific parent."""
|
|
if hasattr(self.model, "parent"):
|
|
return self.filter(parent_id=parent_id)
|
|
return self
|
|
|
|
def with_children_count(self):
|
|
"""Add count of children."""
|
|
if hasattr(self.model, "children"):
|
|
return self.annotate(children_count=Count("children"))
|
|
return self
|
|
|
|
|
|
class HierarchicalManager(BaseManager):
|
|
"""Manager for hierarchical models."""
|
|
|
|
def get_queryset(self):
|
|
return HierarchicalQuerySet(self.model, using=self._db)
|
|
|
|
def root_level(self):
|
|
return self.get_queryset().root_level()
|
|
|
|
|
|
class TimestampedQuerySet(BaseQuerySet):
|
|
"""QuerySet for models with created_at/updated_at timestamps."""
|
|
|
|
def created_between(self, *, start_date, end_date):
|
|
"""Filter by creation date range."""
|
|
return self.filter(created_at__date__range=[start_date, end_date])
|
|
|
|
def updated_since(self, *, since_date):
|
|
"""Filter for records updated since a date."""
|
|
return self.filter(updated_at__gte=since_date)
|
|
|
|
def by_creation_date(self, *, descending: bool = True):
|
|
"""Order by creation date."""
|
|
order = "-created_at" if descending else "created_at"
|
|
return self.order_by(order)
|
|
|
|
|
|
class TimestampedManager(BaseManager):
|
|
"""Manager for timestamped models."""
|
|
|
|
def get_queryset(self):
|
|
return TimestampedQuerySet(self.model, using=self._db)
|
|
|
|
def created_between(self, *, start_date, end_date):
|
|
return self.get_queryset().created_between(
|
|
start_date=start_date, end_date=end_date
|
|
)
|
|
|
|
|
|
class StatusQuerySet(BaseQuerySet):
|
|
"""QuerySet for models with status fields."""
|
|
|
|
def with_status(self, *, status: Union[str, List[str]]):
|
|
"""Filter by status."""
|
|
if isinstance(status, list):
|
|
return self.filter(status__in=status)
|
|
return self.filter(status=status)
|
|
|
|
def operating(self):
|
|
"""Filter for operating/active status."""
|
|
return self.filter(status="OPERATING")
|
|
|
|
def closed(self):
|
|
"""Filter for closed status."""
|
|
return self.filter(status__in=["CLOSED_TEMP", "CLOSED_PERM"])
|
|
|
|
|
|
class StatusManager(BaseManager):
|
|
"""Manager for status-based models."""
|
|
|
|
def get_queryset(self):
|
|
return StatusQuerySet(self.model, using=self._db)
|
|
|
|
def operating(self):
|
|
return self.get_queryset().operating()
|
|
|
|
def closed(self):
|
|
return self.get_queryset().closed()
|