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

@@ -465,6 +465,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,27 @@ 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 qualified_name(self, database: str | None, schema: str | None, name: str) -> str:
         """Build a quoted qualified identifier, skipping empty segments.
 

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

@@ -8,6 +8,7 @@
 from sqlit.domains.connections.providers.adapters.base import (
     ColumnInfo,
     CursorBasedAdapter,
+    ForeignKeyInfo,
     IndexInfo,
     SequenceInfo,
     TableInfo,
@@ -56,6 +57,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
@@ -361,6 +367,101 @@ 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, ccu.table_schema, "
+            f"       ccu.table_name, ccu.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.CONSTRAINT_COLUMN_USAGE ccu "
+            f"  ON tc.constraint_catalog = ccu.constraint_catalog "
+            f"  AND tc.constraint_schema = ccu.constraint_schema "
+            f"  AND tc.constraint_name = ccu.constraint_name "
+            f"  AND kcu.position_in_unique_constraint = ccu.ordinal_position "
+            f"WHERE tc.constraint_type = 'FOREIGN KEY' "
+            f"  AND tc.table_name = @table_name "
+            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, ccu.column_name "
+            f"FROM `{dataset}`.INFORMATION_SCHEMA.CONSTRAINT_COLUMN_USAGE ccu "
+            f"JOIN `{dataset}`.INFORMATION_SCHEMA.TABLE_CONSTRAINTS tc "
+            f"  ON tc.constraint_catalog = ccu.constraint_catalog "
+            f"  AND tc.constraint_schema = ccu.constraint_schema "
+            f"  AND tc.constraint_name = ccu.constraint_name "
+            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"  AND kcu.position_in_unique_constraint = ccu.ordinal_position "
+            f"WHERE tc.constraint_type = 'FOREIGN KEY' "
+            f"  AND ccu.table_name = @table_name",
+            {"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}`"

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

@@ -9,6 +9,7 @@
 from sqlit.domains.connections.providers.adapters.base import (
     ColumnInfo,
     DatabaseAdapter,
+    ForeignKeyInfo,
     IndexInfo,
     SequenceInfo,
     TableInfo,
@@ -70,6 +71,10 @@ def supports_stored_procedures(self) -> bool:
         """D1 is SQLite-based and does not support stored procedures."""
         return False
 
+    @property
+    def supports_foreign_keys(self) -> bool:
+        return True
+
     @property
     def supports_cross_database_queries(self) -> bool:
         """D1 databases are isolated; cross-database queries not supported."""
@@ -261,6 +266,93 @@ def get_sequences(self, conn: D1Connection, database: str | None = None) -> list
         """D1/SQLite doesn't support sequences - return empty list."""
         return []
 
+    def get_foreign_keys(
+        self,
+        conn: D1Connection,
+        table: str,
+        database: str | None = None,
+        schema: str | None = None,
+    ) -> list[ForeignKeyInfo]:
+        """SQLite-compatible PRAGMA foreign_key_list over D1's HTTP query API."""
+        result = self._execute(conn, f"PRAGMA foreign_key_list({self.quote_identifier(table)});")
+        rows = result.get("results", [])
+        if not isinstance(rows, list):
+            return []
+        out: list[ForeignKeyInfo] = []
+        for row in rows:
+            if not isinstance(row, dict):
+                continue
+            fk_id = row.get("id")
+            seq = row.get("seq")
+            ref_table = row.get("table")
+            from_col = row.get("from")
+            to_col = row.get("to")
+            if not isinstance(ref_table, str) or not isinstance(from_col, str) or not isinstance(to_col, str):
+                continue
+            out.append(
+                ForeignKeyInfo(
+                    owner_table=table,
+                    column=from_col,
+                    referenced_table=ref_table,
+                    referenced_column=to_col,
+                    constraint_name=f"fk_{fk_id}",
+                    ordinal=(int(seq) if isinstance(seq, int) else 0) + 1,
+                )
+            )
+        return out
+
+    def get_referencing_foreign_keys(
+        self,
+        conn: D1Connection,
+        table: str,
+        database: str | None = None,
+        schema: str | None = None,
+    ) -> list[ForeignKeyInfo]:
+        """Scan every user table for FKs pointing at `table`."""
+        tables_result = self._execute(conn, "PRAGMA table_list;")
+        tables_rows = tables_result.get("results", [])
+        if not isinstance(tables_rows, list):
+            return []
+        target_lower = table.lower()
+        results: list[ForeignKeyInfo] = []
+        for tr in tables_rows:
+            if not isinstance(tr, dict):
+                continue
+            if tr.get("type") != "table":
+                continue
+            other = tr.get("name", "")
+            if not isinstance(other, str) or not other or other.startswith("sqlite_"):
+                continue
+            if other.lower() == target_lower:
+                continue
+            fk_result = self._execute(conn, f"PRAGMA foreign_key_list({self.quote_identifier(other)});")
+            fk_rows = fk_result.get("results", [])
+            if not isinstance(fk_rows, list):
+                continue
+            for row in fk_rows:
+                if not isinstance(row, dict):
+                    continue
+                ref_table = row.get("table")
+                if not isinstance(ref_table, str) or ref_table.lower() != target_lower:
+                    continue
+                from_col = row.get("from")
+                to_col = row.get("to")
+                if not isinstance(from_col, str) or not isinstance(to_col, str):
+                    continue
+                fk_id = row.get("id")
+                seq = row.get("seq")
+                results.append(
+                    ForeignKeyInfo(
+                        owner_table=other,
+                        column=from_col,
+                        referenced_table=table,
+                        referenced_column=to_col,
+                        constraint_name=f"{other}_fk_{fk_id}",
+                        ordinal=(int(seq) if isinstance(seq, int) else 0) + 1,
+                    )
+                )
+        return results
+
     def quote_identifier(self, name: str) -> str:
         """Quotes an identifier with double quotes."""
         return f'"{name}"'