feat(secrets): scope- and type-aware ResolvedSecrets + name-keying

Resolved secrets are now keyed by name (the connector change). Make
SecretsAccessor.to_dict() return a ResolvedSecrets dict subclass with
for_scope / for_scope_of_type / of_type / secret_type / field_for, and make
the scalar @secret(type) resolver match on each secret's serialized "type"
field (column-name fallback for the old format). Update the secret fixtures
to select by type via of_type, and add a multi_secret_demo fixture that
resolves two same-type scoped secrets in one bind. Unit-tested.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: rustyconover

tests/test_resolved_secrets.py

@@ -0,0 +1,61 @@
+"""Unit tests for ResolvedSecrets type- and scope-aware selection."""
+
+from vgi.table_function import ResolvedSecrets
+
+
+def _secrets() -> ResolvedSecrets:
+    # Values are plain strings here; ResolvedSecrets also accepts pyarrow Scalars
+    # (it calls .as_py() when present).
+    return ResolvedSecrets(
+        {
+            "my_s3": {"type": "s3", "key_id": "AAA", "scope": "s3://bucket-a"},
+            "my_s3_b": {
+                "type": "s3",
+                "key_id": "BBB",
+                "scope": "s3://bucket-b\ns3://bucket-b2",
+            },
+            "my_gcs": {"type": "gcs", "key_id": "G"},
+        }
+    )
+
+
+def test_type_aware() -> None:
+    """Type-aware accessors find/identify secrets by type."""
+    s = _secrets()
+    assert s.secret_type("my_s3") == "s3"
+    assert s.secret_type("my_gcs") == "gcs"
+    assert len(s.of_type("s3")) == 2
+    assert len(s.of_type("gcs")) == 1
+    assert s.of_type("azure") == []
+
+
+def test_for_scope_of_type_per_bucket() -> None:
+    """Per-bucket scope selection picks the right s3 secret."""
+    s = _secrets()
+    assert s.for_scope_of_type("s3://bucket-a/x.dat", "s3")["key_id"] == "AAA"
+    assert s.for_scope_of_type("s3://bucket-b2/y.dat", "s3")["key_id"] == "BBB"
+    assert s.field_for("s3://bucket-a/x.dat", "key_id") == "AAA"
+
+
+def test_longest_prefix_and_fallback() -> None:
+    """Longest scope prefix wins; unscoped is the fallback."""
+    s = ResolvedSecrets(
+        {
+            "broad": {"type": "s3", "key_id": "broad", "scope": "s3://bucket"},
+            "narrow": {"type": "s3", "key_id": "narrow", "scope": "s3://bucket/data"},
+        }
+    )
+    assert s.for_scope("s3://bucket/data/x.dat")["key_id"] == "narrow"
+    assert s.for_scope("s3://bucket/other/x.dat")["key_id"] == "broad"
+
+    unscoped = ResolvedSecrets({"only": {"type": "s3", "key_id": "only"}})
+    assert unscoped.for_scope("s3://any/x")["key_id"] == "only"
+
+    assert s.for_scope("s3://nope/x") is None
+
+
+def test_dict_access_still_works() -> None:
+    """ResolvedSecrets keeps plain dict access."""
+    s = _secrets()
+    assert s["my_s3"]["key_id"] == "AAA"
+    assert s.get("missing") is None

vgi/_test_fixtures/table/__init__.py

@@ -119,6 +119,7 @@
     TenThousandFunction,
 )
 from vgi._test_fixtures.table.settings import (
+    MultiSecretDemoFunction,
     ScopedSecretDemoFunction,
     SecretDemoFunction,
     SettingsAwareFunction,
@@ -212,6 +213,7 @@
     "RffStructScanFunction",
     "RowIdSequenceFunction",
     "SampleEchoFunction",
+    "MultiSecretDemoFunction",
     "ScopedSecretDemoFunction",
     "SecretDemoFunction",
     "SequenceFunction",

vgi/_test_fixtures/table/settings.py

@@ -321,7 +321,7 @@ def on_bind(
     @classmethod
     def initial_state(cls, params: ProcessParams[None]) -> SecretDemoState:
         """Build initial state from secret key-value pairs."""
-        secret = params.secrets.get("vgi_example", {})
+        secret = next(iter(params.secrets.of_type("vgi_example")), {})
         keys = list(secret.keys())
         values = [str(v.as_py()) for v in secret.values()]
         types = [str(v.type) for v in secret.values()]
@@ -406,12 +406,52 @@ def on_bind(
     @classmethod
     def initial_state(cls, params: ProcessParams[ScopedSecretDemoArgs]) -> ScopedSecretDemoState:
         """Build state from resolved secrets."""
-        secret = params.secrets.get("vgi_example", {})
+        secret = next(iter(params.secrets.of_type("vgi_example")), {})
         return ScopedSecretDemoState(
             found=bool(secret),
             secret_keys=",".join(secret.keys()) if secret else "",
         )
 
+
+@dataclass(kw_only=True)
+class MultiSecretDemoState(ArrowSerializableDataclass):
+    """State for MultiSecretDemoFunction."""
+
+    api_key: str = ""
+
+
+@init_single_worker
+class MultiSecretDemoFunction(TableFunctionGenerator[ScopedSecretDemoArgs, MultiSecretDemoState]):
+    """Resolve TWO same-type scoped secrets in one bind, then select per path.
+
+    Requests the ``vgi_example`` secret for both ``s3://bucket-a/`` and
+    ``s3://bucket-b/`` scopes in a single bind. Because resolved secrets are keyed
+    by name, both survive; ``for_scope_of_type`` then picks the one whose scope
+    matches the ``path`` argument and returns its ``api_key``.
+    """
+
+    class Meta:
+        """Metadata for MultiSecretDemoFunction."""
+
+        name = "multi_secret_demo"
+        description = "Demo: two same-type scoped secrets resolved in one bind"
+
+    @classmethod
+    def on_bind(cls, params: BindParams[ScopedSecretDemoArgs]) -> BindResponse:
+        """Request the secret for two distinct scopes of the same type."""
+        params.secrets.get("vgi_example", scope="s3://bucket-a/")
+        params.secrets.get("vgi_example", scope="s3://bucket-b/")
+        return BindResponse(output_schema=schema({"api_key": pa.string()}))
+
+    @classmethod
+    def initial_state(cls, params: ProcessParams[ScopedSecretDemoArgs]) -> MultiSecretDemoState:
+        """Select the resolved secret matching the path and return its api_key."""
+        secret = params.secrets.for_scope_of_type(params.args.path, "vgi_example") or {}
+        api_key = secret.get("api_key")
+        if api_key is not None and hasattr(api_key, "as_py"):
+            api_key = api_key.as_py()
+        return MultiSecretDemoState(api_key="" if api_key is None else str(api_key))
+
     @classmethod
     def process(
         cls,

vgi/_test_fixtures/worker.py

@@ -171,6 +171,7 @@
     RffStructScanFunction,
     RowIdSequenceFunction,
     SampleEchoFunction,
+    MultiSecretDemoFunction,
     ScopedSecretDemoFunction,
     SecretDemoFunction,
     SequenceFunction,
@@ -412,6 +413,7 @@ def _build_enum_stats() -> dict[str, ColumnStatisticsInput]:
                 SampleEchoFunction,
                 RowIdSequenceFunction,
                 SecretDemoFunction,
+                MultiSecretDemoFunction,
                 ScopedSecretDemoFunction,
                 ExpressionFilterTestFunction,
                 SequenceFunction,

vgi/scalar_function.py

@@ -1067,12 +1067,30 @@ def _extract_compute_kwargs(
                 col_idx = settings_schema.get_field_index(setting_key)
                 kwargs[name] = bind_call.settings.column(col_idx)[0] if col_idx >= 0 else None
 
-        # Secret params: extract dict[str, pa.Scalar] from secrets RecordBatch
+        # Secret params: extract dict[str, pa.Scalar] from secrets RecordBatch.
+        # Secrets are keyed by secret NAME (not type), so match the requested
+        # secret_type on each resolved secret's serialized "type" field. The
+        # first secret of that type wins (use the scope-aware accessors on a
+        # table function's params.secrets to disambiguate several of one type).
         if bind_call.secrets is not None and cls._secret_params:
-            secrets_schema = bind_call.secrets.schema
+            by_type: dict[str, dict[str, Any]] = {}
+            by_col: dict[str, dict[str, Any]] = {}
+            names = bind_call.secrets.schema.names
+            for i in range(bind_call.secrets.num_columns):
+                scalar = bind_call.secrets.column(i)[0]
+                if not scalar.is_valid:
+                    continue
+                fields = _struct_scalar_to_dict(scalar)
+                t = fields.get("type")
+                t = t.as_py() if t is not None and hasattr(t, "as_py") else t
+                if t is not None and str(t) not in by_type:
+                    by_type[str(t)] = fields
+                by_col.setdefault(names[i], fields)
             for name, secret in cls._secret_params.items():
-                col_idx = secrets_schema.get_field_index(secret.secret_type)
-                kwargs[name] = _struct_scalar_to_dict(bind_call.secrets.column(col_idx)[0]) if col_idx >= 0 else None
+                chosen = by_type.get(secret.secret_type)
+                if chosen is None:
+                    chosen = by_col.get(secret.secret_type)
+                kwargs[name] = chosen
 
         # OutputLength param: pass the batch row count
         if cls._output_length_param is not None:

vgi/table_function.py

@@ -240,24 +240,28 @@ def pending_lookups(self) -> list[SecretLookupEntry]:
         """Return the list of pending secret lookups."""
         return list(self._pending_lookups)
 
-    def to_dict(self) -> dict[str, dict[str, pa.Scalar[Any]]]:
-        """Return all resolved secrets as a flat dict keyed by secret_type.
+    def to_dict(self) -> "ResolvedSecrets":
+        """Return all resolved secrets keyed by secret name.
 
-        Combines unscoped entries (column name = secret_type) with scoped
-        entries (``secret_N`` columns, keyed by ``secret_type`` from Arrow
-        field metadata).  Null/unresolved entries are omitted.
+        Resolved secrets are keyed by their unique DuckDB secret name, so several
+        secrets of the same type (e.g. one per S3 bucket) coexist. Each carries a
+        ``type`` field (the DuckDB secret type) and a ``scope`` field
+        (newline-joined scope prefixes). Scoped ``secret_N`` columns (keyed by
+        ``secret_type`` from Arrow field metadata) are merged in. Null/unresolved
+        entries are omitted.
 
         Returns:
-            Mapping of secret_type to its resolved key/scalar dict.
+            A :class:`ResolvedSecrets` (a dict keyed by secret name) with
+            type- and scope-aware selection helpers.
 
         """
         result = dict(self._unscoped)
         for meta, secret_dict in self._scoped:
             if secret_dict is not None:
-                key = meta.get("secret_type", "")
+                key = meta.get("secret_name") or meta.get("secret_type", "")
                 if key:
                     result[key] = secret_dict
-        return result
+        return ResolvedSecrets(result)
 
     def _find_scoped(
         self,
@@ -287,6 +291,74 @@ def _find_scoped(
         return None
 
 
+def _secret_scalar_str(v: Any) -> str:
+    """Render a resolved-secret field (a pyarrow Scalar or plain value) to str."""
+    if v is None:
+        return ""
+    py = v.as_py() if hasattr(v, "as_py") else v
+    return "" if py is None else str(py)
+
+
+class ResolvedSecrets(dict):
+    """Resolved secrets keyed by secret name, with type- and scope-aware lookup.
+
+    A plain ``dict`` (so ``secrets[name]`` and ``secrets.get(name)`` still work)
+    plus selectors that read each secret's connector-serialized ``type`` and
+    ``scope`` fields. Mirrors ``vgi::Secrets`` in the Rust SDK.
+    """
+
+    def secret_type(self, name: str) -> str | None:
+        """The DuckDB secret type of the named secret (its ``type`` field)."""
+        fields = self.get(name)
+        if not fields or "type" not in fields:
+            return None
+        return _secret_scalar_str(fields["type"])
+
+    def of_type(self, secret_type: str) -> list[dict[str, Any]]:
+        """Every resolved secret whose ``type`` field matches ``secret_type``."""
+        return [
+            f for f in self.values() if _secret_scalar_str(f.get("type")) == secret_type
+        ]
+
+    def for_scope(self, path: str) -> dict[str, Any] | None:
+        """The secret whose ``scope`` is the longest prefix of ``path``.
+
+        The connector serializes each secret's scope as a newline-joined list of
+        prefixes; a secret with no (or empty) scope matches as a last-resort
+        fallback. Returns ``None`` only when there are no candidate secrets.
+        """
+        return self._select_for_scope(path, None)
+
+    def for_scope_of_type(self, path: str, secret_type: str) -> dict[str, Any] | None:
+        """Like :meth:`for_scope` but only over secrets of ``secret_type``."""
+        return self._select_for_scope(path, secret_type)
+
+    def field_for(self, path: str, field: str) -> Any | None:
+        """A field of the best scope-matching secret for ``path``."""
+        fields = self.for_scope(path)
+        return None if fields is None else fields.get(field)
+
+    def _select_for_scope(
+        self, path: str, secret_type: str | None
+    ) -> dict[str, Any] | None:
+        best: dict[str, Any] | None = None
+        best_len = -1
+        fallback: dict[str, Any] | None = None
+        for fields in self.values():
+            if secret_type is not None and _secret_scalar_str(fields.get("type")) != secret_type:
+                continue
+            scope = _secret_scalar_str(fields.get("scope"))
+            if not scope:
+                if fallback is None:
+                    fallback = fields
+                continue
+            for prefix in scope.split("\n"):
+                if prefix and path.startswith(prefix) and len(prefix) > best_len:
+                    best_len = len(prefix)
+                    best = fields
+        return best if best is not None else fallback
+
+
 def project_schema(projection_ids: list[int] | None, schema: pa.Schema) -> pa.Schema:
     """Create the projected schema if projection_ids are supplied.