Django's Manager and QuerySet system provides a powerful abstraction for database operations. Understanding how to create custom managers and querysets enables you to build reusable, chainable query logic that keeps your code DRY and maintainable.
# models.py
from django.db import models
class Post(models.Model):
title = models.CharField(max_length=200)
content = models.TextField()
status = models.CharField(max_length=20, default='draft')
is_published = models.BooleanField(default=False)
created_at = models.DateTimeField(auto_now_add=True)
# Django automatically adds: objects = models.Manager()
# Usage of default manager
all_posts = Post.objects.all()
published_posts = Post.objects.filter(is_published=True)
post_count = Post.objects.count()
# models.py
from django.db import models
from django.utils import timezone
class PublishedManager(models.Manager):
"""Manager that returns only published posts"""
def get_queryset(self):
return super().get_queryset().filter(
is_published=True,
status='published'
)
class FeaturedManager(models.Manager):
"""Manager for featured content"""
def get_queryset(self):
return super().get_queryset().filter(
is_published=True,
is_featured=True
)
class Post(models.Model):
title = models.CharField(max_length=200)
content = models.TextField()
status = models.CharField(
max_length=20,
choices=[
('draft', 'Draft'),
('published', 'Published'),
('archived', 'Archived')
],
default='draft'
)
is_published = models.BooleanField(default=False)
is_featured = models.BooleanField(default=False)
created_at = models.DateTimeField(auto_now_add=True)
published_at = models.DateTimeField(null=True, blank=True)
# Multiple managers
objects = models.Manager() # Default manager
published = PublishedManager() # Published posts only
featured = FeaturedManager() # Featured posts only
def save(self, *args, **kwargs):
if self.is_published and not self.published_at:
self.published_at = timezone.now()
super().save(*args, **kwargs)
# Usage
all_posts = Post.objects.all() # All posts including drafts
published_posts = Post.published.all() # Only published posts
featured_posts = Post.featured.all() # Only featured posts
# models.py
from django.db import models
from django.utils import timezone
from datetime import timedelta
class AdvancedPostManager(models.Manager):
"""Manager with custom methods"""
def published(self):
"""Get published posts"""
return self.filter(
is_published=True,
published_at__lte=timezone.now()
)
def drafts(self):
"""Get draft posts"""
return self.filter(status='draft')
def recent(self, days=30):
"""Get recent posts"""
cutoff_date = timezone.now() - timedelta(days=days)
return self.filter(created_at__gte=cutoff_date)
def popular(self, min_views=1000):
"""Get popular posts"""
return self.filter(view_count__gte=min_views)
def by_author(self, author):
"""Get posts by specific author"""
return self.filter(author=author)
def in_category(self, category):
"""Get posts in specific category"""
return self.filter(category=category)
def with_tag(self, tag_name):
"""Get posts with specific tag"""
return self.filter(tags__name=tag_name)
def search(self, query):
"""Search posts by title and content"""
from django.db.models import Q
return self.filter(
Q(title__icontains=query) | Q(content__icontains=query)
)
def create_post(self, title, content, author, **kwargs):
"""Custom creation method"""
post = self.create(
title=title,
content=content,
author=author,
**kwargs
)
# Custom logic after creation
# e.g., send notifications, update caches, etc.
return post
class Post(models.Model):
title = models.CharField(max_length=200)
content = models.TextField()
author = models.ForeignKey('auth.User', on_delete=models.CASCADE)
category = models.ForeignKey('Category', on_delete=models.SET_NULL, null=True)
tags = models.ManyToManyField('Tag', blank=True)
status = models.CharField(max_length=20, default='draft')
is_published = models.BooleanField(default=False)
view_count = models.PositiveIntegerField(default=0)
created_at = models.DateTimeField(auto_now_add=True)
published_at = models.DateTimeField(null=True, blank=True)
objects = AdvancedPostManager()
# Usage
recent_published = Post.objects.published().recent(7)
popular_drafts = Post.objects.drafts().popular(500)
author_posts = Post.objects.by_author(user).published()
search_results = Post.objects.search('django').published()
# models.py
class BaseManager(models.Manager):
"""Base manager with common functionality"""
def active(self):
"""Get active records"""
return self.filter(is_active=True)
def inactive(self):
"""Get inactive records"""
return self.filter(is_active=False)
def recent(self, days=30):
"""Get recent records"""
cutoff_date = timezone.now() - timedelta(days=days)
return self.filter(created_at__gte=cutoff_date)
class PostManager(BaseManager):
"""Post-specific manager extending base functionality"""
def published(self):
return self.active().filter(status='published')
def featured(self):
return self.published().filter(is_featured=True)
class ProductManager(BaseManager):
"""Product-specific manager extending base functionality"""
def in_stock(self):
return self.active().filter(stock_quantity__gt=0)
def on_sale(self):
return self.active().filter(sale_price__isnull=False)
class Post(models.Model):
title = models.CharField(max_length=200)
status = models.CharField(max_length=20, default='draft')
is_featured = models.BooleanField(default=False)
is_active = models.BooleanField(default=True)
created_at = models.DateTimeField(auto_now_add=True)
objects = PostManager()
class Product(models.Model):
name = models.CharField(max_length=200)
stock_quantity = models.PositiveIntegerField(default=0)
sale_price = models.DecimalField(max_digits=10, decimal_places=2, null=True, blank=True)
is_active = models.BooleanField(default=True)
created_at = models.DateTimeField(auto_now_add=True)
objects = ProductManager()
# Usage
active_posts = Post.objects.active()
featured_posts = Post.objects.featured()
recent_products = Product.objects.recent(7)
sale_products = Product.objects.on_sale()
# models.py
from django.db import models
class PostQuerySet(models.QuerySet):
"""Custom QuerySet with chainable methods"""
def published(self):
return self.filter(status='published', is_published=True)
def drafts(self):
return self.filter(status='draft')
def featured(self):
return self.filter(is_featured=True)
def by_author(self, author):
return self.filter(author=author)
def in_category(self, category):
return self.filter(category=category)
def recent(self, days=30):
cutoff_date = timezone.now() - timedelta(days=days)
return self.filter(created_at__gte=cutoff_date)
def popular(self, min_views=1000):
return self.filter(view_count__gte=min_views)
def search(self, query):
from django.db.models import Q
return self.filter(
Q(title__icontains=query) |
Q(content__icontains=query) |
Q(tags__name__icontains=query)
).distinct()
def with_comments(self):
return self.filter(comments__isnull=False).distinct()
def without_comments(self):
return self.filter(comments__isnull=True)
class PostManager(models.Manager):
def get_queryset(self):
return PostQuerySet(self.model, using=self._db)
# Proxy methods to make manager methods available
def published(self):
return self.get_queryset().published()
def featured(self):
return self.get_queryset().featured()
def search(self, query):
return self.get_queryset().search(query)
class Post(models.Model):
title = models.CharField(max_length=200)
content = models.TextField()
author = models.ForeignKey('auth.User', on_delete=models.CASCADE)
category = models.ForeignKey('Category', on_delete=models.SET_NULL, null=True)
tags = models.ManyToManyField('Tag', blank=True)
status = models.CharField(max_length=20, default='draft')
is_published = models.BooleanField(default=False)
is_featured = models.BooleanField(default=False)
view_count = models.PositiveIntegerField(default=0)
created_at = models.DateTimeField(auto_now_add=True)
objects = PostManager()
# Usage - all methods are chainable
recent_featured = Post.objects.published().featured().recent(7)
popular_by_author = Post.objects.by_author(user).popular().published()
search_results = Post.objects.search('django').published().recent(30)
# models.py
class AdvancedPostQuerySet(models.QuerySet):
"""Advanced QuerySet with complex operations"""
def published(self):
return self.filter(
status='published',
is_published=True,
published_at__lte=timezone.now()
)
def scheduled(self):
"""Posts scheduled for future publication"""
return self.filter(
status='published',
published_at__gt=timezone.now()
)
def with_stats(self):
"""Annotate posts with calculated statistics"""
from django.db.models import Count, Avg, F
return self.annotate(
comment_count=Count('comments'),
avg_rating=Avg('ratings__score'),
engagement_score=F('view_count') + F('like_count') * 2
)
def popular_this_month(self):
"""Popular posts from current month"""
from django.db.models import Count
start_of_month = timezone.now().replace(day=1, hour=0, minute=0, second=0, microsecond=0)
return self.filter(
created_at__gte=start_of_month
).annotate(
total_engagement=Count('comments') + Count('likes')
).filter(
total_engagement__gte=10
).order_by('-total_engagement')
def by_tag_popularity(self):
"""Order by tag popularity"""
from django.db.models import Count
return self.annotate(
tag_count=Count('tags__posts')
).order_by('-tag_count')
def with_related(self):
"""Optimize queries with select_related and prefetch_related"""
return self.select_related(
'author',
'category'
).prefetch_related(
'tags',
'comments__author'
)
def for_sitemap(self):
"""Optimized queryset for sitemap generation"""
return self.published().only(
'slug',
'updated_at'
).order_by('-updated_at')
def bulk_publish(self):
"""Bulk publish draft posts"""
return self.filter(status='draft').update(
status='published',
is_published=True,
published_at=timezone.now()
)
def archive_old_posts(self, days=365):
"""Archive posts older than specified days"""
cutoff_date = timezone.now() - timedelta(days=days)
return self.filter(
created_at__lt=cutoff_date,
status='published'
).update(status='archived')
class Post(models.Model):
# ... field definitions ...
objects = PostManager.from_queryset(AdvancedPostQuerySet)()
# Usage
popular_posts = Post.objects.with_stats().popular_this_month()
optimized_posts = Post.objects.published().with_related()
sitemap_posts = Post.objects.for_sitemap()
# Bulk operations
Post.objects.filter(author=user).bulk_publish()
Post.objects.archive_old_posts(days=730) # Archive 2-year-old posts
# querysets.py - Separate file for reusable querysets
from django.db import models
from django.utils import timezone
from datetime import timedelta
class TimestampedQuerySet(models.QuerySet):
"""Base queryset for models with timestamps"""
def recent(self, days=30):
cutoff_date = timezone.now() - timedelta(days=days)
return self.filter(created_at__gte=cutoff_date)
def older_than(self, days=30):
cutoff_date = timezone.now() - timedelta(days=days)
return self.filter(created_at__lt=cutoff_date)
def this_week(self):
start_of_week = timezone.now() - timedelta(days=timezone.now().weekday())
return self.filter(created_at__gte=start_of_week)
def this_month(self):
start_of_month = timezone.now().replace(day=1, hour=0, minute=0, second=0, microsecond=0)
return self.filter(created_at__gte=start_of_month)
class PublishableQuerySet(models.QuerySet):
"""Base queryset for publishable content"""
def published(self):
return self.filter(
is_published=True,
published_at__lte=timezone.now()
)
def unpublished(self):
return self.filter(is_published=False)
def scheduled(self):
return self.filter(
is_published=True,
published_at__gt=timezone.now()
)
class PostQuerySet(TimestampedQuerySet, PublishableQuerySet):
"""Composed queryset inheriting from multiple base querysets"""
def featured(self):
return self.filter(is_featured=True)
def by_author(self, author):
return self.filter(author=author)
def in_category(self, category):
return self.filter(category=category)
class EventQuerySet(TimestampedQuerySet, PublishableQuerySet):
"""Events with similar functionality"""
def upcoming(self):
return self.filter(event_date__gte=timezone.now())
def past(self):
return self.filter(event_date__lt=timezone.now())
# models.py
class Post(models.Model):
# ... field definitions ...
objects = models.Manager.from_queryset(PostQuerySet)()
class Event(models.Model):
# ... field definitions ...
objects = models.Manager.from_queryset(EventQuerySet)()
# Usage - both models inherit common functionality
recent_posts = Post.objects.recent(7).published()
recent_events = Event.objects.recent(7).upcoming()
this_month_posts = Post.objects.this_month().featured()
# models.py
class OptimizedPostQuerySet(models.QuerySet):
"""Performance-optimized queryset methods"""
def with_minimal_data(self):
"""Load only essential fields"""
return self.only(
'id', 'title', 'slug', 'created_at', 'author_id'
)
def with_author_info(self):
"""Efficiently load author information"""
return self.select_related('author').only(
'id', 'title', 'slug', 'created_at',
'author__username', 'author__first_name', 'author__last_name'
)
def with_full_data(self):
"""Load all related data efficiently"""
return self.select_related(
'author',
'category'
).prefetch_related(
'tags',
models.Prefetch(
'comments',
queryset=Comment.objects.select_related('author').filter(is_approved=True)
)
)
def for_api(self):
"""Optimized for API responses"""
return self.select_related('author', 'category').prefetch_related('tags').only(
'id', 'title', 'slug', 'excerpt', 'created_at', 'view_count',
'author__username', 'category__name'
)
def popular_with_stats(self):
"""Get popular posts with engagement statistics"""
from django.db.models import Count, F, Case, When, IntegerField
return self.annotate(
comment_count=Count('comments', filter=models.Q(comments__is_approved=True)),
like_count=Count('likes'),
engagement_score=F('view_count') + F('comment_count') * 5 + F('like_count') * 2,
popularity_tier=Case(
When(engagement_score__gte=1000, then=models.Value('high')),
When(engagement_score__gte=100, then=models.Value('medium')),
default=models.Value('low'),
output_field=models.CharField()
)
).filter(engagement_score__gte=50).order_by('-engagement_score')
def bulk_update_views(self, post_ids, increment=1):
"""Efficiently update view counts"""
from django.db.models import F
return self.filter(id__in=post_ids).update(
view_count=F('view_count') + increment
)
class Post(models.Model):
# ... field definitions ...
objects = models.Manager.from_queryset(OptimizedPostQuerySet)()
# Usage
# For list views
post_list = Post.objects.published().with_minimal_data()
# For detail views
post_detail = Post.objects.with_full_data().get(slug=slug)
# For API endpoints
api_posts = Post.objects.published().for_api()
# Bulk operations
Post.objects.bulk_update_views([1, 2, 3, 4, 5], increment=1)
Custom managers and querysets provide a clean, reusable way to encapsulate complex database logic. They promote code reuse, improve maintainability, and enable you to build expressive, chainable APIs for your data access layer.
Filtering, Ordering, and Slicing
Django's QuerySet API provides powerful tools for filtering, ordering, and slicing data. Mastering these techniques is essential for building efficient database queries and creating responsive applications.
Aggregation
Django's aggregation framework provides powerful tools for performing calculations across multiple database records. Understanding aggregation functions, grouping, and annotation enables you to generate reports, statistics, and analytical data efficiently at the database level.