feat(results): foreign-key navigation between related rows

sqlit/core/input_context.py

@@ -34,3 +34,9 @@ class InputContext:
     has_results: bool
     stacked_result_count: int = 0
     count_buffer: str = ""
+    # True when the column under the results-table cursor is an outgoing FK column
+    # (i.e. pressing the navigate-FK key would take you to the referenced row).
+    cursor_column_is_foreign_key: bool = False
+    # True when the column under the cursor is referenced by some other table's FK
+    # (i.e. pressing the navigate-referrers key would open the picker).
+    cursor_column_is_foreign_key_target: bool = False

sqlit/core/keymap.py

@@ -489,6 +489,8 @@ def _build_action_keys(self) -> list[ActionKeyDef]:
             ActionKeyDef("V", "view_cell_full", "results"),
             ActionKeyDef("u", "edit_cell", "results"),
             ActionKeyDef("d", "delete_row", "results"),
+            ActionKeyDef("o", "navigate_fk", "results"),
+            ActionKeyDef("O", "navigate_referrers", "results"),
             ActionKeyDef("y", "results_yank_leader_key", "results"),
             ActionKeyDef("x", "clear_results", "results"),
             ActionKeyDef("slash", "results_filter", "results"),

sqlit/domains/connections/app/mock_provider.py

@@ -94,6 +94,7 @@ def _display_info(config: ConnectionConfig) -> str:
         supports_sequences=bool(getattr(adapter, "supports_sequences", False)),
         default_schema=str(getattr(adapter, "default_schema", "")),
         system_databases=frozenset(getattr(adapter, "system_databases", frozenset())),
+        supports_foreign_keys=bool(getattr(adapter, "supports_foreign_keys", False)),
     )
 
     def apply_database_override(config: ConnectionConfig, database: str | None) -> ConnectionConfig:

sqlit/domains/connections/providers/adapter_provider.py

@@ -92,6 +92,7 @@ def build_adapter_provider(spec: ProviderSpec, schema: ConnectionSchema, adapter
         supports_sequences=bool(getattr(adapter, "supports_sequences", False)),
         default_schema=str(getattr(adapter, "default_schema", "")),
         system_databases=frozenset(getattr(adapter, "system_databases", frozenset())),
+        supports_foreign_keys=bool(getattr(adapter, "supports_foreign_keys", False)),
     )
 
     def display_info(config: ConnectionConfig) -> str:

sqlit/domains/connections/providers/adapters/base.py

@@ -69,6 +69,41 @@ class SequenceInfo:
     name: str
 
 
+@dataclass(frozen=True)
+class ForeignKeyInfo:
+    """A single foreign-key relationship between two columns.
+
+    The FK is declared on the *owner* (child) side, which has `column`,
+    and points to the *referenced* (parent) side, which has
+    `referenced_column` on `referenced_table`.
+
+    For outgoing FKs returned by `get_foreign_keys(T)`, every entry has
+    `owner_table == T`. For incoming refs returned by
+    `get_referencing_foreign_keys(T)`, every entry has
+    `referenced_table == T` and `owner_table` is whatever table is
+    pointing in.
+
+    Composite FKs are represented as multiple entries sharing
+    `constraint_name`, ordered by `ordinal`. The navigation UI only
+    auto-jumps single-column FKs; composite ones surface but require
+    user disambiguation.
+
+    Schema/database fields are empty strings on engines that don't
+    expose them (e.g. SQLite).
+    """
+
+    owner_table: str
+    column: str
+    referenced_table: str
+    referenced_column: str
+    owner_schema: str = ""
+    owner_database: str = ""
+    referenced_schema: str = ""
+    referenced_database: str = ""
+    constraint_name: str = ""
+    ordinal: int = 1
+
+
 # Type alias for table/view info: (schema, name)
 TableInfo = tuple[str, str]
 
@@ -194,6 +229,16 @@ def supports_sequences(self) -> bool:
         """
         return False
 
+    @property
+    def supports_foreign_keys(self) -> bool:
+        """Whether this database exposes foreign-key relationships via catalog queries.
+
+        Override in subclasses that implement `get_foreign_keys` /
+        `get_referencing_foreign_keys`. Defaults to False so the FK
+        navigation UI is hidden for adapters without an implementation.
+        """
+        return False
+
     @property
     def supports_process_worker(self) -> bool:
         """Whether this adapter supports running queries in a separate process."""
@@ -343,6 +388,36 @@ def get_sequences(self, conn: Any, database: str | None = None) -> list[Sequence
         """
         pass
 
+    def get_foreign_keys(
+        self,
+        conn: Any,
+        table: str,
+        database: str | None = None,
+        schema: str | None = None,
+    ) -> list[ForeignKeyInfo]:
+        """Get outgoing foreign keys defined on `table` (FKs whose owner is this table).
+
+        Default returns []. Override in adapters that support FK introspection
+        and also set `supports_foreign_keys = True`.
+        """
+        return []
+
+    def get_referencing_foreign_keys(
+        self,
+        conn: Any,
+        table: str,
+        database: str | None = None,
+        schema: str | None = None,
+    ) -> list[ForeignKeyInfo]:
+        """Get incoming foreign keys that reference `table`.
+
+        Each entry's `column` is the FK column on the *referring* table;
+        `referenced_table` and `referenced_column` point back to this table.
+
+        Default returns []. Override in adapters that support FK introspection.
+        """
+        return []
+
     def get_index_definition(
         self, conn: Any, index_name: str, table_name: str, database: str | None = None
     ) -> dict[str, Any]:
@@ -417,6 +492,50 @@ def quote_identifier(self, name: str) -> str:
         """Quote an identifier (table name, column name, etc.)."""
         pass
 
+    def quote_literal(self, value: Any) -> str:
+        """Render a Python value as a SQL literal.
+
+        Used by the FK-navigation feature to build WHERE clauses against
+        cell values. Default handles None, bool, int/float, str, and bytes
+        with single-quote escaping (the SQL-standard form supported by
+        SQLite, Postgres, MySQL with NO_BACKSLASH_ESCAPES, and MSSQL).
+        Subclasses can override for engines with different conventions
+        (e.g. mysql backslash escaping, mssql bytes as 0x...).
+        """
+        if value is None:
+            return "NULL"
+        if isinstance(value, bool):
+            return "TRUE" if value else "FALSE"
+        if isinstance(value, int | float):
+            return repr(value) if isinstance(value, float) else str(value)
+        if isinstance(value, bytes | bytearray | memoryview):
+            return "X'" + bytes(value).hex() + "'"
+        text = str(value)
+        return "'" + text.replace("'", "''") + "'"
+
+    def build_filtered_select_query(
+        self,
+        table: str,
+        column: str,
+        value: Any,
+        limit: int,
+        database: str | None = None,
+        schema: str | None = None,
+    ) -> str:
+        """Build a limited SELECT filtered to one column value."""
+        qualified = self.catalog_qualified_name(database, schema, table)
+        return f"SELECT * FROM {qualified} WHERE {self.quote_identifier(column)} = {self.quote_literal(value)} LIMIT {limit}"
+
+    def catalog_qualified_name(self, database: str | None, schema: str | None, name: str) -> str:
+        """Quote an exact catalog identity without omitting the default schema."""
+        parts: list[str] = []
+        if database and self.supports_cross_database_queries:
+            parts.append(self.quote_identifier(database))
+        if schema:
+            parts.append(self.quote_identifier(schema))
+        parts.append(self.quote_identifier(name))
+        return ".".join(parts)
+
     def format_autocomplete_identifier(self, name: str) -> str:
         """Format an identifier for insertion from autocomplete.
 

sqlit/domains/connections/providers/bigquery/adapter.py

@@ -3,11 +3,13 @@
 from __future__ import annotations
 
 import os
-from typing import TYPE_CHECKING, Any, Iterable
+from collections.abc import Iterable
+from typing import TYPE_CHECKING, Any
 
 from sqlit.domains.connections.providers.adapters.base import (
     ColumnInfo,
     CursorBasedAdapter,
+    ForeignKeyInfo,
     IndexInfo,
     SequenceInfo,
     TableInfo,
@@ -17,6 +19,7 @@
 if TYPE_CHECKING:
     from google.cloud import bigquery
     from google.cloud.bigquery.dbapi import Connection as BigQueryConnection
+
     from sqlit.domains.connections.domain.config import ConnectionConfig
 
 
@@ -56,6 +59,11 @@ def supports_triggers(self) -> bool:
     def supports_indexes(self) -> bool:
         return False
 
+    @property
+    def supports_foreign_keys(self) -> bool:
+        # Informational FKs, GA since 2023. Not enforced but queryable.
+        return True
+
     @property
     def supports_cross_database_queries(self) -> bool:
         # BigQuery supports cross-dataset queries
@@ -180,11 +188,11 @@ def connect(self, config: ConnectionConfig) -> BigQueryConnection:
                 client.default_query_job_config = job_config
 
         conn = dbapi.connect(client=client)
-        setattr(conn, "_sqlit_bq_client", client)
+        conn._sqlit_bq_client = client
         if default_dataset:
-            setattr(conn, "_sqlit_bq_default_dataset", default_dataset)
+            conn._sqlit_bq_default_dataset = default_dataset
         if job_config:
-            setattr(conn, "_sqlit_bq_job_config", job_config)
+            conn._sqlit_bq_job_config = job_config
 
         return conn
 
@@ -361,10 +369,119 @@ def get_sequences(self, conn: Any, database: str | None = None) -> list[Sequence
         """BigQuery doesn't support sequences."""
         return []
 
+    def get_foreign_keys(
+        self,
+        conn: Any,
+        table: str,
+        database: str | None = None,
+        schema: str | None = None,
+    ) -> list[ForeignKeyInfo]:
+        """List FKs via dataset-qualified INFORMATION_SCHEMA views (BigQuery).
+
+        Joins KEY_COLUMN_USAGE (child columns) to CONSTRAINT_COLUMN_USAGE
+        (parent columns) by POSITION_IN_UNIQUE_CONSTRAINT, scoped to
+        constraint_type = 'FOREIGN KEY'. The dataset (schema) provides the
+        catalog qualification; BigQuery requires every INFORMATION_SCHEMA
+        access to be dataset-qualified.
+        """
+        dataset = schema or database
+        if not dataset:
+            return []
+        cursor = conn.cursor()
+        cursor.execute(
+            f"SELECT kcu.constraint_name, kcu.ordinal_position, "
+            f"       kcu.column_name, ukcu.table_schema, "
+            f"       ukcu.table_name, ukcu.column_name "
+            f"FROM `{dataset}`.INFORMATION_SCHEMA.TABLE_CONSTRAINTS tc "
+            f"JOIN `{dataset}`.INFORMATION_SCHEMA.KEY_COLUMN_USAGE kcu "
+            f"  ON tc.constraint_catalog = kcu.constraint_catalog "
+            f"  AND tc.constraint_schema = kcu.constraint_schema "
+            f"  AND tc.constraint_name = kcu.constraint_name "
+            f"JOIN `{dataset}`.INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS rc "
+            f"  ON tc.constraint_catalog = rc.constraint_catalog "
+            f"  AND tc.constraint_schema = rc.constraint_schema "
+            f"  AND tc.constraint_name = rc.constraint_name "
+            f"JOIN `{dataset}`.INFORMATION_SCHEMA.KEY_COLUMN_USAGE ukcu "
+            f"  ON ukcu.constraint_catalog = rc.unique_constraint_catalog "
+            f"  AND ukcu.constraint_schema = rc.unique_constraint_schema "
+            f"  AND ukcu.constraint_name = rc.unique_constraint_name "
+            f"  AND ukcu.ordinal_position = kcu.position_in_unique_constraint "
+            f"WHERE tc.constraint_type = 'FOREIGN KEY' "
+            f"  AND tc.table_name = %(table_name)s "
+            f"ORDER BY kcu.constraint_name, kcu.ordinal_position",
+            {"table_name": table},
+        )
+        return [
+            ForeignKeyInfo(
+                owner_table=table,
+                column=row[2],
+                referenced_table=row[4],
+                referenced_column=row[5],
+                owner_schema=dataset,
+                referenced_schema=row[3] or "",
+                constraint_name=row[0],
+                ordinal=int(row[1]),
+            )
+            for row in cursor.fetchall()
+        ]
+
+    def get_referencing_foreign_keys(
+        self,
+        conn: Any,
+        table: str,
+        database: str | None = None,
+        schema: str | None = None,
+    ) -> list[ForeignKeyInfo]:
+        dataset = schema or database
+        if not dataset:
+            return []
+        cursor = conn.cursor()
+        cursor.execute(
+            f"SELECT kcu.constraint_name, kcu.ordinal_position, "
+            f"       kcu.table_schema, kcu.table_name, "
+            f"       kcu.column_name, ukcu.column_name "
+            f"FROM `{dataset}`.INFORMATION_SCHEMA.TABLE_CONSTRAINTS tc "
+            f"JOIN `{dataset}`.INFORMATION_SCHEMA.KEY_COLUMN_USAGE kcu "
+            f"  ON tc.constraint_catalog = kcu.constraint_catalog "
+            f"  AND tc.constraint_schema = kcu.constraint_schema "
+            f"  AND tc.constraint_name = kcu.constraint_name "
+            f"JOIN `{dataset}`.INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS rc "
+            f"  ON tc.constraint_catalog = rc.constraint_catalog "
+            f"  AND tc.constraint_schema = rc.constraint_schema "
+            f"  AND tc.constraint_name = rc.constraint_name "
+            f"JOIN `{dataset}`.INFORMATION_SCHEMA.KEY_COLUMN_USAGE ukcu "
+            f"  ON ukcu.constraint_catalog = rc.unique_constraint_catalog "
+            f"  AND ukcu.constraint_schema = rc.unique_constraint_schema "
+            f"  AND ukcu.constraint_name = rc.unique_constraint_name "
+            f"  AND ukcu.ordinal_position = kcu.position_in_unique_constraint "
+            f"WHERE tc.constraint_type = 'FOREIGN KEY' "
+            f"  AND ukcu.table_name = %(table_name)s",
+            {"table_name": table},
+        )
+        return [
+            ForeignKeyInfo(
+                owner_table=row[3],
+                column=row[4],
+                referenced_table=table,
+                referenced_column=row[5],
+                owner_schema=row[2] or "",
+                referenced_schema=dataset,
+                constraint_name=row[0],
+                ordinal=int(row[1]),
+            )
+            for row in cursor.fetchall()
+        ]
+
     def quote_identifier(self, name: str) -> str:
         """Quote an identifier for BigQuery."""
         return f"`{name}`"
 
+    def quote_literal(self, value: Any) -> str:
+        """Render BigQuery strings without allowing backslash escapes to alter quoting."""
+        if not isinstance(value, str):
+            return super().quote_literal(value)
+        return "'" + value.replace("\\", "\\\\").replace("'", "\\'") + "'"
+
     def build_select_query(
         self, table: str, limit: int, database: str | None = None, schema: str | None = None
     ) -> str: