Add ER diagram with FK introspection across all databases

sqlit/core/keymap.py

@@ -196,6 +196,7 @@ def _build_leader_commands(self) -> list[LeaderCommandDef]:
             # Actions
             LeaderCommandDef("z", "cancel_operation", "Cancel", "Actions", guard="query_executing"),
             LeaderCommandDef("t", "change_theme", "Change Theme", "Actions"),
+            LeaderCommandDef("d", "show_diagram_picker", "ER Diagram", "Actions", guard="has_connection"),
             LeaderCommandDef("h", "show_help", "Help", "Actions"),
             LeaderCommandDef("space", "telescope", "Telescope", "Actions"),
             LeaderCommandDef("slash", "telescope_filter", "Telescope Search", "Actions"),
@@ -309,6 +310,7 @@ def _build_action_keys(self) -> list[ActionKeyDef]:
             ActionKeyDef("v", "exit_tree_visual_mode", "tree_visual", primary=False),
             ActionKeyDef("escape", "clear_connection_selection", "tree"),
             ActionKeyDef("s", "select_table", "tree"),
+            ActionKeyDef("S", "show_diagram", "tree"),
             ActionKeyDef("f", "refresh_tree", "tree"),
             ActionKeyDef("R", "refresh_tree", "tree", primary=False),
             ActionKeyDef("e", "edit_connection", "tree"),

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

@@ -69,6 +69,19 @@ class SequenceInfo:
     name: str
 
 
+@dataclass
+class ForeignKeyInfo:
+    """Information about a foreign key relationship."""
+
+    constraint_name: str
+    source_table: str
+    source_column: str
+    target_table: str
+    target_column: str
+    source_schema: str = ""
+    target_schema: str = ""
+
+
 # Type alias for table/view info: (schema, name)
 TableInfo = tuple[str, str]
 
@@ -343,6 +356,14 @@ def get_sequences(self, conn: Any, database: str | None = None) -> list[Sequence
         """
         pass
 
+    def get_foreign_keys(self, conn: Any, database: str | None = None) -> list[ForeignKeyInfo]:
+        """Get list of foreign keys in the database.
+
+        Returns:
+            List of ForeignKeyInfo objects describing FK relationships.
+        """
+        return []
+
     def get_index_definition(
         self, conn: Any, index_name: str, table_name: str, database: str | None = None
     ) -> dict[str, Any]:
@@ -518,6 +539,7 @@ def execute_non_query(self, conn: Any, query: str) -> int:
 __all__ = [
     "ColumnInfo",
     "DatabaseAdapter",
+    "ForeignKeyInfo",
     "IndexInfo",
     "SequenceInfo",
     "TableInfo",

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

@@ -7,6 +7,7 @@
 from sqlit.domains.connections.providers.adapters.base import (
     ColumnInfo,
     CursorBasedAdapter,
+    ForeignKeyInfo,
     IndexInfo,
     SequenceInfo,
     TableInfo,
@@ -151,6 +152,34 @@ def get_procedures(self, conn: Any, database: str | None = None) -> list[str]:
         )
         return [row[0] for row in cursor.fetchall()]
 
+    def get_foreign_keys(self, conn: Any, database: str | None = None) -> list[ForeignKeyInfo]:
+        """Get foreign keys from DB2."""
+        cursor = conn.cursor()
+        cursor.execute(
+            "SELECT r.constname, r.tabschema, r.tabname, fk.colname, "
+            "r.reftabschema, r.reftabname, pk.colname "
+            "FROM syscat.references r "
+            "JOIN syscat.keycoluse fk ON r.constname = fk.constname "
+            "AND r.tabschema = fk.tabschema AND r.tabname = fk.tabname "
+            "JOIN syscat.keycoluse pk ON r.refkeyname = pk.constname "
+            "AND r.reftabschema = pk.tabschema AND r.reftabname = pk.tabname "
+            "AND fk.colseq = pk.colseq "
+            "WHERE r.tabschema NOT LIKE 'SYS%' "
+            "ORDER BY r.tabname, r.constname"
+        )
+        return [
+            ForeignKeyInfo(
+                constraint_name=row[0],
+                source_schema=row[1],
+                source_table=row[2],
+                source_column=row[3],
+                target_schema=row[4],
+                target_table=row[5],
+                target_column=row[6],
+            )
+            for row in cursor.fetchall()
+        ]
+
     def get_indexes(self, conn: Any, database: str | None = None) -> list[IndexInfo]:
         cursor = conn.cursor()
         cursor.execute(

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

@@ -7,6 +7,7 @@
 from sqlit.domains.connections.providers.adapters.base import (
     ColumnInfo,
     DatabaseAdapter,
+    ForeignKeyInfo,
     IndexInfo,
     SequenceInfo,
     TableInfo,
@@ -155,6 +156,45 @@ def get_indexes(self, conn: Any, database: str | None = None) -> list[IndexInfo]
             for row in result.fetchall()
         ]
 
+    def get_foreign_keys(self, conn: Any, database: str | None = None) -> list[ForeignKeyInfo]:
+        """Get foreign keys from DuckDB."""
+        result = conn.execute(
+            "SELECT "
+            "  tc.constraint_name, "
+            "  tc.table_schema AS source_schema, "
+            "  tc.table_name AS source_table, "
+            "  kcu.column_name AS source_column, "
+            "  kcu2.table_schema AS target_schema, "
+            "  kcu2.table_name AS target_table, "
+            "  kcu2.column_name AS target_column "
+            "FROM information_schema.table_constraints tc "
+            "JOIN information_schema.key_column_usage kcu "
+            "  ON tc.constraint_name = kcu.constraint_name "
+            "  AND tc.table_schema = kcu.table_schema "
+            "JOIN information_schema.referential_constraints rc "
+            "  ON tc.constraint_name = rc.constraint_name "
+            "  AND tc.constraint_schema = rc.constraint_schema "
+            "JOIN information_schema.key_column_usage kcu2 "
+            "  ON rc.unique_constraint_name = kcu2.constraint_name "
+            "  AND rc.unique_constraint_schema = kcu2.constraint_schema "
+            "  AND kcu.ordinal_position = kcu2.ordinal_position "
+            "WHERE tc.constraint_type = 'FOREIGN KEY' "
+            "AND tc.table_schema NOT IN ('pg_catalog', 'information_schema') "
+            "ORDER BY tc.table_name, tc.constraint_name"
+        )
+        return [
+            ForeignKeyInfo(
+                constraint_name=row[0],
+                source_schema=row[1],
+                source_table=row[2],
+                source_column=row[3],
+                target_schema=row[4],
+                target_table=row[5],
+                target_column=row[6],
+            )
+            for row in result.fetchall()
+        ]
+
     def get_triggers(self, conn: Any, database: str | None = None) -> list[TriggerInfo]:
         """DuckDB doesn't support triggers - return empty list."""
         return []

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

@@ -5,6 +5,7 @@
 from sqlit.domains.connections.providers.adapters.base import (
     ColumnInfo,
     CursorBasedAdapter,
+    ForeignKeyInfo,
     IndexInfo,
     SequenceInfo,
     TableInfo,
@@ -166,6 +167,31 @@ def get_index_definition(
             "definition": " ".join(definition_parts),
         }
 
+    def get_foreign_keys(self, conn: Any, database: str | None = None) -> list[ForeignKeyInfo]:
+        """Get foreign keys from Firebird."""
+        cursor = conn.cursor()
+        cursor.execute(
+            "SELECT rc.rdb$constraint_name, rc.rdb$relation_name, sg.rdb$field_name, "
+            "rfc.rdb$relation_name, sg2.rdb$field_name "
+            "FROM rdb$relation_constraints rc "
+            "JOIN rdb$ref_constraints ref ON rc.rdb$constraint_name = ref.rdb$constraint_name "
+            "JOIN rdb$relation_constraints rfc ON ref.rdb$const_name_uq = rfc.rdb$constraint_name "
+            "JOIN rdb$index_segments sg ON rc.rdb$index_name = sg.rdb$index_name "
+            "JOIN rdb$index_segments sg2 ON rfc.rdb$index_name = sg2.rdb$index_name "
+            "AND sg.rdb$field_position = sg2.rdb$field_position "
+            "WHERE rc.rdb$constraint_type = 'FOREIGN KEY'"
+        )
+        return [
+            ForeignKeyInfo(
+                constraint_name=row[0].rstrip(),
+                source_table=row[1].rstrip(),
+                source_column=row[2].rstrip(),
+                target_table=row[3].rstrip(),
+                target_column=row[4].rstrip(),
+            )
+            for row in cursor.fetchall()
+        ]
+
     def get_sequences(self, conn: Any, database: str | None = None) -> list[SequenceInfo]:
         cursor = conn.cursor()
         cursor.execute("SELECT rdb$generator_name FROM rdb$generators WHERE rdb$system_flag = 0")

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

@@ -7,6 +7,7 @@
 from sqlit.domains.connections.providers.adapters.base import (
     ColumnInfo,
     CursorBasedAdapter,
+    ForeignKeyInfo,
     IndexInfo,
     SequenceInfo,
     TableInfo,
@@ -143,6 +144,30 @@ def get_procedures(self, conn: Any, database: str | None = None) -> list[str]:
         )
         return [row[0] for row in cursor.fetchall()]
 
+    def get_foreign_keys(self, conn: Any, database: str | None = None) -> list[ForeignKeyInfo]:
+        """Get foreign keys from SAP HANA."""
+        cursor = conn.cursor()
+        cursor.execute(
+            "SELECT constraint_name, schema_name, table_name, column_name, "
+            "referenced_schema_name, referenced_table_name, referenced_column_name "
+            "FROM sys.referential_constraints "
+            "WHERE schema_name NOT LIKE '_SYS%' "
+            "AND schema_name NOT IN ('SYS', 'SYSTEM') "
+            "ORDER BY table_name, constraint_name"
+        )
+        return [
+            ForeignKeyInfo(
+                constraint_name=row[0],
+                source_schema=row[1],
+                source_table=row[2],
+                source_column=row[3],
+                target_schema=row[4],
+                target_table=row[5],
+                target_column=row[6],
+            )
+            for row in cursor.fetchall()
+        ]
+
     def get_indexes(self, conn: Any, database: str | None = None) -> list[IndexInfo]:
         cursor = conn.cursor()
         cursor.execute(

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

@@ -7,6 +7,7 @@
 
 from sqlit.domains.connections.providers.adapters.base import (
     ColumnInfo,
+    ForeignKeyInfo,
     IndexInfo,
     SequenceInfo,
     TableInfo,
@@ -218,6 +219,37 @@ def get_procedures(self, conn: Any, database: str | None = None) -> list[str]:
             )
         return [row[0] for row in cursor.fetchall()]
 
+    def get_foreign_keys(self, conn: Any, database: str | None = None) -> list[ForeignKeyInfo]:
+        """Get foreign keys from MariaDB (uses ? placeholders)."""
+        cursor = conn.cursor()
+        if database:
+            cursor.execute(
+                "SELECT constraint_name, table_name, column_name, "
+                "referenced_table_name, referenced_column_name "
+                "FROM information_schema.key_column_usage "
+                "WHERE table_schema = ? AND referenced_table_name IS NOT NULL "
+                "ORDER BY table_name, constraint_name",
+                (database,),
+            )
+        else:
+            cursor.execute(
+                "SELECT constraint_name, table_name, column_name, "
+                "referenced_table_name, referenced_column_name "
+                "FROM information_schema.key_column_usage "
+                "WHERE table_schema = DATABASE() AND referenced_table_name IS NOT NULL "
+                "ORDER BY table_name, constraint_name"
+            )
+        return [
+            ForeignKeyInfo(
+                constraint_name=row[0],
+                source_table=row[1],
+                source_column=row[2],
+                target_table=row[3],
+                target_column=row[4],
+            )
+            for row in cursor.fetchall()
+        ]
+
     def get_indexes(self, conn: Any, database: str | None = None) -> list[IndexInfo]:
         """Get indexes from MariaDB (uses ? placeholders)."""
         cursor = conn.cursor()

sqlit/domains/connections/providers/model.py

@@ -116,6 +116,10 @@ def get_sequence_definition(self, conn: Any, sequence_name: str, database: str |
 
 
 @runtime_checkable
+class ForeignKeyInspector(Protocol):
+    def get_foreign_keys(self, conn: Any, database: str | None = None) -> list[Any]: ...
+
+
 @runtime_checkable
 class ProcedureInspector(Protocol):
     def get_procedures(self, conn: Any, database: str | None = None) -> list[Any]: ...

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

@@ -4,7 +4,7 @@
 
 from typing import TYPE_CHECKING, Any
 
-from sqlit.domains.connections.providers.adapters.base import TableInfo
+from sqlit.domains.connections.providers.adapters.base import ForeignKeyInfo, TableInfo
 from sqlit.domains.connections.providers.duckdb.adapter import DuckDBAdapter
 
 if TYPE_CHECKING:
@@ -115,6 +115,51 @@ def get_views(self, conn: Any, database: str | None = None) -> list[TableInfo]:
             )
         return [(row[0], row[1]) for row in result.fetchall()]
 
+    def get_foreign_keys(self, conn: Any, database: str | None = None) -> list[ForeignKeyInfo]:
+        """Get foreign keys from a specific MotherDuck database."""
+        query = (
+            "SELECT "
+            "  tc.constraint_name, "
+            "  tc.table_schema AS source_schema, "
+            "  tc.table_name AS source_table, "
+            "  kcu.column_name AS source_column, "
+            "  kcu2.table_schema AS target_schema, "
+            "  kcu2.table_name AS target_table, "
+            "  kcu2.column_name AS target_column "
+            "FROM information_schema.table_constraints tc "
+            "JOIN information_schema.key_column_usage kcu "
+            "  ON tc.constraint_name = kcu.constraint_name "
+            "  AND tc.table_schema = kcu.table_schema "
+            "JOIN information_schema.referential_constraints rc "
+            "  ON tc.constraint_name = rc.constraint_name "
+            "  AND tc.constraint_schema = rc.constraint_schema "
+            "JOIN information_schema.key_column_usage kcu2 "
+            "  ON rc.unique_constraint_name = kcu2.constraint_name "
+            "  AND rc.unique_constraint_schema = kcu2.constraint_schema "
+            "  AND kcu.ordinal_position = kcu2.ordinal_position "
+            "WHERE tc.constraint_type = 'FOREIGN KEY' "
+            "AND tc.table_schema NOT IN ('pg_catalog', 'information_schema') "
+        )
+        if database:
+            query += "AND tc.table_catalog = ? "
+            query += "ORDER BY tc.table_name, tc.constraint_name"
+            result = conn.execute(query, (database,))
+        else:
+            query += "ORDER BY tc.table_name, tc.constraint_name"
+            result = conn.execute(query)
+        return [
+            ForeignKeyInfo(
+                constraint_name=row[0],
+                source_schema=row[1],
+                source_table=row[2],
+                source_column=row[3],
+                target_schema=row[4],
+                target_table=row[5],
+                target_column=row[6],
+            )
+            for row in result.fetchall()
+        ]
+
     def build_select_query(
         self, table: str, limit: int, database: str | None = None, schema: str | None = None
     ) -> str: