QuerySet class as a descriptor

Author: davegaeddert

plain-cache/plain/cache/chores.py

@@ -8,5 +8,5 @@ def clear_expired() -> str:
     """
     Delete cache items that have expired.
     """
-    result = CachedItem.query.expired().delete()  # type: ignore[attr-defined]
+    result = CachedItem.query.expired().delete()
     return f"{result[0]} expired cache items deleted"

plain-cache/plain/cache/cli.py

@@ -14,7 +14,7 @@ def cli() -> None:
 @cli.command()
 def clear_expired() -> None:
     click.echo("Clearing expired cache items...")
-    result = CachedItem.query.expired().delete()  # type: ignore[attr-defined]
+    result = CachedItem.query.expired().delete()
     click.echo(f"Deleted {result[0]} expired cache items.")
 
 
@@ -33,9 +33,9 @@ def clear_all(force: bool) -> None:
 @cli.command()
 def stats() -> None:
     total = CachedItem.query.count()
-    expired = CachedItem.query.expired().count()  # type: ignore[attr-defined]
-    unexpired = CachedItem.query.unexpired().count()  # type: ignore[attr-defined]
-    forever = CachedItem.query.forever().count()  # type: ignore[attr-defined]
+    expired = CachedItem.query.expired().count()
+    unexpired = CachedItem.query.unexpired().count()
+    forever = CachedItem.query.forever().count()
 
     click.echo(f"Total: {click.style(total, bold=True)}")
     click.echo(f"Expired: {click.style(expired, bold=True)}")

plain-cache/plain/cache/models.py

@@ -1,17 +1,19 @@
 from __future__ import annotations
 
+from typing import Self
+
 from plain import models
 from plain.utils import timezone
 
 
-class CachedItemQuerySet(models.QuerySet):
-    def expired(self) -> CachedItemQuerySet:
+class CachedItemQuerySet(models.QuerySet["CachedItem"]):
+    def expired(self) -> Self:
         return self.filter(expires_at__lt=timezone.now())
 
-    def unexpired(self) -> CachedItemQuerySet:
+    def unexpired(self) -> Self:
         return self.filter(expires_at__gte=timezone.now())
 
-    def forever(self) -> CachedItemQuerySet:
+    def forever(self) -> Self:
         return self.filter(expires_at=None)
 
 
@@ -23,8 +25,9 @@ class CachedItem(models.Model):
     created_at = models.DateTimeField(auto_now_add=True)
     updated_at = models.DateTimeField(auto_now=True)
 
+    query = CachedItemQuerySet()
+
     class Meta:
-        queryset_class = CachedItemQuerySet
         indexes = [
             models.Index(fields=["expires_at"]),
         ]

plain-models/plain/models/README.md

@@ -213,69 +213,52 @@ class User(models.Model):
 
 ## Custom QuerySets
 
-With the Manager functionality now merged into QuerySet, you can customize [`QuerySet`](./query.py#QuerySet) classes to provide specialized query methods. There are several ways to use custom QuerySets:
+With the Manager functionality now merged into QuerySet, you can customize [`QuerySet`](./query.py#QuerySet) classes to provide specialized query methods.
 
-### Setting a default QuerySet for a model
-
-Use `Meta.queryset_class` to set a custom QuerySet that will be used by `Model.query`:
+Define a custom QuerySet and assign it to your model's `query` attribute:
 
 ```python
-class PublishedQuerySet(models.QuerySet):
-    def published_only(self):
+from typing import Self
+
+class PublishedQuerySet(models.QuerySet["Article"]):
+    def published_only(self) -> Self:
         return self.filter(status="published")
 
-    def draft_only(self):
+    def draft_only(self) -> Self:
         return self.filter(status="draft")
 
 @models.register_model
 class Article(models.Model):
     title = models.CharField(max_length=200)
     status = models.CharField(max_length=20)
 
-    class Meta:
-        queryset_class = PublishedQuerySet
+    query = PublishedQuerySet()
 
-# Usage - all methods available on Article.objects
+# Usage - all methods available on Article.query
 all_articles = Article.query.all()
 published_articles = Article.query.published_only()
 draft_articles = Article.query.draft_only()
 ```
 
-### Using custom QuerySets without formal attachment
-
-You can also use custom QuerySets manually without setting them as the default:
+Custom methods can be chained with built-in QuerySet methods:
 
 ```python
-class SpecialQuerySet(models.QuerySet):
-    def special_filter(self):
-        return self.filter(special=True)
-
-# Create and use the QuerySet manually
-special_qs = SpecialQuerySet(model=Article)
-special_articles = special_qs.special_filter()
+# Chaining works naturally
+recent_published = Article.query.published_only().order_by("-created_at")[:10]
 ```
 
-### Using classmethods for convenience
+### Programmatic QuerySet usage
 
-For even cleaner API, add classmethods to your model:
+For internal code that needs to create QuerySet instances programmatically, use `from_model()`:
 
 ```python
-@models.register_model
-class Article(models.Model):
-    title = models.CharField(max_length=200)
-    status = models.CharField(max_length=20)
-
-    @classmethod
-    def published(cls):
-        return PublishedQuerySet(model=cls).published_only()
-
-    @classmethod
-    def drafts(cls):
-        return PublishedQuerySet(model=cls).draft_only()
+class SpecialQuerySet(models.QuerySet["Article"]):
+    def special_filter(self) -> Self:
+        return self.filter(special=True)
 
-# Usage
-published_articles = Article.published()
-draft_articles = Article.drafts()
+# Create and use the QuerySet programmatically
+special_qs = SpecialQuerySet.from_model(Article)
+special_articles = special_qs.special_filter()
 ```
 
 ## Forms

plain-models/plain/models/base.py

@@ -122,11 +122,6 @@ def _setup_meta(cls) -> None:
 
         Options(meta, package_label).contribute_to_class(cls, "_meta")
 
-    @property
-    def query(cls) -> QuerySet:
-        """Create a new QuerySet for this model."""
-        return cls._meta.queryset
-
 
 class ModelStateFieldsCacheDescriptor:
     def __get__(
@@ -159,6 +154,9 @@ class Model(metaclass=ModelBase):
     # Every model gets an automatic id field
     id = PrimaryKeyField()
 
+    # QuerySet descriptor for model queries
+    query = QuerySet()
+
     def __init__(self, *args: Any, **kwargs: Any):
         # Alias some things as locals to avoid repeat global lookups
         cls = self.__class__

plain-models/plain/models/fields/related_managers.py

@@ -70,8 +70,6 @@ def __init__(self, instance: Any, rel: Any):
         self.instance = instance
         self.field = rel.field
         self.core_filters = {self.field.name: instance}
-        # Store the base queryset class for this model
-        self.base_queryset_class = rel.related_model._meta.queryset.__class__
         self.allow_null = rel.field.allow_null
 
     def _check_fk_val(self) -> None:
@@ -137,15 +135,14 @@ def get_queryset(self) -> QuerySet:
                 self.field.remote_field.get_cache_name()
             ]
         except (AttributeError, KeyError):
-            # Use the base queryset class for this model
-            queryset = self.base_queryset_class(model=self.model)
+            queryset = self.model.query
             return self._apply_rel_filters(queryset)
 
     def get_prefetch_queryset(
         self, instances: Any, queryset: QuerySet | None = None
     ) -> tuple[QuerySet, Any, Any, bool, str, bool]:
         if queryset is None:
-            queryset = self.base_queryset_class(model=self.model)
+            queryset = self.model.query
 
         rel_obj_attr = self.field.get_local_related_value
         instance_attr = self.field.get_foreign_related_value
@@ -196,17 +193,17 @@ def check_and_update_obj(obj: Any) -> None:
     def create(self, **kwargs: Any) -> Any:
         self._check_fk_val()
         kwargs[self.field.name] = self.instance
-        return self.base_queryset_class(model=self.model).create(**kwargs)
+        return self.model.query.create(**kwargs)
 
     def get_or_create(self, **kwargs: Any) -> tuple[Any, bool]:
         self._check_fk_val()
         kwargs[self.field.name] = self.instance
-        return self.base_queryset_class(model=self.model).get_or_create(**kwargs)
+        return self.model.query.get_or_create(**kwargs)
 
     def update_or_create(self, **kwargs: Any) -> tuple[Any, bool]:
         self._check_fk_val()
         kwargs[self.field.name] = self.instance
-        return self.base_queryset_class(model=self.model).update_or_create(**kwargs)
+        return self.model.query.update_or_create(**kwargs)
 
     def remove(self, *objs: Any, bulk: bool = True) -> None:
         # remove() is only provided if the ForeignKey can have a value of null
@@ -297,8 +294,6 @@ class BaseManyToManyManager(BaseRelatedManager):
     def __init__(self, instance: Any, rel: Any):
         self.instance = instance
         self.through = rel.through
-        # Subclasses must set model before calling super().__init__
-        self.base_queryset_class = self.model._meta.queryset.__class__
 
         self.source_field = self.through._meta.get_field(self.source_field_name)
         self.target_field = self.through._meta.get_field(self.target_field_name)
@@ -338,14 +333,14 @@ def get_queryset(self) -> QuerySet:
         try:
             return self.instance._prefetched_objects_cache[self.prefetch_cache_name]
         except (AttributeError, KeyError):
-            queryset = self.base_queryset_class(model=self.model)
+            queryset = self.model.query
             return self._apply_rel_filters(queryset)
 
     def get_prefetch_queryset(
         self, instances: Any, queryset: QuerySet | None = None
     ) -> tuple[QuerySet, Any, Any, bool, str, bool]:
         if queryset is None:
-            queryset = self.base_queryset_class(model=self.model)
+            queryset = self.model.query
 
         queryset = _filter_prefetch_queryset(
             queryset._next_is_sticky(), self.query_field_name, instances
@@ -380,9 +375,7 @@ def get_prefetch_queryset(
     def clear(self) -> None:
         with transaction.atomic(savepoint=False):
             self._remove_prefetched_objects()
-            filters = self._build_remove_filters(
-                self.base_queryset_class(model=self.model)
-            )
+            filters = self._build_remove_filters(self.model.query)
             self.through.query.filter(filters).delete()
 
     def set(
@@ -425,16 +418,14 @@ def set(
     def create(
         self, *, through_defaults: dict[str, Any] | None = None, **kwargs: Any
     ) -> Any:
-        new_obj = self.base_queryset_class(model=self.model).create(**kwargs)
+        new_obj = self.model.query.create(**kwargs)
         self.add(new_obj, through_defaults=through_defaults)
         return new_obj
 
     def get_or_create(
         self, *, through_defaults: dict[str, Any] | None = None, **kwargs: Any
     ) -> tuple[Any, bool]:
-        obj, created = self.base_queryset_class(model=self.model).get_or_create(
-            **kwargs
-        )
+        obj, created = self.model.query.get_or_create(**kwargs)
         # We only need to add() if created because if we got an object back
         # from get() then the relationship already exists.
         if created:
@@ -444,9 +435,7 @@ def get_or_create(
     def update_or_create(
         self, *, through_defaults: dict[str, Any] | None = None, **kwargs: Any
     ) -> tuple[Any, bool]:
-        obj, created = self.base_queryset_class(model=self.model).update_or_create(
-            **kwargs
-        )
+        obj, created = self.model.query.update_or_create(**kwargs)
         # We only need to add() if created because if we got an object back
         # from get() then the relationship already exists.
         if created:
@@ -534,7 +523,7 @@ def _remove_items(
                 old_ids.add(obj)
 
         with transaction.atomic(savepoint=False):
-            target_model_qs = self.base_queryset_class(model=self.model)
+            target_model_qs = self.model.query
             if target_model_qs._has_filters():
                 old_vals = target_model_qs.filter(
                     **{f"{self.target_field.target_field.attname}__in": old_ids}

plain-models/plain/models/options.py

@@ -29,7 +29,6 @@
 DEFAULT_NAMES = (
     "db_table",
     "db_table_comment",
-    "queryset_class",
     "ordering",
     "package_label",
     "models_registry",
@@ -53,7 +52,6 @@ class Options:
         "_non_pk_concrete_field_names",
         "_forward_fields_map",
         "base_queryset",
-        "queryset",
     }
     REVERSE_PROPERTIES = {"related_objects", "fields_map", "_relation_tree"}
 
@@ -63,7 +61,6 @@ def __init__(self, meta: Any, package_label: str | None = None):
         self._get_fields_cache: dict[tuple[bool, bool, bool, bool], Any] = {}
         self.local_fields: list[Field] = []
         self.local_many_to_many: list[Field] = []
-        self.queryset_class: type[QuerySet] | None = None
         self.model_name: str | None = None
         self.db_table: str = ""
         self.db_table_comment: str = ""
@@ -220,13 +217,7 @@ def base_queryset(self) -> QuerySet:
         objects), the base queryset must never filter out rows to prevent
         incomplete results in related queries.
         """
-        return QuerySet(model=self.model)
-
-    @property
-    def queryset(self) -> QuerySet:
-        if self.queryset_class:
-            return self.queryset_class(model=self.model)
-        return QuerySet(model=self.model)
+        return QuerySet.from_model(self.model)
 
     @cached_property
     def fields(self) -> ImmutableList: