refactor: move dependency graph queries into adapter methods

Extract the backend-specific primary key and foreign key SQL from
dependencies.py into load_primary_keys_sql() and load_foreign_keys_sql()
adapter methods, eliminating the if/else backend fork.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: dimitri-yatsenko

src/datajoint/adapters/base.py

@@ -727,6 +727,55 @@ def get_foreign_keys_sql(self, schema_name: str, table_name: str) -> str:
         """
         ...
 
+    @abstractmethod
+    def load_primary_keys_sql(self, schemas_list: str, like_pattern: str) -> str:
+        """
+        Generate query to load primary key columns for all tables across schemas.
+
+        Used by the dependency graph to build the schema graph.
+
+        Parameters
+        ----------
+        schemas_list : str
+            Comma-separated, quoted schema names for an IN clause.
+        like_pattern : str
+            SQL LIKE pattern to exclude (e.g., "'~%%'" for internal tables).
+
+        Returns
+        -------
+        str
+            SQL query returning rows with columns:
+            - tab: fully qualified table name (quoted)
+            - column_name: primary key column name
+        """
+        ...
+
+    @abstractmethod
+    def load_foreign_keys_sql(self, schemas_list: str, like_pattern: str) -> str:
+        """
+        Generate query to load foreign key relationships across schemas.
+
+        Used by the dependency graph to build the schema graph.
+
+        Parameters
+        ----------
+        schemas_list : str
+            Comma-separated, quoted schema names for an IN clause.
+        like_pattern : str
+            SQL LIKE pattern to exclude (e.g., "'~%%'" for internal tables).
+
+        Returns
+        -------
+        str
+            SQL query returning rows (as dicts) with columns:
+            - constraint_name: FK constraint name
+            - referencing_table: fully qualified child table name (quoted)
+            - referenced_table: fully qualified parent table name (quoted)
+            - column_name: FK column in child table
+            - referenced_column_name: referenced column in parent table
+        """
+        ...
+
     @abstractmethod
     def get_constraint_info_sql(self, constraint_name: str, schema_name: str, table_name: str) -> str:
         """

src/datajoint/adapters/mysql.py

@@ -656,6 +656,32 @@ def get_foreign_keys_sql(self, schema_name: str, table_name: str) -> str:
             f"ORDER BY constraint_name, ordinal_position"
         )
 
+    def load_primary_keys_sql(self, schemas_list: str, like_pattern: str) -> str:
+        """Query to load all primary key columns across schemas."""
+        tab_expr = "concat('`', table_schema, '`.`', table_name, '`')"
+        return (
+            f"SELECT {tab_expr} as tab, column_name "
+            f"FROM information_schema.key_column_usage "
+            f"WHERE table_name NOT LIKE {like_pattern} "
+            f"AND table_schema in ({schemas_list}) "
+            f"AND constraint_name='PRIMARY'"
+        )
+
+    def load_foreign_keys_sql(self, schemas_list: str, like_pattern: str) -> str:
+        """Query to load all foreign key relationships across schemas."""
+        tab_expr = "concat('`', table_schema, '`.`', table_name, '`')"
+        ref_tab_expr = "concat('`', referenced_table_schema, '`.`', referenced_table_name, '`')"
+        return (
+            f"SELECT constraint_name, "
+            f"{tab_expr} as referencing_table, "
+            f"{ref_tab_expr} as referenced_table, "
+            f"column_name, referenced_column_name "
+            f"FROM information_schema.key_column_usage "
+            f"WHERE referenced_table_name NOT LIKE {like_pattern} "
+            f"AND (referenced_table_schema in ({schemas_list}) "
+            f"OR referenced_table_schema is not NULL AND table_schema in ({schemas_list}))"
+        )
+
     def get_constraint_info_sql(self, constraint_name: str, schema_name: str, table_name: str) -> str:
         """Query to get FK constraint details from information_schema."""
         return (

src/datajoint/adapters/postgres.py

@@ -799,6 +799,44 @@ def get_foreign_keys_sql(self, schema_name: str, table_name: str) -> str:
             f"ORDER BY kcu.constraint_name, kcu.ordinal_position"
         )
 
+    def load_primary_keys_sql(self, schemas_list: str, like_pattern: str) -> str:
+        """Query to load all primary key columns across schemas."""
+        tab_expr = "'\"' || kcu.table_schema || '\".\"' || kcu.table_name || '\"'"
+        return (
+            f"SELECT {tab_expr} as tab, kcu.column_name "
+            f"FROM information_schema.key_column_usage kcu "
+            f"JOIN information_schema.table_constraints tc "
+            f"ON kcu.constraint_name = tc.constraint_name "
+            f"AND kcu.table_schema = tc.table_schema "
+            f"WHERE kcu.table_name NOT LIKE {like_pattern} "
+            f"AND kcu.table_schema in ({schemas_list}) "
+            f"AND tc.constraint_type = 'PRIMARY KEY'"
+        )
+
+    def load_foreign_keys_sql(self, schemas_list: str, like_pattern: str) -> str:
+        """Query to load all foreign key relationships across schemas."""
+        return (
+            f"SELECT "
+            f"c.conname as constraint_name, "
+            f"'\"' || ns1.nspname || '\".\"' || cl1.relname || '\"' as referencing_table, "
+            f"'\"' || ns2.nspname || '\".\"' || cl2.relname || '\"' as referenced_table, "
+            f"a1.attname as column_name, "
+            f"a2.attname as referenced_column_name "
+            f"FROM pg_constraint c "
+            f"JOIN pg_class cl1 ON c.conrelid = cl1.oid "
+            f"JOIN pg_namespace ns1 ON cl1.relnamespace = ns1.oid "
+            f"JOIN pg_class cl2 ON c.confrelid = cl2.oid "
+            f"JOIN pg_namespace ns2 ON cl2.relnamespace = ns2.oid "
+            f"CROSS JOIN LATERAL unnest(c.conkey, c.confkey) WITH ORDINALITY AS cols(conkey, confkey, ord) "
+            f"JOIN pg_attribute a1 ON a1.attrelid = cl1.oid AND a1.attnum = cols.conkey "
+            f"JOIN pg_attribute a2 ON a2.attrelid = cl2.oid AND a2.attnum = cols.confkey "
+            f"WHERE c.contype = 'f' "
+            f"AND cl1.relname NOT LIKE {like_pattern} "
+            f"AND (ns2.nspname in ({schemas_list}) "
+            f"OR ns1.nspname in ({schemas_list})) "
+            f"ORDER BY c.conname, cols.ord"
+        )
+
     def get_constraint_info_sql(self, constraint_name: str, schema_name: str, table_name: str) -> str:
         """
         Query to get FK constraint details from information_schema.

src/datajoint/dependencies.py

@@ -164,92 +164,21 @@ def load(self, force: bool = True) -> None:
         # Build schema list for IN clause
         schemas_list = ", ".join(adapter.quote_string(s) for s in self._conn.schemas)
 
-        # Backend-specific queries for primary keys and foreign keys
-        # Note: Both PyMySQL and psycopg2 use %s placeholders, so escape % as %%
+        # Load primary keys and foreign keys via adapter methods
+        # Note: Both PyMySQL and psycopg use %s placeholders, so escape % as %%
         like_pattern = "'~%%'"
 
-        if adapter.backend == "mysql":
-            # MySQL: use concat() and MySQL-specific information_schema columns
-            tab_expr = "concat('`', table_schema, '`.`', table_name, '`')"
-
-            # load primary key info (MySQL uses constraint_name='PRIMARY')
-            keys = self._conn.query(
-                f"""
-                SELECT {tab_expr} as tab, column_name
-                FROM information_schema.key_column_usage
-                WHERE table_name NOT LIKE {like_pattern}
-                    AND table_schema in ({schemas_list})
-                    AND constraint_name='PRIMARY'
-                """
-            )
-            pks = defaultdict(set)
-            for key in keys:
-                pks[key[0]].add(key[1])
-
-            # load foreign keys (MySQL has referenced_* columns)
-            ref_tab_expr = "concat('`', referenced_table_schema, '`.`', referenced_table_name, '`')"
-            fk_keys = self._conn.query(
-                f"""
-                SELECT constraint_name,
-                    {tab_expr} as referencing_table,
-                    {ref_tab_expr} as referenced_table,
-                    column_name, referenced_column_name
-                FROM information_schema.key_column_usage
-                WHERE referenced_table_name NOT LIKE {like_pattern}
-                    AND (referenced_table_schema in ({schemas_list})
-                         OR referenced_table_schema is not NULL AND table_schema in ({schemas_list}))
-                """,
-                as_dict=True,
-            )
-        else:
-            # PostgreSQL: use || concatenation and different query structure
-            tab_expr = "'\"' || kcu.table_schema || '\".\"' || kcu.table_name || '\"'"
-
-            # load primary key info (PostgreSQL uses constraint_type='PRIMARY KEY')
-            keys = self._conn.query(
-                f"""
-                SELECT {tab_expr} as tab, kcu.column_name
-                FROM information_schema.key_column_usage kcu
-                JOIN information_schema.table_constraints tc
-                    ON kcu.constraint_name = tc.constraint_name
-                    AND kcu.table_schema = tc.table_schema
-                WHERE kcu.table_name NOT LIKE {like_pattern}
-                    AND kcu.table_schema in ({schemas_list})
-                    AND tc.constraint_type = 'PRIMARY KEY'
-                """
-            )
-            pks = defaultdict(set)
-            for key in keys:
-                pks[key[0]].add(key[1])
-
-            # load foreign keys using pg_constraint system catalogs
-            # The information_schema approach creates a Cartesian product for composite FKs
-            # because constraint_column_usage doesn't have ordinal_position.
-            # Using pg_constraint with unnest(conkey, confkey) WITH ORDINALITY gives correct mapping.
-            fk_keys = self._conn.query(
-                f"""
-                SELECT
-                    c.conname as constraint_name,
-                    '"' || ns1.nspname || '"."' || cl1.relname || '"' as referencing_table,
-                    '"' || ns2.nspname || '"."' || cl2.relname || '"' as referenced_table,
-                    a1.attname as column_name,
-                    a2.attname as referenced_column_name
-                FROM pg_constraint c
-                JOIN pg_class cl1 ON c.conrelid = cl1.oid
-                JOIN pg_namespace ns1 ON cl1.relnamespace = ns1.oid
-                JOIN pg_class cl2 ON c.confrelid = cl2.oid
-                JOIN pg_namespace ns2 ON cl2.relnamespace = ns2.oid
-                CROSS JOIN LATERAL unnest(c.conkey, c.confkey) WITH ORDINALITY AS cols(conkey, confkey, ord)
-                JOIN pg_attribute a1 ON a1.attrelid = cl1.oid AND a1.attnum = cols.conkey
-                JOIN pg_attribute a2 ON a2.attrelid = cl2.oid AND a2.attnum = cols.confkey
-                WHERE c.contype = 'f'
-                    AND cl1.relname NOT LIKE {like_pattern}
-                    AND (ns2.nspname in ({schemas_list})
-                         OR ns1.nspname in ({schemas_list}))
-                ORDER BY c.conname, cols.ord
-                """,
-                as_dict=True,
-            )
+        # load primary key info
+        keys = self._conn.query(adapter.load_primary_keys_sql(schemas_list, like_pattern))
+        pks = defaultdict(set)
+        for key in keys:
+            pks[key[0]].add(key[1])
+
+        # load foreign keys
+        fk_keys = self._conn.query(
+            adapter.load_foreign_keys_sql(schemas_list, like_pattern),
+            as_dict=True,
+        )
 
         # add nodes to the graph
         for n, pk in pks.items():