feat: Add json_path_expr() method to database adapters (Phase 6 Part 1)

Add json_path_expr() method to support backend-agnostic JSON path extraction:
- Add abstract method to DatabaseAdapter base class
- Implement for MySQL: json_value(`col`, _utf8mb4'$.path' returning type)
- Implement for PostgreSQL: jsonb_extract_path_text("col", 'path_part1', 'path_part2')
- Add comprehensive unit tests for both backends

This is Part 1 of Phase 6. Parts 2-3 will update condition.py and expression.py
to use adapter methods for WHERE clauses and query expression SQL.

All tests pass. Fully backward compatible.

Part of multi-backend PostgreSQL support implementation.
Related: #1338

Author: dimitri-yatsenko

src/datajoint/adapters/base.py

@@ -683,6 +683,32 @@ def interval_expr(self, value: int, unit: str) -> str:
         """
         ...
 
+    @abstractmethod
+    def json_path_expr(self, column: str, path: str, return_type: str | None = None) -> str:
+        """
+        Generate JSON path extraction expression.
+
+        Parameters
+        ----------
+        column : str
+            Column name containing JSON data.
+        path : str
+            JSON path (e.g., 'field' or 'nested.field').
+        return_type : str, optional
+            Return type specification (MySQL-specific).
+
+        Returns
+        -------
+        str
+            Database-specific JSON extraction SQL expression.
+
+        Examples
+        --------
+        MySQL: json_value(`column`, _utf8mb4'$.path' returning type)
+        PostgreSQL: jsonb_extract_path_text("column", 'path_part1', 'path_part2')
+        """
+        ...
+
     # =========================================================================
     # Error Translation
     # =========================================================================

src/datajoint/adapters/mysql.py

@@ -666,6 +666,35 @@ def interval_expr(self, value: int, unit: str) -> str:
         # MySQL uses singular unit names
         return f"INTERVAL {value} {unit.upper()}"
 
+    def json_path_expr(self, column: str, path: str, return_type: str | None = None) -> str:
+        """
+        Generate MySQL json_value() expression.
+
+        Parameters
+        ----------
+        column : str
+            Column name containing JSON data.
+        path : str
+            JSON path (e.g., 'field' or 'nested.field').
+        return_type : str, optional
+            Return type specification (e.g., 'decimal(10,2)').
+
+        Returns
+        -------
+        str
+            MySQL json_value() expression.
+
+        Examples
+        --------
+        >>> adapter.json_path_expr('data', 'field')
+        "json_value(`data`, _utf8mb4'$.field')"
+        >>> adapter.json_path_expr('data', 'value', 'decimal(10,2)')
+        "json_value(`data`, _utf8mb4'$.value' returning decimal(10,2))"
+        """
+        quoted_col = self.quote_identifier(column)
+        return_clause = f" returning {return_type}" if return_type else ""
+        return f"json_value({quoted_col}, _utf8mb4'$.{path}'{return_clause})"
+
     # =========================================================================
     # Error Translation
     # =========================================================================

src/datajoint/adapters/postgres.py

@@ -727,6 +727,38 @@ def interval_expr(self, value: int, unit: str) -> str:
         unit_plural = unit.lower() + "s" if not unit.endswith("s") else unit.lower()
         return f"INTERVAL '{value} {unit_plural}'"
 
+    def json_path_expr(self, column: str, path: str, return_type: str | None = None) -> str:
+        """
+        Generate PostgreSQL jsonb_extract_path_text() expression.
+
+        Parameters
+        ----------
+        column : str
+            Column name containing JSON data.
+        path : str
+            JSON path (e.g., 'field' or 'nested.field').
+        return_type : str, optional
+            Return type specification (not used in PostgreSQL jsonb_extract_path_text).
+
+        Returns
+        -------
+        str
+            PostgreSQL jsonb_extract_path_text() expression.
+
+        Examples
+        --------
+        >>> adapter.json_path_expr('data', 'field')
+        'jsonb_extract_path_text("data", \\'field\\')'
+        >>> adapter.json_path_expr('data', 'nested.field')
+        'jsonb_extract_path_text("data", \\'nested\\', \\'field\\')'
+        """
+        quoted_col = self.quote_identifier(column)
+        # Split path by '.' for nested access
+        path_parts = path.split(".")
+        path_args = ", ".join(f"'{part}'" for part in path_parts)
+        # Note: PostgreSQL jsonb_extract_path_text doesn't use return type parameter
+        return f"jsonb_extract_path_text({quoted_col}, {path_args})"
+
     # =========================================================================
     # Error Translation
     # =========================================================================

tests/unit/test_adapters.py

@@ -171,6 +171,16 @@ def test_interval_expr(self, adapter):
         assert adapter.interval_expr(5, "second") == "INTERVAL 5 SECOND"
         assert adapter.interval_expr(10, "minute") == "INTERVAL 10 MINUTE"
 
+    def test_json_path_expr(self, adapter):
+        """Test JSON path extraction."""
+        assert adapter.json_path_expr("data", "field") == "json_value(`data`, _utf8mb4'$.field')"
+        assert adapter.json_path_expr("record", "nested") == "json_value(`record`, _utf8mb4'$.nested')"
+
+    def test_json_path_expr_with_return_type(self, adapter):
+        """Test JSON path extraction with return type."""
+        result = adapter.json_path_expr("data", "value", "decimal(10,2)")
+        assert result == "json_value(`data`, _utf8mb4'$.value' returning decimal(10,2))"
+
     def test_transaction_sql(self, adapter):
         """Test transaction statements."""
         assert "START TRANSACTION" in adapter.start_transaction_sql()
@@ -306,6 +316,16 @@ def test_interval_expr(self, adapter):
         assert adapter.interval_expr(5, "second") == "INTERVAL '5 seconds'"
         assert adapter.interval_expr(10, "minute") == "INTERVAL '10 minutes'"
 
+    def test_json_path_expr(self, adapter):
+        """Test JSON path extraction for PostgreSQL."""
+        assert adapter.json_path_expr("data", "field") == "jsonb_extract_path_text(\"data\", 'field')"
+        assert adapter.json_path_expr("record", "name") == "jsonb_extract_path_text(\"record\", 'name')"
+
+    def test_json_path_expr_nested(self, adapter):
+        """Test JSON path extraction with nested paths."""
+        result = adapter.json_path_expr("data", "nested.field")
+        assert result == "jsonb_extract_path_text(\"data\", 'nested', 'field')"
+
     def test_transaction_sql(self, adapter):
         """Test transaction statements."""
         assert adapter.start_transaction_sql() == "BEGIN"