refactor: backend-agnostic abstractions for multi-backend support

Centralize patterns that were duplicated or hardcoded for MySQL:

- Add make_full_table_name() to adapter ABC — consolidates quoted
  name construction from 7 call sites into one overridable method.
  Backends with additional namespace levels can override.

- Add foreign_key_action_clause property to adapter ABC — FK
  referential actions via adapter instead of hardcoded in declare.py.
  Backends without referential action support can override.

- Use adapter.split_full_table_name() in declare() — replaces
  fragile manual quote-char detection.

- Guard transaction methods against empty SQL — supports backends
  without multi-table transaction semantics.

- Add "bytes"/"binary" to blob type detection — supports backends
  that use BINARY instead of longblob.

- Route lineage table check through adapter.get_table_info_sql() —
  replaces hardcoded information_schema query.

Co-Authored-By: Kushal Bakshi <kushal.bakshi@datajoint.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: 3 people

pixi.lock

@@ -238,6 +238,38 @@ def parameter_placeholder(self) -> str:
         """
         ...
 
+    def make_full_table_name(self, database: str, table_name: str) -> str:
+        """
+        Construct a fully-qualified table name for this backend.
+
+        Default implementation produces a two-part name (``schema.table``).
+        Backends that require additional namespace levels can override.
+
+        Parameters
+        ----------
+        database : str
+            Schema/database name.
+        table_name : str
+            Table name (including tier prefix).
+
+        Returns
+        -------
+        str
+            Fully-qualified, quoted table name.
+        """
+        return f"{self.quote_identifier(database)}.{self.quote_identifier(table_name)}"
+
+    @property
+    def foreign_key_action_clause(self) -> str:
+        """
+        Referential action clause appended to FOREIGN KEY declarations.
+
+        Default: ``ON UPDATE CASCADE ON DELETE RESTRICT`` (MySQL/PostgreSQL).
+        Backends that don't support referential actions can override to
+        return ``""``.
+        """
+        return " ON UPDATE CASCADE ON DELETE RESTRICT"
+
     # =========================================================================
     # Type Mapping
     # =========================================================================

src/datajoint/adapters/base.py

@@ -557,7 +557,7 @@ def decode_attribute(attr, data, squeeze: bool = False, connection=None):
             # psycopg2 auto-deserializes JSON to dict/list; only parse strings
             if isinstance(data, str):
                 data = json.loads(data)
-        elif final_dtype.lower() in ("longblob", "blob", "mediumblob", "tinyblob"):
+        elif final_dtype.lower() in ("longblob", "blob", "mediumblob", "tinyblob", "bytes", "binary"):
             pass  # Blob data is already bytes
         elif final_dtype.lower() == "binary(16)":
             data = uuid_module.UUID(bytes=data)

src/datajoint/codecs.py

@@ -486,19 +486,25 @@ def start_transaction(self) -> None:
         """
         if self.in_transaction:
             raise errors.DataJointError("Nested connections are not supported.")
-        self.query(self.adapter.start_transaction_sql())
+        sql = self.adapter.start_transaction_sql()
+        if sql:
+            self.query(sql)
         self._in_transaction = True
         logger.debug("Transaction started")
 
     def cancel_transaction(self) -> None:
         """Cancel the current transaction and roll back all changes."""
-        self.query(self.adapter.rollback_sql())
+        sql = self.adapter.rollback_sql()
+        if sql:
+            self.query(sql)
         self._in_transaction = False
         logger.debug("Transaction cancelled. Rolling back ...")
 
     def commit_transaction(self) -> None:
         """Commit all changes and close the transaction."""
-        self.query(self.adapter.commit_sql())
+        sql = self.adapter.commit_sql()
+        if sql:
+            self.query(sql)
         self._in_transaction = False
         logger.debug("Transaction committed and closed.")
 

src/datajoint/connection.py

@@ -296,10 +296,10 @@ def compile_foreign_key(
     parent_full_name = ref.support[0]
     # Parse as database.table using the adapter's quoting convention
     parts = adapter.split_full_table_name(parent_full_name)
-    ref_table_name = f"{adapter.quote_identifier(parts[0])}.{adapter.quote_identifier(parts[1])}"
+    ref_table_name = adapter.make_full_table_name(parts[0], parts[1])
 
     foreign_key_sql.append(
-        f"FOREIGN KEY ({fk_cols}) REFERENCES {ref_table_name} ({pk_cols}) ON UPDATE CASCADE ON DELETE RESTRICT"
+        f"FOREIGN KEY ({fk_cols}) REFERENCES {ref_table_name} ({pk_cols}){adapter.foreign_key_action_clause}"
     )
 
     # declare unique index
@@ -432,16 +432,8 @@ def declare(
     DataJointError
         If table name exceeds max length or has no primary key.
     """
-    # Parse table name without assuming quote character
-    # Extract schema.table from quoted name using adapter
-    quote_char = adapter.quote_identifier("x")[0]  # Get quote char from adapter
-    parts = full_table_name.split(".")
-    if len(parts) == 2:
-        schema_name = parts[0].strip(quote_char)
-        table_name = parts[1].strip(quote_char)
-    else:
-        schema_name = None
-        table_name = parts[0].strip(quote_char)
+    # Parse table name using adapter (handles backend-specific quoting)
+    schema_name, table_name = adapter.split_full_table_name(full_table_name)
 
     if len(table_name) > MAX_TABLE_NAME_LENGTH:
         raise DataJointError(
@@ -924,7 +916,7 @@ def compile_attribute(
     # Check for invalid default values on blob types (after type substitution)
     # Note: blob → longblob, so check for NATIVE_BLOB or longblob result
     final_type = match["type"].lower()
-    if ("blob" in final_type) and match["default"] not in {"DEFAULT NULL", "NOT NULL"}:
+    if ("blob" in final_type or final_type == "binary") and match["default"] not in {"DEFAULT NULL", "NOT NULL"}:
         raise DataJointError("The default value for blob attributes can only be NULL in:\n{line}".format(line=line))
 
     # Use adapter to format column definition

src/datajoint/declare.py

@@ -79,14 +79,12 @@ def lineage_table_exists(connection, database):
     bool
         True if the table exists, False otherwise.
     """
-    result = connection.query(
-        """
-        SELECT COUNT(*) FROM information_schema.tables
-        WHERE table_schema = %s AND table_name = '~lineage'
-        """,
-        args=(database,),
-    ).fetchone()
-    return result[0] > 0
+    try:
+        result = connection.query(connection.adapter.get_table_info_sql(database, "~lineage")).fetchone()
+        return result is not None
+    except Exception:
+        # Schema or catalog query may fail on some backends
+        return False
 
 
 def get_lineage(connection, database, table_name, attribute_name):

src/datajoint/lineage.py

@@ -345,12 +345,7 @@ def make_classes(self, into: dict[str, Any] | None = None) -> None:
         tables = [
             row[0]
             for row in self.connection.query(adapter.list_tables_sql(self.database))
-            if lookup_class_name(
-                f"{adapter.quote_identifier(self.database)}.{adapter.quote_identifier(row[0])}",
-                into,
-                0,
-            )
-            is None
+            if lookup_class_name(adapter.make_full_table_name(self.database, row[0]), into, 0) is None
         ]
         master_classes = (Lookup, Manual, Imported, Computed)
         part_tables = []
@@ -508,7 +503,7 @@ def jobs(self) -> list[Job]:
         # Iterate over auto-populated tables and check if their job table exists
         for table_name in self.list_tables():
             adapter = self.connection.adapter
-            full_name = f"{adapter.quote_identifier(self.database)}." f"{adapter.quote_identifier(table_name)}"
+            full_name = adapter.make_full_table_name(self.database, table_name)
             table = FreeTable(self.connection, full_name)
             tier = _get_tier(table.full_table_name)
             if tier in (Computed, Imported):
@@ -608,8 +603,7 @@ def get_table(self, name: str) -> FreeTable:
         if table_name is None:
             raise DataJointError(f"Table `{name}` does not exist in schema `{self.database}`.")
 
-        adapter = self.connection.adapter
-        full_name = f"{adapter.quote_identifier(self.database)}.{adapter.quote_identifier(table_name)}"
+        full_name = self.connection.adapter.make_full_table_name(self.database, table_name)
         return FreeTable(self.connection, full_name)
 
     def __getitem__(self, name: str) -> FreeTable:

src/datajoint/schemas.py

@@ -474,7 +474,7 @@ def full_table_name(self):
                 f"Class {self.__class__.__name__} is not associated with a schema. "
                 "Apply a schema decorator or use schema() to bind it."
             )
-        return f"{self.adapter.quote_identifier(self.database)}.{self.adapter.quote_identifier(self.table_name)}"
+        return self.adapter.make_full_table_name(self.database, self.table_name)
 
     @property
     def adapter(self):

src/datajoint/table.py

@@ -106,8 +106,7 @@ def full_table_name(cls):
         """The fully qualified table name (quoted per backend)."""
         if cls.database is None:
             return None
-        adapter = cls._connection.adapter
-        return f"{adapter.quote_identifier(cls.database)}.{adapter.quote_identifier(cls.table_name)}"
+        return cls._connection.adapter.make_full_table_name(cls.database, cls.table_name)
 
 
 class UserTable(Table, metaclass=TableMeta):
@@ -186,8 +185,7 @@ def full_table_name(cls):
         """The fully qualified table name (quoted per backend)."""
         if cls.database is None or cls.table_name is None:
             return None
-        adapter = cls._connection.adapter
-        return f"{adapter.quote_identifier(cls.database)}.{adapter.quote_identifier(cls.table_name)}"
+        return cls._connection.adapter.make_full_table_name(cls.database, cls.table_name)
 
     @property
     def master(cls):