Apply row scoping when relations load lazily, and add orm.optimize (v0.16.0).

Security: a resolver returning materialized rows instead of a query object
silently disabled row-level access control on every relation below it. The
optimizer applies scope_rows while building the eager load, so with nothing to
build, Django and SQLAlchemy read the relation straight off each parent and
never scoped it. A single list() call exposed rows the schema was written to
hide, in both directions of the relation. Both backends now reapply the
related type's scope_rows and the field-level scope= at resolve time, which is
what Tortoise already did; Tortoise gains the same for to-one relations. A
relation the optimizer already loaded was scoped on the way in and is returned
from cache, so the fast path costs nothing.

Two tests had encoded the leak as a guarantee, asserting that disabling the
optimizer disabled the scoping hooks. They now assert the opposite.

New: orm.optimize(data, info, at=...) eager-loads what the selection needs from
rows a resolver has already materialized, loading relations onto the instances
given so in-memory values survive. orm.connection(resolver=...) supplies the
rows for a connection while the library still builds the filter/order/groupBy
arguments and the grouped connection type.

Fixes: grouped connections never worked on Tortoise - post-processing was
synchronous while the backend's grouping is async, so groups received an
un-awaited coroutine. Tortoise also folded an inherited ORDER BY into the
GROUP BY of an aggregate, turning a total into a per-row count. Tortoise
relation resolvers re-queried prefetched relations, discarding nested eager
loads; fixing that resolves a known-broken forward-FK case.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: jamietdavidson

README.md

@@ -335,13 +335,13 @@ How a field is written decides what scoping it gets.
 | `@orm.field.custom` returning a query object | Optimizer + that type's `scope_rows` — see [Root custom query](#root-custom-query) |
 | `@orm.field.custom` returning `self.author` | As written; scoping only via prefetch |
 | `@strawberry.field`, fully custom | You own scoping and auth |
-| A resolver returning instances | Optimizer skipped; nested relations may be unscoped |
+| A resolver returning instances | Nested relations still scoped, one query each — see [`orm.optimize`](#ormoptimize) |
 
 The first four rows are all `orm.field`, which has four named forms that run at different times and take different arguments — see [The four kinds of field](#the-four-kinds-of-field).
 
-Type-level and field-level scopes compose in that order — `scope_rows` first, then `scope=` — and both run **before SQL executes**, while the prefetch is being built. `using=` is not a filter; it only adds eager-load paths.
+Type-level and field-level scopes compose in that order — `scope_rows` first, then `scope=` — and both run **before SQL executes**.
 
-Scoping hooks do **not** run when you build with `strawberry.Schema(query=Query)` instead of `orm.schema()` on Django or SQLAlchemy, or when a resolver returns materialized instances.
+Relation scoping does not depend on the optimizer. When the optimizer runs it applies the scope once while building the eager load; when it does not, the scope is applied again as each parent's relation is read. Either way the rows are scoped — the difference is how many queries it takes. The **root** field is the exception: `scope_rows` on a root query object is applied by the optimizer, so build with `orm.schema()`.
 
 ### Root custom query
 
@@ -359,6 +359,31 @@ class Query:
 
 Use `scope_rows` when the same rule applies everywhere the model loads, and a custom root resolver when the criteria belong to that one entry point. See [List Fields](#list-fields) for a comparison.
 
+### `orm.optimize`
+
+Sometimes a resolver cannot return a query object — it has just written the rows, or it returns them inside a wrapper. The rows are still scoped when their relations are read, but each relation costs a query per parent. `orm.optimize` puts them back on the eager-loaded path:
+
+```python
+# Django
+@strawberry.field
+def create_post(self, info: strawberry.Info, ...) -> PostType:
+    post = Post.objects.create(...)
+    return orm.optimize(post, info)
+```
+
+It takes a query object, a model instance, or a list, and returns anything else untouched — so it is safe to wrap a whole payload. Relations are loaded **onto the instances you pass in**, so values you have just set in memory are never overwritten by a re-read of the database. On an async backend the result is awaitable.
+
+When the rows sit below the field being resolved, point it at them with `at`. The optimizer reads the selection set from the current field, and for a payload the relations to load are named under `data`, not beside it:
+
+```python
+# Django
+@strawberry.field
+def recent_posts(self, info: strawberry.Info) -> Payload:
+    return Payload(data=orm.optimize(rows, info, at="data"), errors=None)
+```
+
+`at` also takes a sequence for a deeper path, and matches either `camelCase` or `snake_case`. Getting it wrong is not an error — nothing is eager-loaded and the rows come back as they would have anyway.
+
 ### `orm.schema()`
 
 Build schemas with `orm.schema()`. The optimizer is enabled by default: it executes query objects, eager-loads relations from the selection set, applies field hints, and honours `scope_rows`. On Django and SQLAlchemy nested scoping depends on it, so this is not an optional performance tweak.
@@ -466,17 +491,15 @@ posts: list[PostType] = orm.field.scoped(
 )
 ```
 
-For `{ users { name posts { title } } }` the order is always `PostType.scope_rows` then `UserType.posts.load`. With a plain annotation and no `scope=`, only the first line appears. Neither hook runs again as GraphQL reads each `user.posts`. The repo asserts this by patching `print` — see `tests/backends/*/test_query_scoping_hook_order.py`.
+For `{ users { name posts { title } } }` the order is always `PostType.scope_rows` then `UserType.posts.load`. With a plain annotation and no `scope=`, only the first line appears. When the relation was eager-loaded, the hooks run once for the whole batch; when it was not, they run again for each parent as the relation is read. Either way they run. The repo asserts this by patching `print` — see `tests/backends/*/test_query_scoping_hook_order.py`.
 
 **Fragments.** The optimizer walks inline fragments (`... on PostType`) and named fragment spreads, so relations inside them are prefetched normally.
 
-**Tortoise.** Annotation-only list relations also apply `_apply_nested_queryset` at resolve time when prefetch did not run. The `scope_rows` then `scope=` order is the same there.
-
 **Field permissions.** `orm.field.auto(permission_classes=[...])` — see [Declaring fields](#declaring-fields).
 
 </details>
 
-> If nested rows come back unscoped, check three things in order: that the schema was built with `orm.schema()`, that root resolvers return query objects, and that `scope_rows` exists on every exposed type. See [Security](#security).
+> If nested rows come back unscoped, check that `scope_rows` exists on every exposed type. Scoping does not depend on the optimizer: a resolver returning a list is slower than one returning a query object, but it is not less scoped. See [Security](#security).
 
 ---
 
@@ -1523,6 +1546,22 @@ Filters and ordering are applied *before* pagination, so the connection always s
 
 `orm.connection()` accepts the same keyword arguments as `relay.connection()` — `name`, `description`, `deprecation_reason`, `extensions`, and `max_results`.
 
+### Supplying the queryset yourself
+
+The decorator above is one way to give `orm.connection()` a resolver. You can also pass one by keyword, which is what you want when the connection is a field on a type you are assembling rather than a method you are writing:
+
+```python
+# Django
+def recent_users(info: strawberry.types.Info) -> Iterable[UserNode]:
+    return User.objects.order_by("-created_at")
+
+@strawberry.type
+class Query:
+    users = orm.connection(ORMListConnection[UserNode], resolver=recent_users)
+```
+
+Either way the library still builds everything around your rows: the generated `filter`, `order`, and `groupBy` arguments, the grouped connection type when the node declares a group-by, `totalCount`, and optimizer integration. Your resolver does not need to accept `filter` or `order` — they are applied to the query object you return. Arguments of your own are passed through and appear on the field.
+
 ### Node mutations
 
 `orm.mutations.create_node_input()` and `orm.mutations.update_node_input()` generate catch-all Relay Node mutation *inputs* with recursive nested refs; you supply the resolver. See [Node Mutation Inputs](#node-mutation-inputs) for full documentation.

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "strawberry-orm"
-version = "0.15.1"
+version = "0.16.0"
 description = "Unified, backend-agnostic ORM abstraction for Strawberry GraphQL"
 readme = "README.md"
 license = "MIT"

src/strawberry_orm/backends/django.py

@@ -242,33 +242,45 @@ def is_type_of(inner_cls: type, obj: object, info: Any) -> bool:
                         )
                     else:
 
-                        def _make_resolver(fname: str, return_ann: Any) -> Any:
-                            def resolver(self: Any) -> Any:
-                                return list(getattr(self, fname).all())
+                        def _make_resolver(
+                            fname: str, return_ann: Any, backend: Any
+                        ) -> Any:
+                            def resolver(self: Any, info: Any) -> Any:
+                                return list(
+                                    _scoped_related_queryset(backend, self, fname, info)
+                                )
 
                             resolver.__name__ = fname
-                            resolver.__annotations__ = {"return": return_ann}
-                            if self._django_async_safe:
+                            resolver.__annotations__ = {
+                                "info": strawberry.types.Info,
+                                "return": return_ann,
+                            }
+                            if backend._django_async_safe:
                                 resolver = async_safe_resolver(resolver)
                             return strawberry.field(resolver=resolver)
 
-                        setattr(cls, field_name, _make_resolver(field_name, ann))
+                        setattr(cls, field_name, _make_resolver(field_name, ann, self))
                 elif kind in ("fk", "one"):
 
-                    def _make_fk_resolver(fname: str, return_ann: Any) -> Any:
-                        def resolver(self: Any) -> Any:
-                            return getattr(self, fname)
+                    def _make_fk_resolver(
+                        fname: str, return_ann: Any, backend: Any
+                    ) -> Any:
+                        def resolver(self: Any, info: Any) -> Any:
+                            return _scoped_related_instance(backend, self, fname, info)
 
                         resolver.__name__ = fname
-                        resolver.__annotations__ = {"return": return_ann}
-                        if self._django_async_safe:
+                        resolver.__annotations__ = {
+                            "info": strawberry.types.Info,
+                            "return": return_ann,
+                        }
+                        if backend._django_async_safe:
                             resolver = async_safe_resolver(
                                 resolver,
                                 materialize=False,
                             )
                         return strawberry.field(resolver=resolver)
 
-                    setattr(cls, field_name, _make_fk_resolver(field_name, ann))
+                    setattr(cls, field_name, _make_fk_resolver(field_name, ann, self))
 
             annotations = getattr(cls, "__annotations__", {})
             self._check_lazy_relation_fields(cls, model, annotations)
@@ -571,6 +583,11 @@ def is_query_object(self, value: Any) -> bool:
 
         return isinstance(value, QuerySet)
 
+    def is_model_instance(self, value: Any) -> bool:
+        from django.db.models import Model
+
+        return isinstance(value, Model)
+
     def relation_names(self, model: type) -> set[str]:
         return {
             f.name
@@ -719,7 +736,14 @@ def _resolve_basic(root: Any) -> Any:
     def optimizer_extension(self, **kwargs: Any) -> type[SchemaExtension]:
         return OptimizerExtension.configure(backend=self, store=self._store)
 
-    def apply_optimizer_hints(self, store: Any, query: Any, info: Any) -> Any:
+    def _relation_lookups(
+        self, store: Any, model: type, info: Any
+    ) -> tuple[list[str], list[Any]]:
+        """Return the lookups the current selection needs for *model*.
+
+        Shared by query optimization and by loading relations onto instances
+        the caller already holds, so both apply the same row scoping.
+        """
         import re
 
         from strawberry_orm.optimizer.selections import (
@@ -728,17 +752,7 @@ def apply_optimizer_hints(self, store: Any, query: Any, info: Any) -> Any:
             iter_field_nodes,
         )
 
-        def optimize() -> Any:
-            try:
-                model = query.model
-            except AttributeError:
-                return query
-
-            optimized_query = query
-            get_qs = self._type_querysets.get(model)
-            if get_qs is not None:
-                optimized_query = get_qs(optimized_query, info)
-
+        def compute() -> tuple[list[str], list[Any]]:
             select_related: list[str] = []
             prefetch_related: list[Any] = []
             fragments = fragments_from_info(info)
@@ -924,6 +938,25 @@ def _walk_selections(
             for field_node in field_nodes_from_info(info):
                 _walk_selections(field_node.selection_set, model)
 
+            return select_related, prefetch_related
+
+        return compute()
+
+    def apply_optimizer_hints(self, store: Any, query: Any, info: Any) -> Any:
+        def optimize() -> Any:
+            try:
+                model = query.model
+            except AttributeError:
+                return query
+
+            optimized_query = query
+            get_qs = self._type_querysets.get(model)
+            if get_qs is not None:
+                optimized_query = get_qs(optimized_query, info)
+
+            select_related, prefetch_related = self._relation_lookups(
+                store, model, info
+            )
             if select_related:
                 optimized_query = optimized_query.select_related(*select_related)
             if prefetch_related:
@@ -933,12 +966,121 @@ def _walk_selections(
 
         return run_sync(optimize, thread_sensitive=True)
 
+    def load_relations(self, store: Any, instances: list[Any], info: Any) -> list[Any]:
+        """Eager-load the selected relations onto instances already in memory.
+
+        ``prefetch_related_objects`` fills the relation caches in place, so
+        scalar values the caller is holding - which may be fresher than the
+        database, straight out of a mutation - are never overwritten.
+        """
+
+        def load() -> list[Any]:
+            from django.db.models import prefetch_related_objects
+
+            by_model: dict[type, list[Any]] = {}
+            for instance in instances:
+                by_model.setdefault(type(instance), []).append(instance)
+
+            for model, rows in by_model.items():
+                select_related, prefetch_related = self._relation_lookups(
+                    store, model, info
+                )
+                lookups = _dedupe_lookups([*select_related, *prefetch_related])
+                if lookups:
+                    prefetch_related_objects(rows, *lookups)
+            return instances
+
+        return run_sync(load, thread_sensitive=True)
+
 
 # ---------------------------------------------------------------------------
 # Internal helpers
 # ---------------------------------------------------------------------------
 
 
+def _scoped_related_queryset(
+    backend: Any, instance: Any, field_name: str, info: Any
+) -> Any:
+    """Return the related rows for *field_name*, with row scoping applied.
+
+    The optimizer scopes relations when it builds the prefetch, but a resolver
+    that returns materialized rows never gives it a query to work on. Reading
+    the relation off the instance would then skip the related type's
+    ``scope_rows`` entirely and hand back rows the caller may not read, so the
+    scope is applied here instead.
+
+    A prefetched relation was already scoped on the way in; re-filtering it
+    would throw the cache away and issue the query this path exists to avoid.
+    """
+    from strawberry_orm.lazy_resolution import _django_relation_prefetched
+
+    manager = getattr(instance, field_name)
+    if _django_relation_prefetched(instance, field_name):
+        return manager.all()
+
+    restrict = backend.relation_scope(type(instance), field_name, info)
+    queryset = manager.all()
+    return queryset if restrict is None else restrict(queryset, info)
+
+
+def _scoped_related_instance(
+    backend: Any, instance: Any, field_name: str, info: Any
+) -> Any:
+    """Return the related row for a to-one *field_name*, with scoping applied.
+
+    A scoped-out row reads as absent, matching what the optimizer produces
+    when its scoped load finds nothing on the other end.
+    """
+    from strawberry_orm.lazy_resolution import _django_relation_prefetched
+
+    restrict = backend.relation_scope(type(instance), field_name, info)
+    if restrict is None:
+        return getattr(instance, field_name)
+
+    # With a scope in play a removed row has to read as absent. The scoped
+    # eager load leaves the cache empty, and Django's forward descriptor
+    # raises rather than returning None for a non-nullable column.
+    if _django_relation_prefetched(instance, field_name):
+        return getattr(instance, field_name, None)
+
+    related = getattr(instance, field_name, None)
+    if related is None:
+        return None
+    queryset = type(related)._default_manager.filter(pk=related.pk)
+    return restrict(queryset, info).first()
+
+
+def _dedupe_lookups(lookups: list[Any]) -> list[Any]:
+    """Collapse repeated lookups for one path, keeping the scoped one.
+
+    The same relation can be reached twice - two aliases of one field, or a
+    field both selected and named in ``using=`` - and Django rejects a repeated
+    path when either occurrence carries a queryset. Which duplicate survives
+    matters: only the ``Prefetch`` carries the row scope, so a bare path must
+    never be allowed to shadow it.
+    """
+    position: dict[str, int] = {}
+    unique: list[Any] = []
+    for lookup in lookups:
+        path = (
+            lookup if isinstance(lookup, str) else getattr(lookup, "prefetch_to", None)
+        )
+        if path is None:
+            unique.append(lookup)
+            continue
+        if path not in position:
+            position[path] = len(unique)
+            unique.append(lookup)
+            continue
+        kept = unique[position[path]]
+        if (
+            getattr(kept, "queryset", None) is None
+            and getattr(lookup, "queryset", None) is not None
+        ):
+            unique[position[path]] = lookup
+    return unique
+
+
 def _is_parent_predicate(
     child: Any, parent_pk: Any, base_alias: str | None = None
 ) -> bool:
@@ -990,7 +1132,15 @@ def _resolve(
         filter: Any = None,
         order: Any = None,
     ) -> list[Any]:
+        if filter is None and order is None:
+            return list(_scoped_related_queryset(backend, self, fname, info))
+
+        # Filtering or ordering re-runs the query, so a prefetched result
+        # cannot carry the scope through - it has to be reapplied here.
+        restrict = backend.relation_scope(type(self), fname, info)
         qs = getattr(self, fname).all()
+        if restrict is not None:
+            qs = restrict(qs, info)
         if filter is not None:
             qs = backend.apply_filters(qs, filter, rel_model, info=info)
         if order is not None:

src/strawberry_orm/backends/protocol.py

@@ -201,6 +201,16 @@ def is_query_object(self, value: Any) -> bool:
         """Return ``True`` if *value* is a query object the optimizer can handle."""
         ...
 
+    def is_model_instance(self, value: Any) -> bool:
+        """Return ``True`` if *value* is a persisted model instance."""
+        ...
+
+    def load_relations(
+        self, store: Any, instances: list[Any], info: Any
+    ) -> AwaitableOrValue[list[Any]]:
+        """Eager-load the selected relations onto *instances*, in place."""
+        ...
+
     def relation_names(self, model: type) -> set[str]:
         """Return the names of *model*'s relations, for hint validation."""
         ...