fix: Backend-agnostic JSON path expressions

- Updated translate_attribute() to accept optional adapter parameter
- PostgreSQL adapter's json_path_expr() now handles array notation and type casting
- Pass adapter to translate_attribute() in condition.py, declare.py, expression.py

This enables basic JSON operations to work on both MySQL and PostgreSQL.

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

Author: dimitri-yatsenko

src/datajoint/adapters/postgres.py

@@ -1004,26 +1004,47 @@ def json_path_expr(self, column: str, path: str, return_type: str | None = None)
         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).
+            Return type specification for casting (e.g., 'float', 'decimal(10,2)').
 
         Returns
         -------
         str
-            PostgreSQL jsonb_extract_path_text() expression.
+            PostgreSQL jsonb_extract_path_text() expression, with optional cast.
 
         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\\')'
+        >>> adapter.json_path_expr('data', 'value', 'float')
+        'jsonb_extract_path_text("data", \\'value\\')::float'
         """
         quoted_col = self.quote_identifier(column)
-        # Split path by '.' for nested access
-        path_parts = path.split(".")
+        # Split path by '.' for nested access, handling array notation
+        path_parts = []
+        for part in path.split("."):
+            # Handle array access like field[0]
+            if "[" in part:
+                base, rest = part.split("[", 1)
+                path_parts.append(base)
+                # Extract array indices
+                indices = rest.rstrip("]").split("][")
+                path_parts.extend(indices)
+            else:
+                path_parts.append(part)
         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})"
+        expr = f"jsonb_extract_path_text({quoted_col}, {path_args})"
+        # Add cast if return type specified
+        if return_type:
+            # Map DataJoint types to PostgreSQL types
+            pg_type = return_type.lower()
+            if pg_type in ("unsigned", "signed"):
+                pg_type = "integer"
+            elif pg_type == "double":
+                pg_type = "double precision"
+            expr = f"({expr})::{pg_type}"
+        return expr
 
     def translate_expression(self, expr: str) -> str:
         """

src/datajoint/condition.py

@@ -31,14 +31,17 @@
 JSON_PATTERN = re.compile(r"^(?P<attr>\w+)(\.(?P<path>[\w.*\[\]]+))?(:(?P<type>[\w(,\s)]+))?$")
 
 
-def translate_attribute(key: str) -> tuple[dict | None, str]:
+def translate_attribute(key: str, adapter=None) -> tuple[dict | None, str]:
     """
     Translate an attribute key, handling JSON path notation.
 
     Parameters
     ----------
     key : str
         Attribute name, optionally with JSON path (e.g., ``"attr.path.field"``).
+    adapter : DatabaseAdapter, optional
+        Database adapter for backend-specific SQL generation.
+        If not provided, uses MySQL syntax for backward compatibility.
 
     Returns
     -------
@@ -53,9 +56,14 @@ def translate_attribute(key: str) -> tuple[dict | None, str]:
     if match["path"] is None:
         return match, match["attr"]
     else:
-        return match, "json_value(`{}`, _utf8mb4'$.{}'{})".format(
-            *[((f" returning {v}" if k == "type" else v) if v else "") for k, v in match.items()]
-        )
+        # Use adapter's json_path_expr if available, otherwise fall back to MySQL syntax
+        if adapter is not None:
+            return match, adapter.json_path_expr(match["attr"], match["path"], match["type"])
+        else:
+            # Legacy MySQL syntax for backward compatibility
+            return match, "json_value(`{}`, _utf8mb4'$.{}'{})".format(
+                *[((f" returning {v}" if k == "type" else v) if v else "") for k, v in match.items()]
+            )
 
 
 class PromiscuousOperand:
@@ -306,14 +314,17 @@ def make_condition(
 
     def prep_value(k, v):
         """prepare SQL condition"""
-        key_match, k = translate_attribute(k)
-        if key_match["path"] is None:
+        key_match, k = translate_attribute(k, adapter)
+        is_json_path = key_match is not None and key_match.get("path") is not None
+        has_explicit_type = key_match is not None and key_match.get("type") is not None
+
+        if not is_json_path:
             k = adapter.quote_identifier(k)
-        if query_expression.heading[key_match["attr"]].json and key_match["path"] is not None and isinstance(v, dict):
+        if is_json_path and isinstance(v, dict):
             return f"{k}='{json.dumps(v)}'"
         if v is None:
             return f"{k} IS NULL"
-        if query_expression.heading[key_match["attr"]].uuid:
+        if key_match is not None and query_expression.heading[key_match["attr"]].uuid:
             if not isinstance(v, uuid.UUID):
                 try:
                     v = uuid.UUID(v)

src/datajoint/declare.py

@@ -755,7 +755,7 @@ def compile_index(line: str, index_sql: list[str], adapter) -> None:
     """
 
     def format_attribute(attr):
-        match, attr = translate_attribute(attr)
+        match, attr = translate_attribute(attr, adapter)
         if match is None:
             return attr
         if match["path"] is None:

src/datajoint/expression.py

@@ -457,7 +457,8 @@ def proj(self, *attributes, **named_attributes):
         from other attributes available before the projection.
         Each attribute name can only be used once.
         """
-        named_attributes = {k: translate_attribute(v)[1] for k, v in named_attributes.items()}
+        adapter = self.connection.adapter if hasattr(self, 'connection') and self.connection else None
+        named_attributes = {k: translate_attribute(v, adapter)[1] for k, v in named_attributes.items()}
         # new attributes in parentheses are included again with the new name without removing original
         duplication_pattern = re.compile(rf"^\s*\(\s*(?!{'|'.join(CONSTANT_LITERALS)})(?P<name>[a-zA-Z_]\w*)\s*\)\s*$")
         # attributes without parentheses renamed