BigQuery user-scoped OAuth creds

docs/configuration/datasources.md

@@ -160,12 +160,28 @@ Statement-level timeout is enforced via
 | `username` | string | No | Database username |
 | `password` | string | No | Database password |
 | `connection_string` | string | No | Full connection string (alternative to individual fields) |
-| `credentials_json` | string | No | Credentials JSON (used for BigQuery service accounts) |
+| `credentials_json` | string | No | BigQuery service-account key JSON |
+| `oauth_credentials_json` | string | No | BigQuery OAuth authorized-user grant JSON |
 | `schema_name` | string | No | Default schema name |
 
 !!! note
     Both `username` and `user` field names are accepted. The `user` alias is automatically mapped to `username` for compatibility with common database tooling conventions.
 
+### BigQuery credentials
+
+Three ways to authenticate, in the order SLayer prefers them:
+
+1. **`oauth_credentials_json`** — an OAuth *authorized user* grant, the shape `google.oauth2.credentials.Credentials.from_authorized_user_info` accepts (`token`, `refresh_token`, `client_id`, `client_secret`, `token_uri`). Queries run as that end user, against that user's own BigQuery permissions. An OAuth grant carries no project of its own, so one must come from the connection string — `bigquery://<project>/<dataset>` — or from a `quota_project_id` in the grant, which SLayer falls back to when the connection string omits it.
+2. **`credentials_json`** — a service-account key file's contents. One shared identity for every caller.
+3. **Neither** — Application Default Credentials (`GOOGLE_APPLICATION_CREDENTIALS`, or the attached compute identity).
+
+Setting both (1) and (2) is an error rather than a silent precedence win: they name different identities, and guessing is how a per-user query quietly runs as the shared service account.
+
+An authorized-user grant will **not** work in `credentials_json` — `sqlalchemy-bigquery` routes that field to `service_account.Credentials.from_service_account_info`, which only understands key files. SLayer detects the mix-up and says so.
+
+!!! note "Engine caching and credentials"
+    Cached engines are keyed by credentials as well as by URL, so two identities never share a connection pool. The cache is bounded (default 64 engines, `SLAYER_MAX_CACHED_ENGINES`) and evicts least-recently-used, which matters once per-user credentials make its size track *users* rather than datasources. For OAuth grants the key uses the durable part of the grant, so a refreshed access token reuses its engine instead of leaking a new one. An engine whose credentials get rejected — a revoked grant, a rotated key — is dropped and rebuilt on the next call.
+
 ## Ingesting at Startup
 
 To run idempotent auto-ingestion across every configured datasource each time `slayer serve` or `slayer mcp` boots, pass `--ingest-on-startup` (or set `SLAYER_INGEST_ON_STARTUP=1`). See [Ingesting at Startup](../concepts/ingestion.md#ingesting-at-startup) for the full contract.

slayer/core/models.py

@@ -792,6 +792,12 @@ class DatasourceConfig(BaseModel):
     # When unset, BigQuery falls back to Application Default Credentials
     # (``GOOGLE_APPLICATION_CREDENTIALS`` env var or attached compute identity).
     credentials_json: str | None = Field(default=None, repr=False)
+    # BigQuery-specific. A Google OAuth authorized-user grant as JSON, in the
+    # shape ``Credentials.from_authorized_user_info`` consumes — the per-end-user
+    # auth path, so queries run with that user's permissions. Mutually exclusive
+    # with ``credentials_json``, which cannot carry a grant: the driver feeds it
+    # to ``from_service_account_info``.
+    oauth_credentials_json: str | None = Field(default=None, repr=False)
 
     @model_validator(mode="before")
     @classmethod

slayer/engine/cache.py

@@ -27,6 +27,8 @@
 from sqlglot.optimizer.normalize_identifiers import normalize_identifiers
 from sqlglot.optimizer.scope import Scope, traverse_scope
 
+from slayer.sql.engine_factory import EngineCacheKey
+
 # Marker alias prefix for refresh-key scan projections. Also lets tests
 # distinguish a refresh-key scan query from a data query.
 _RK_ALIAS_PREFIX = "slayer_rk_"
@@ -123,7 +125,7 @@ class _CacheEntry(BaseModel):
     sql: str
     ds_fingerprint: str
     dialect: str
-    ds_key: tuple[str, str]
+    ds_key: EngineCacheKey
     resolved_data_source: str | None = None
     original_input: Any = None
     variables: dict[str, Any] | None = None

slayer/engine/query_engine.py

@@ -73,7 +73,8 @@
 from slayer.sql.client import SlayerSQLClient
 from slayer.sql.dialects import SqlDialect, dialect_for_ds_type, get_dialect
 from slayer.sql import engine_factory
-from slayer.sql.engine_factory import _runtime_fingerprint
+from slayer.sql.engine_factory import EngineCacheKey
+from slayer.sql.engine_factory import _cache_key as _engine_cache_key
 from slayer.sql.generator import SQLGenerator
 from slayer.sql.reserved_keywords import SLAYER_RESERVED_KEYWORDS
 from slayer.sql.session_policy import ScopedTable, apply_session_policy
@@ -82,15 +83,17 @@
 logger = logging.getLogger(__name__)
 
 
-def _sql_client_cache_key(datasource: DatasourceConfig) -> tuple[str, str]:
+def _sql_client_cache_key(datasource: DatasourceConfig) -> EngineCacheKey:
     """Cache key for ``SlayerQueryEngine._sql_clients``.
 
-    Mirrors ``engine_factory``'s cache key so two datasources differing
-    in (e.g.) Snowflake ``warehouse`` get distinct ``SlayerSQLClient``
-    instances (and therefore distinct factory-cached engines with the
-    correct per-connection ``USE`` listener).
+    Delegates to ``engine_factory``'s key builder rather than re-deriving it:
+    each client memoizes the engine it got from the factory, so if the two keys
+    disagreed a caller could be handed a client whose engine was built for
+    different credentials. Sharing one implementation makes that impossible.
     """
-    return (datasource.get_connection_string(), _runtime_fingerprint(datasource))
+    return _engine_cache_key(
+        datasource=datasource, connection_string=datasource.get_connection_string(),
+    )
 
 
 class _ResolvedItem(BaseModel):
@@ -495,7 +498,7 @@ class _Prepared(BaseModel):
     sql: str
     attributes: "ResponseAttributes"
     expected_columns: list[str]
-    ds_key: tuple[str, str]
+    ds_key: EngineCacheKey
     ds_fingerprint: str
 
 
@@ -549,7 +552,7 @@ def __init__(
         # ``engine_factory``'s cache so Snowflake datasources sharing a
         # connection_name but differing in warehouse/role/database/schema
         # get distinct clients (DEV-1551).
-        self._sql_clients: dict[tuple[str, str], SlayerSQLClient] = {}
+        self._sql_clients: dict[EngineCacheKey, SlayerSQLClient] = {}
         # DEV-1578: immutable, engine-global forced-filter policy. When set,
         # every generated SQL is rewritten to scope each physical table to the
         # configured tenant before execution / dry-run / explain / profiling.
@@ -563,7 +566,7 @@ def __init__(
         # datasource, for the correlated-subquery join-rule gate. ``None`` (or
         # a missing entry) fails closed. Populated by
         # ``_preflight_clickhouse_correlated`` before the policy rewrite.
-        self._ch_version_cache: dict[tuple[str, str], tuple[int, int] | None] = {}
+        self._ch_version_cache: dict[EngineCacheKey, tuple[int, int] | None] = {}
         # DEV-1587: per-engine, in-memory, opt-in query result cache. The
         # cache is local to this engine instance so two engines with
         # different RLS / connection settings keep separate caches.
@@ -1357,7 +1360,7 @@ async def _execute_pipeline(
         return await self._run_and_build(prepared=prepared, client=client)
 
     def _get_client(
-        self, datasource: DatasourceConfig, ds_key: tuple[str, str]
+        self, datasource: DatasourceConfig, ds_key: EngineCacheKey
     ) -> SlayerSQLClient:
         """Reuse (or open) the SQL client + connection pool for a datasource.
 
@@ -1604,21 +1607,21 @@ async def _run_and_cleanup() -> RefreshResult:
 
     async def _refresh_scan_all(
         self, snapshot: "dict[str, _CacheEntry]", result: RefreshResult,
-    ) -> "tuple[dict[tuple[tuple[str, str], str, str], Any], set]":
+    ) -> "tuple[dict[tuple[EngineCacheKey, str, str], Any], set]":
         """Collate applicable ``(ds_key, table)`` scan targets across all
         entries and run ONE batched scan per pair. Scan failures are recorded
         as continue-on-error ``RefreshError(phase="refresh_key_scan")`` and the
         table is marked failed so its dependent entries are left unchanged."""
-        targets: dict[tuple[tuple[str, str], str], set] = {}
-        dialects: dict[tuple[str, str], str] = {}
-        clients: dict[tuple[str, str], SlayerSQLClient] = {}
+        targets: dict[tuple[EngineCacheKey, str], set] = {}
+        dialects: dict[EngineCacheKey, str] = {}
+        clients: dict[EngineCacheKey, SlayerSQLClient] = {}
         for entry in snapshot.values():
             ds_key = tuple(entry.ds_key)
             dialects[ds_key] = entry.dialect
             for table, expr in entry.applicable:
                 targets.setdefault((ds_key, table), set()).add(expr)
 
-        fresh: dict[tuple[tuple[str, str], str, str], Any] = {}
+        fresh: dict[tuple[EngineCacheKey, str, str], Any] = {}
         failed: set = set()
         for (ds_key, table), exprs in targets.items():
             expr_list = sorted(exprs)
@@ -1639,9 +1642,9 @@ async def _refresh_scan_all(
 
     async def _client_for_refresh(
         self,
-        ds_key: tuple[str, str],
+        ds_key: EngineCacheKey,
         snapshot: "dict[str, _CacheEntry]",
-        clients: dict[tuple[str, str], SlayerSQLClient],
+        clients: dict[EngineCacheKey, SlayerSQLClient],
     ) -> SlayerSQLClient:
         """Get the cached client for ``ds_key``, reopening it from an entry's
         recorded resolved datasource name if it isn't cached."""
@@ -1671,7 +1674,7 @@ async def _refresh_one(
         *,
         key: str,
         entry: _CacheEntry,
-        fresh_values: "dict[tuple[tuple[str, str], str, str], Any]",
+        fresh_values: "dict[tuple[EngineCacheKey, str, str], Any]",
         failed_tables: set,
         result: RefreshResult,
     ) -> None:

slayer/engine/schema_drift.py

@@ -49,6 +49,7 @@
     _sa_type_to_data_type,
 )
 from slayer.sql.client import SlayerSQLClient
+from slayer.sql.engine_factory import EngineCacheKey
 
 logger = logging.getLogger(__name__)
 
@@ -2051,7 +2052,7 @@ async def _collect_sql_diffs(
     *,
     datasource: DatasourceConfig,
     sql_models: list[SlayerModel],
-    sql_clients: dict[tuple[str, str], SlayerSQLClient] | None,
+    sql_clients: dict[EngineCacheKey, SlayerSQLClient] | None,
 ) -> dict[str, tuple[ToDeleteEntry | None, set[str]]]:
     """Trial-execute each sql-mode model concurrently and produce its diff."""
     out: dict[str, tuple[ToDeleteEntry | None, set[str]]] = {}
@@ -2087,7 +2088,7 @@ async def validate_datasource(
     *,
     datasource: DatasourceConfig,
     models: list[SlayerModel],
-    sql_clients: dict[tuple[str, str], SlayerSQLClient] | None = None,
+    sql_clients: dict[EngineCacheKey, SlayerSQLClient] | None = None,
 ) -> list[ToDeleteEntry]:
     """Validate every persisted model in ``models`` (all in the same DS)
     against the live schema of ``datasource``. Read-only.

slayer/sql/client.py

@@ -528,12 +528,58 @@ def _get_sync_engine_for_client(self) -> sa.Engine | None:
         self._sync_engine = engine_factory.get_engine(self.datasource)
         return self._sync_engine
 
+    def _discard_sync_engine_on_auth_failure(self, exc: BaseException) -> bool:
+        """Drop the sync engine on a credential rejection; returns whether
+        ``exc`` was one.
+
+        An engine's credentials are fixed at construction, so a revoked grant
+        poisons it permanently — every later call through the cache fails the
+        same way. Evicting lets the next one rebuild from whatever the
+        datasource now carries. Best-effort: cleanup must not displace the
+        original error.
+        """
+        if not _is_auth_failure(exc):
+            return False
+        from slayer.sql import engine_factory  # noqa: PLC0415
+        self._sync_engine = None
+        try:
+            engine_factory.invalidate_engine(self.datasource)
+        except Exception:
+            logger.warning(
+                "Failed to invalidate engine for datasource %r after an "
+                "authentication failure.", self.datasource.name, exc_info=True,
+            )
+        return True
+
+    async def _discard_engines_on_auth_failure(self, exc: BaseException) -> None:
+        """Async-path cleanup: the sync engine plus this client's async one.
+
+        Native-async dialects hold a second pool ``invalidate_engine`` knows
+        nothing about, and disposing it needs a loop — hence the split from the
+        sync variant used by ``execute_sync``.
+        """
+        if not self._discard_sync_engine_on_auth_failure(exc):
+            return
+        await self.aclose()
+
     async def execute(
         self,
         sql: str,
         timeout_seconds: int = 120,
     ) -> list[dict[str, Any]]:
         """Execute SQL asynchronously."""
+        try:
+            return await self._execute(sql=sql, timeout_seconds=timeout_seconds)
+        except Exception as exc:
+            await self._discard_engines_on_auth_failure(exc)
+            raise
+
+    async def _execute(
+        self,
+        *,
+        sql: str,
+        timeout_seconds: int,
+    ) -> list[dict[str, Any]]:
         async_engine = self._get_async_engine()
         db_type = self.datasource.type
         if async_engine is not None:
@@ -565,6 +611,13 @@ async def get_column_types(self, sql: str) -> dict[str, str]:
         Returns {column_name: type_category} where type_category is
         "number", "string", "time", or "boolean".
         """
+        try:
+            return await self._get_column_types(sql=sql)
+        except Exception as exc:
+            await self._discard_engines_on_auth_failure(exc)
+            raise
+
+    async def _get_column_types(self, *, sql: str) -> dict[str, str]:
         async_engine = self._get_async_engine()
         if async_engine is not None:
             return await _get_column_types_async(
@@ -589,14 +642,22 @@ def execute_sync(
         sql: str,
         timeout_seconds: int = 120,
     ) -> list[dict[str, Any]]:
-        """Execute SQL synchronously (for CLI, notebooks, tests)."""
-        return _execute_with_retry_sync(
-            sql=sql,
-            connection_string=self.datasource.get_connection_string(),
-            db_type=self.datasource.type,
-            timeout_seconds=timeout_seconds,
-            engine=self._get_sync_engine_for_client(),
-        )
+        """Execute SQL synchronously (for CLI, notebooks, tests).
+
+        Discards only the sync engine on a credential rejection — no loop here
+        to dispose an async pool on; ``execute`` / ``get_column_types`` cover that.
+        """
+        try:
+            return _execute_with_retry_sync(
+                sql=sql,
+                connection_string=self.datasource.get_connection_string(),
+                db_type=self.datasource.type,
+                timeout_seconds=timeout_seconds,
+                engine=self._get_sync_engine_for_client(),
+            )
+        except Exception as exc:
+            self._discard_sync_engine_on_auth_failure(exc)
+            raise
 
 
 # ---------------------------------------------------------------------------
@@ -647,6 +708,62 @@ def _is_transient_db_error(exc: BaseException) -> bool:
     return any(sig in msg for sig in _TRANSIENT_DB_ERROR_SIGNALS)
 
 
+# Credential rejection, deliberately distinct from _TRANSIENT_DB_ERROR_SIGNALS:
+# retrying is pointless (the credentials are baked into the engine), so the
+# response is to throw the engine away rather than sleep. Kept narrow —
+# Postgres' table-level "permission denied for table" is absent on purpose,
+# since the credentials worked and evicting over it is pure pool churn.
+_AUTH_ERROR_SIGNALS = (
+    "invalid_grant",                      # OAuth refresh token revoked / expired
+    "invalid_client",
+    "unauthorized_client",
+    "token has been expired or revoked",
+    "could not refresh access token",
+    "reauthentication is needed",
+    "authentication failed",
+    "password authentication failed",     # libpq
+    "invalid credentials",
+    "invalid username or password",
+)
+
+# Matched by class name to stay dependency-free: google-auth ships only with
+# the optional 'bigquery' extra.
+_AUTH_ERROR_TYPE_NAMES = frozenset({
+    "RefreshError",             # google.auth.exceptions
+    "DefaultCredentialsError",  # google.auth.exceptions
+    "Unauthorized",             # google.api_core.exceptions — HTTP 401
+})
+
+
+def _is_auth_failure(exc: BaseException) -> bool:
+    """True when the server rejected the *credentials* themselves.
+
+    Walks the cause/context chain plus SQLAlchemy's ``orig`` — the signal
+    surfaces wrapped a layer or two deep.
+    """
+    seen: set[int] = set()
+    pending: list[BaseException] = [exc]
+    while pending:
+        current = pending.pop()
+        if current is None or id(current) in seen:
+            continue
+        seen.add(id(current))
+        if type(current).__name__ in _AUTH_ERROR_TYPE_NAMES:
+            return True
+        text = str(current).lower()
+        if any(signal in text for signal in _AUTH_ERROR_SIGNALS):
+            return True
+        pending.extend(
+            nested for nested in (
+                getattr(current, "orig", None),
+                current.__cause__,
+                current.__context__,
+            )
+            if isinstance(nested, BaseException)
+        )
+    return False
+
+
 async def _retry_with_backoff(
     *,
     sql: str,