在Django中,QuerySet是数据库查询的主要接口,它表示数据库中对象的集合,并允许你构建复杂的数据库查询。扩展QuerySet意味着为它添加自定义的方法或功能,以便在整个项目中复用查询逻辑。
from django.db import models
class PublishedQuerySet(models.QuerySet):
def published(self):
return self.filter(status='published')
def by_author(self, author):
return self.filter(author=author)
def recent(self, days=7):
return self.filter(publish_date__gte=timezone.now() - timedelta(days=days))
class Article(models.Model):
title = models.CharField(max_length=200)
content = models.TextField()
status = models.CharField(max_length=20, choices=[
('draft', 'Draft'),
('published', 'Published'),
])
author = models.ForeignKey(User, on_delete=models.CASCADE)
publish_date = models.DateTimeField(auto_now_add=True)
objects = PublishedQuerySet.as_manager()
class CustomManager(models.Manager):
def get_queryset(self):
return CustomQuerySet(self.model, using=self._db)
def active(self):
return self.get_queryset().active()
def with_comments(self):
return self.get_queryset().with_comments()
class CustomQuerySet(models.QuerySet):
def active(self):
return self.filter(is_active=True)
def with_comments(self):
return self.annotate(comment_count=Count('comments'))
class Post(models.Model):
title = models.CharField(max_length=100)
content = models.TextField()
is_active = models.BooleanField(default=True)
objects = CustomManager()
class ProductQuerySet(models.QuerySet):
def available(self):
return self.filter(stock__gt=0)
def expensive(self):
return self.filter(price__gte=100)
class ProductManager(models.Manager):
def get_queryset(self):
return ProductQuerySet(self.model, using=self._db)
@classmethod
def create_product(cls, name, price):
return cls.create(name=name, price=price)
class Product(models.Model):
name = models.CharField(max_length=100)
price = models.DecimalField(max_digits=10, decimal_places=2)
stock = models.IntegerField(default=0)
objects = ProductManager()
class AdvancedQuerySet(models.QuerySet):
def for_user(self, user):
return self.filter(user=user)
def public(self):
return self.filter(is_public=True)
def recent(self):
return self.order_by('-created_at')
def popular(self):
return self.order_by('-views')
# 使用示例: Post.objects.for_user(request.user).public().recent()
class StatsQuerySet(models.QuerySet):
@property
def stats(self):
return {
'count': self.count(),
'avg_price': self.aggregate(avg=Avg('price'))['avg'],
'max_price': self.aggregate(max=Max('price'))['max'],
}
class Item(models.Model):
name = models.CharField(max_length=100)
price = models.DecimalField(max_digits=10, decimal_places=2)
objects = StatsQuerySet.as_manager()
# 使用示例: Item.objects.filter(category='books').stats
class DynamicFilterQuerySet(models.QuerySet):
def __getattr__(self, name):
if name.startswith('filter_by_'):
field = name[10:]
def filter_func(value):
return self.filter(**{field: value})
return filter_func
raise AttributeError(f"'{self.__class__.__name__}' object has no attribute '{name}'")
class DynamicModel(models.Model):
name = models.CharField(max_length=100)
age = models.IntegerField()
is_active = models.BooleanField(default=True)
objects = DynamicFilterQuerySet.as_manager()
# 使用示例: DynamicModel.objects.filter_by_name('John').filter_by_age(30)
现象:自定义方法后无法继续链式调用其他方法
原因:方法没有返回QuerySet实例
解决:确保所有自定义方法返回self或新的QuerySet
# 错误示例
def bad_method(self):
self.filter(...) # 没有return
# 正确示例
def good_method(self):
return self.filter(...)
现象:自定义方法导致N+1查询问题
解决:使用select_related或prefetch_related优化
class OptimizedQuerySet(models.QuerySet):
def with_related(self):
return self.select_related('author').prefetch_related('comments')
现象:IDE无法识别自定义方法
解决:添加类型提示
from typing import TypeVar, Any
T = TypeVar('T', bound='Model')
class TypedQuerySet(models.QuerySet[T]):
def active(self) -> 'TypedQuerySet[T]':
return self.filter(is_active=True)
通过合理扩展QuerySet,可以显著提高Django项目的代码质量和开发效率。
没有搜到相关的文章