Tech-debt batch 25: INSERT…SELECT into tenant tables; Inertia once() custom keys (#693 #694)

CHANGELOG.md

@@ -33,6 +33,10 @@ All notable changes to this project are documented in this file. The format is b
   into the volume before replacing old containers (#689).
 - Load-test cookie seeding resolves the server's stored session-signing key
   through `merge_host_settings()`, preserving explicit environment overrides (#697).
+- DB: stamp tenant destinations in INSERT ... SELECT without appending VALUES;
+  validate explicit tenant literals and retain strict isolation and source scoping (#693).
+- Inertia: honor custom once-prop keys in client exclusions, skipping loaders
+  and values on full and partial requests (#694).
 
 ## [0.0.41] — 2026-10-25
 

docs/database/mixins.md

@@ -93,9 +93,20 @@ When `SM_MULTI_TENANT=true` and a request has an active tenant, the `do_orm_exec
 
 INSERTs populate `tenant_id` from the `current_tenant_id` contextvar. Creating a row for a different tenant than the active one — or changing `tenant_id` on an existing row — raises `TenantIsolationError`.
 
+`insert(Model).from_select(...)` and Core table targets follow the same rules:
+when the target column list omits `tenant_id`, the guard adds it and a constant
+tenant projection without changing the source's tenant/soft-delete filters.
+An explicit `tenant_id` projection must be a bound literal matching the active
+tenant (including execution-time parameter overrides). Other SQL expressions
+cannot be validated and raise `TenantIsolationError`; omit the tenant column
+for automatic stamping, or use an explicit `all_tenants` bypass for deliberate
+cross-tenant copies. Strict mode requires a tenant even if the source reads no
+tenant tables and explicitly supplies `tenant_id` (#693).
+
 ### Bypass
 
-A row inserted outside any tenant context (no active `current_tenant_id`) is stamped `None` and, because the column is non-nullable, fails at the DB rather than silently leaking. There is no per-statement `skip_tenant_filter` option; cross-tenant work uses `all_tenants()` (see [multi-tenancy](../framework/multi-tenancy.md)). `no_tenant()` is the opposite: it forces the unscoped state, where strict mode fails closed.
+Without an active tenant, strict mode raises `TenantIsolationError`; non-strict
+inserts use the install's fallback tenant (`default_tenant` or `"default"`). There is no per-statement `skip_tenant_filter` option; cross-tenant work uses `all_tenants()` (see [multi-tenancy](../framework/multi-tenancy.md)). `no_tenant()` is the opposite: it forces the unscoped state, where strict mode fails closed.
 
 ## `VersionedMixin`
 

docs/module-authoring.md

@@ -741,6 +741,11 @@ await inertia.render(
 )
 ```
 
+`once(loader, key="feature-flags")` announces its cache entry under the custom
+key, defaulting to the prop name when no key is given. Clients send that key in
+`X-Inertia-Except-Once-Props` to skip both the loader and the returned prop on
+full and partial requests (#694).
+
 `errors` is always present in props (it is the protocol's always-prop) and is
 filled from validation failures flashed by the previous request. A page `url`
 is always root-relative, and a prop that cannot be serialised raises

framework/db/simple_module_db/insert_guard.py

@@ -10,7 +10,9 @@
 
 from typing import Any
 
+from sqlalchemy import literal, select
 from sqlalchemy.orm import ORMExecuteState
+from sqlalchemy.sql.elements import BindParameter, Label
 
 from simple_module_db import model_registry as registry
 from simple_module_db.tenancy import (
@@ -77,6 +79,8 @@ def guard_insert(
     bound = current_tenant_id.get()
     tenant_id = None if bypass else bound
     stamp = bound or fallback
+    if getattr(execute_state.statement, "select", None) is not None:
+        return _guard_from_select(execute_state, tenant_id, stamp, mappers[0], strict)
     if getattr(execute_state.statement, "_multi_values", None):
         return _guard_multi_values(execute_state, tenant_id, stamp, mappers[0], strict)
     for value in assigned_tenant_ids(execute_state):
@@ -100,6 +104,46 @@ def guard_insert(
     return None
 
 
+def _guard_from_select(
+    execute_state: ORMExecuteState, tenant_id: str | None, stamp: str, entity: str, strict: bool
+) -> None:
+    stmt = execute_state.statement
+    if tenant_id is None and strict:
+        raise missing_tenant_error(entity, "INSERT")
+    names = list(stmt._select_names)
+    if TENANT_COLUMN in names:
+        if tenant_id is None:
+            return
+        # SQL expressions cannot be validated as Python values; require a bound
+        # literal, or an explicit all_tenants bypass for cross-tenant copies.
+        params = _param_rows(execute_state)
+        bind_names = stmt.compile().bind_names if params else {}
+        sources = [stmt.select]
+        while sources:
+            source = sources.pop()
+            if hasattr(source, "selects"):
+                sources.extend(source.selects)
+                continue
+            value = list(source.selected_columns)[names.index(TENANT_COLUMN)]
+            while isinstance(value, Label):
+                value = value.element
+            if not isinstance(value, BindParameter) or value.callable is not None:
+                raise TenantIsolationError("INSERT ... SELECT tenant_id must be a bound literal")
+            for row in params or [{}]:
+                assigned = row.get(
+                    value.key, row.get(bind_names.get(value, value.key), value.value)
+                )
+                if assigned != tenant_id:
+                    raise _foreign_insert(assigned, tenant_id)
+        return
+    source = stmt.select.subquery()
+    execute_state.statement = stmt.from_select(
+        [*names, TENANT_COLUMN],
+        select(*source.c, literal(stamp)),
+        include_defaults=stmt.include_insert_from_select_defaults,
+    )
+
+
 def _guard_multi_values(
     execute_state: ORMExecuteState, tenant_id: str | None, stamp: str, entity: str, strict: bool
 ) -> Any:

framework/db/tests/test_db_insert_select_tenant.py

@@ -0,0 +1,129 @@
+"""Tenant destinations for INSERT ... SELECT (#693)."""
+
+import pytest
+from _core_scope_models import Tag, _Base, tag
+from simple_module_db.listeners import attach_session_listeners
+from simple_module_db.query_filter import EngineTenancy, bind_engine_policy
+from simple_module_db.tenancy import (
+    TenantIsolationError,
+    current_tenant_id,
+    no_tenant,
+    tenant_context,
+)
+from sqlalchemy import bindparam, create_engine, insert, literal, select, union_all
+from sqlalchemy.orm import Session, configure_mappers
+
+
+@pytest.fixture
+def seeded():
+    configure_mappers()
+    engine = create_engine("sqlite://")
+    _Base.metadata.create_all(engine)
+    with engine.begin() as conn:
+        conn.execute(
+            insert(tag),
+            [
+                {"id": 1, "name": "used", "tenant_id": "A"},
+                {"id": 2, "name": "unused", "tenant_id": "A"},
+                {"id": 3, "name": "b-tag", "tenant_id": "B"},
+            ],
+        )
+
+    class TenantSession(Session):
+        pass
+
+    attach_session_listeners(TenantSession)
+    bind_engine_policy(engine, EngineTenancy(tenant_strict=False))
+    try:
+        with TenantSession(engine) as session, tenant_context("A"):
+            yield session
+    finally:
+        engine.dispose()
+
+
+def _rows(session, stmt, **options):
+    return sorted(session.execute(stmt.execution_options(**options)).all(), key=repr)
+
+
+@pytest.mark.parametrize("compound", [False, True])
+@pytest.mark.parametrize("target", [tag, Tag])
+def test_insert_select_stamps_tenant_and_scopes_source(seeded, compound, target):
+    source = select((tag.c.id + 100).label("id"), literal("copy").label("name"))
+    if compound:
+        source = union_all(source.where(tag.c.id == 1), source.where(tag.c.id == 2))
+    seeded.execute(insert(target).from_select(["id", "name"], source))
+    assert _rows(
+        seeded, select(tag.c.id, tag.c.tenant_id).where(tag.c.id >= 100), all_tenants=True
+    ) == [(101, "A"), (102, "A")]
+
+
+def test_insert_select_accepts_matching_explicit_tenant(seeded):
+    source = select(literal(101), literal("copy"), literal("A"))
+    seeded.execute(insert(tag).from_select(["id", "name", "tenant_id"], source))
+    assert _rows(seeded, select(tag.c.id).where(tag.c.id == 101)) == [(101,)]
+
+
+def test_insert_select_refuses_foreign_explicit_tenant(seeded):
+    source = select(literal(101), literal("copy"), literal("B"))
+    with pytest.raises(TenantIsolationError):
+        seeded.execute(insert(tag).from_select(["id", "name", "tenant_id"], source))
+
+
+@pytest.mark.parametrize("explicit", [False, True])
+def test_strict_insert_select_requires_tenant_even_for_plain_source(seeded, explicit):
+    bind_engine_policy(seeded.bind, EngineTenancy(tenant_strict=True))
+    current_tenant_id.set(None)
+    columns, values = ["id", "name"], [literal(101), literal("copy")]
+    if explicit:
+        columns.append("tenant_id")
+        values.append(literal("A"))
+    with pytest.raises(TenantIsolationError):
+        seeded.execute(insert(tag).from_select(columns, select(*values)))
+
+
+def test_unbound_insert_select_uses_fallback_tenant(seeded):
+    with no_tenant():
+        seeded.execute(
+            insert(tag).from_select(["id", "name"], select(literal(101), literal("copy")))
+        )
+    assert _rows(seeded, select(tag.c.tenant_id).where(tag.c.id == 101), all_tenants=True) == [
+        ("default",)
+    ]
+
+
+@pytest.mark.parametrize("strict", [False, True])
+def test_insert_select_bypass_preserves_explicit_tenant(seeded, strict):
+    bind_engine_policy(seeded.bind, EngineTenancy(tenant_strict=strict))
+    stmt = insert(tag).from_select(
+        ["id", "name", "tenant_id"], select(literal(101), literal("copy"), literal("B"))
+    )
+    seeded.execute(stmt.execution_options(all_tenants=True))
+    assert _rows(seeded, select(tag.c.tenant_id).where(tag.c.id == 101), all_tenants=True) == [
+        ("B",)
+    ]
+
+
+def test_insert_select_refuses_unverifiable_tenant_expression(seeded):
+    source = select(tag.c.id + 100, tag.c.name, tag.c.tenant_id)
+    with pytest.raises(TenantIsolationError):
+        seeded.execute(insert(tag).from_select(["id", "name", "tenant_id"], source))
+
+
+def test_insert_select_checks_every_compound_branch(seeded):
+    source = union_all(
+        select(literal(101), literal("copy"), literal("A")),
+        select(literal(102), literal("copy"), literal("B")),
+    )
+    with pytest.raises(TenantIsolationError):
+        seeded.execute(insert(tag).from_select(["id", "name", "tenant_id"], source))
+
+
+@pytest.mark.parametrize("anonymous", [False, True])
+def test_insert_select_checks_execution_parameter_override(seeded, anonymous):
+    tenant = literal("A") if anonymous else bindparam("tenant", "A")
+    source = select(literal(101), literal("copy"), tenant)
+    parameter = "param_3" if anonymous else "tenant"
+    with pytest.raises(TenantIsolationError):
+        seeded.execute(
+            insert(tag).from_select(["id", "name", "tenant_id"], source), {parameter: "B"}
+        )

framework/inertia/simple_module_inertia/resolve.py

@@ -41,7 +41,7 @@ async def _deep_resolve(value: Any) -> Any:
 
 def _wants(name: str, prop: Prop, req: InertiaRequest, partial: bool) -> bool:
     """Should this prop's value be resolved for this response?"""
-    if prop.is_once and name in req.except_once_props:
+    if prop.is_once and (prop.once_key or name) in req.except_once_props:
         return False
     if prop.always:
         return True

framework/inertia/tests/test_inertia_once_custom_key.py

@@ -0,0 +1,35 @@
+"""Once exclusions use the announced cache key (#694)."""
+
+import pytest
+from simple_module_inertia.props import once
+from simple_module_inertia.request import InertiaRequest
+from simple_module_inertia.resolve import resolve_props
+
+
+@pytest.mark.parametrize("partial", [False, True])
+async def test_custom_once_key_exclusion_skips_loader_and_value(partial):
+    def fail_if_called():
+        pytest.fail("Excluded once loader must not run")
+
+    headers = {"X-Inertia": "true", "X-Inertia-Except-Once-Props": "feature-flags"}
+    if partial:
+        headers.update({"X-Inertia-Partial-Component": "Page", "X-Inertia-Partial-Data": "flags"})
+    out = await resolve_props(
+        {"flags": once(fail_if_called, key="feature-flags")},
+        {},
+        "Page",
+        InertiaRequest.from_headers(headers, method="GET"),
+    )
+    assert out.props == {}
+    assert out.once == {}
+
+
+async def test_custom_once_key_does_not_exclude_by_prop_name():
+    out = await resolve_props(
+        {"flags": once(lambda: ["enabled"], key="feature-flags")},
+        {},
+        "Page",
+        InertiaRequest.from_headers({"X-Inertia-Except-Once-Props": "flags"}, method="GET"),
+    )
+    assert out.props == {"flags": ["enabled"]}
+    assert out.once == {"feature-flags": {"prop": "flags", "expiresAt": None}}