From 373a9640fc2927647d1aabc4774af2574593fbed Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tim=20Swe=C3=B1a?= <swast@google.com>
Date: Wed, 11 Nov 2026 22:37:53 +0000
Subject: [PATCH 1/4] fix: escape JSON subfields

---
 .../sqlalchemy_bigquery/_struct.py            |  11 +-
 .../sqlalchemy_bigquery/base.py               | 209 ++++++++++++++-
 .../tests/system/test__struct.py              |   6 +-
 .../tests/system/test_sqlalchemy_bigquery.py  |  64 ++++-
 .../tests/unit/test__struct.py                |  74 +++++-
 .../tests/unit/test_compiler.py               | 247 +++++++++++++++++-
 6 files changed, 580 insertions(+), 31 deletions(-)

diff --git a/packages/sqlalchemy-bigquery/sqlalchemy_bigquery/_struct.py b/packages/sqlalchemy-bigquery/sqlalchemy_bigquery/_struct.py
index 5b9678d63502..35562c7cd26a 100644
--- a/packages/sqlalchemy-bigquery/sqlalchemy_bigquery/_struct.py
+++ b/packages/sqlalchemy-bigquery/sqlalchemy_bigquery/_struct.py
@@ -91,6 +91,11 @@ def _setup_getitem(self, name):
             subtype = self.expr.type._STRUCT_byname.get(name.lower())
             if subtype is None:
                 raise KeyError(name)
+            # Reuse `operators.json_getitem_op` to represent STRUCT subfield
+            # access. `BigQueryCompiler.visit_json_getitem_op_binary` in
+            # `base.py` checks whether the left operand is a `STRUCT` (vs. a
+            # `sqlalchemy.JSON` column) and compiles STRUCT subfields as
+            # backtick-quoted SQL identifiers (`struct_col.`field``).
             operator = operators.json_getitem_op
             index = _field_index(self, name, operator)
             return operator, index, subtype
@@ -112,9 +117,3 @@ def _field_index(self, name, operator):
         operator=operator,
         bindparam_type=sqlalchemy.types.String(),
     )
-
-
-# class SQLCompiler:
-#     def visit_json_getitem_op_binary(self, binary, operator_, **kw):
-#         left = self.process(binary.left, **kw)
-#         return f"{left}.{binary.right.value}"
diff --git a/packages/sqlalchemy-bigquery/sqlalchemy_bigquery/base.py b/packages/sqlalchemy-bigquery/sqlalchemy_bigquery/base.py
index 40b56bfbb715..09b1ebe1a6e3 100644
--- a/packages/sqlalchemy-bigquery/sqlalchemy_bigquery/base.py
+++ b/packages/sqlalchemy-bigquery/sqlalchemy_bigquery/base.py
@@ -59,7 +59,7 @@
 from sqlalchemy.sql.selectable import CTE
 from sqlalchemy.sql.sqltypes import Integer, NullType, Numeric, String
 
-from . import _helpers, _types
+from . import _helpers, _struct, _types
 from .parse_url import parse_url
 
 # Illegal characters is intended to be all characters that are not explicitly
@@ -85,8 +85,22 @@ def __init__(self, dialect):
         super(BigQueryIdentifierPreparer, self).__init__(
             dialect,
             initial_quote="`",
+            escape_quote="`",
         )
 
+    def _escape_identifier(self, value):
+        # SQLAlchemy's base IdentifierPreparer defaults to escaping quotes by
+        # doubling them (e.g. `""`), whereas BigQuery GoogleSQL escapes
+        # backslashes and backticks inside backtick-quoted identifiers using a
+        # leading backslash (`\\` and ``\` ``).
+        value = value.replace("\\", "\\\\").replace("`", "\\`")
+        if self._double_percents:
+            value = value.replace("%", "%%")
+        return value
+
+    def _unescape_identifier(self, value):
+        return value.replace("\\`", "`").replace("\\\\", "\\")
+
     def quote_column(self, value):
         """
         Quote a column.
@@ -216,9 +230,106 @@ def visit_insert(self, insert_stmt, asfrom=False, **kw):
             insert_stmt, asfrom=False, **kw
         )
 
+    @staticmethod
+    def _get_struct_type(type_):
+        # Unwrap TypeDecorator.impl so custom types wrapping STRUCT are still
+        # recognized as STRUCT columns during compilation.
+        if isinstance(type_, _struct.STRUCT):
+            return type_
+        impl = getattr(type_, "impl", None)
+        if isinstance(impl, _struct.STRUCT):
+            return impl
+        return None
+
     def visit_json_getitem_op_binary(self, binary, operator_, **kw):
+        # Both `STRUCT` subfield access (see `_struct.py`) and `sqlalchemy.JSON`
+        # subscripting (`json_col[key]`) use `operators.json_getitem_op`, but
+        # they compile to completely different BigQuery constructs:
+        #
+        # 1. For `STRUCT`: subfields are SQL identifiers (`struct_col.`field``),
+        #    which cannot be query parameters in BigQuery. We verify that the
+        #    requested field exists in the static STRUCT schema (in case
+        #    `json_getitem_op` was constructed directly, bypassing
+        #    `STRUCT.Comparator._setup_getitem`) and backtick-quote the
+        #    identifier via `quote_identifier`.
+        struct_type = self._get_struct_type(binary.left.type)
+        if struct_type is not None:
+            field_name = getattr(binary.right, "value", None)
+            if (
+                not isinstance(field_name, str)
+                or field_name.lower() not in struct_type._STRUCT_byname
+            ):
+                raise sqlalchemy.exc.CompileError(
+                    f"Invalid STRUCT field name: {field_name!r}"
+                )
+            left = self.process(binary.left, **kw)
+            return f"{left}.{self.preparer.quote_identifier(field_name)}"
+
+        # 2. For `JSON`: if a sequence (tuple/list) was passed to
+        #    `json_getitem_op`, delegate to `visit_json_path_getitem_op_binary`
+        #    so it compiles as a multi-step `JSON_QUERY` path lookup.
+        if isinstance(
+            binary.right, sqlalchemy.sql.elements.BindParameter
+        ) and _is_json_path_sequence(binary.right.value):
+            return self.visit_json_path_getitem_op_binary(binary, operator_, **kw)
+
+        # Otherwise, compile single-step `JSON` subscripting using BigQuery's
+        # native JSON subscript operator `{left}[{right}]` rather than
+        # `JSON_VALUE` or unquoted dot access:
+        # - Unlike raw dot interpolation (`{left}.{right}`), `{left}[{right}]`
+        #   accepts query parameters (`%(param:STRING)s` / `%(param:INT64)s`),
+        #   preventing SQL injection and supporting integer array indices.
+        # - Unlike `JSON_VALUE` (which extracts a scalar `STRING` and returns
+        #   `NULL` for nested objects/arrays), `{left}[{right}]` returns `JSON`,
+        #   preserving the return type and chained subscripts (`col["a"][0]["b"]`).
         left = self.process(binary.left, **kw)
-        return f"{left}.{binary.right.value}"
+        right = binary.right
+        if isinstance(right, sqlalchemy.sql.elements.BindParameter):
+            if isinstance(right.value, bool):
+                # In Python `isinstance(True, int)` is True, so SQLAlchemy's
+                # `JSON.Comparator` assigns `JSONIntIndexType` to `col[True]`.
+                # BigQuery rejects `BOOL` as a JSON subscript, and JSON object
+                # keys are strings, so coerce boolean subscripts to string keys.
+                right = right._clone(maintain_key=True)
+                right.value = str(right.value)
+                right.type = BQJSONStrIndexType()
+            elif isinstance(right.type, type):
+                # When subscripting with an untyped `sa.bindparam("k")`,
+                # SQLAlchemy's `coercions.expect` sets `bindparam.type` to the
+                # `JSONStrIndexType` class rather than an instance. Instantiate
+                # it on a clone so `type_compiler` can dispatch it.
+                right = right._clone(maintain_key=True)
+                right.type = sqlalchemy.sql.type_api.to_instance(right.type)
+            elif isinstance(right.type, NullType):
+                # When `json_getitem_op` is constructed directly with an untyped
+                # BindParameter, infer the index type so `visit_bindparam` can
+                # emit the required `:INT64` or `:STRING` BigQuery type suffix.
+                right = right._clone(maintain_key=True)
+                right.type = (
+                    BQJSONIntIndexType()
+                    if isinstance(right.value, int)
+                    else BQJSONStrIndexType()
+                )
+        right_text = self.process(right, **kw)
+        return f"{left}[{right_text}]"
+
+    def visit_json_path_getitem_op_binary(self, binary, operator_, **kw):
+        # BigQuery's bracket subscript `json_col[...]` only steps one key or
+        # array offset at a time. Multi-step path tuples (`col[("a", 0, "b")]`)
+        # use `JSON_QUERY(json_col, json_path)`, which also returns a `JSON`
+        # value (consistent with single-step subscripting) and accepts a
+        # parameterized JSONPath string formatted by `BQJSONPathType`.
+        left = self.process(binary.left, **kw)
+        right = binary.right
+        if isinstance(right, sqlalchemy.sql.elements.BindParameter) and not isinstance(
+            right.type, sqlalchemy.sql.sqltypes.JSON.JSONPathType
+        ):
+            right = right._clone(maintain_key=True)
+            right.type = (
+                BQJSONPathType() if _is_json_path_sequence(right.value) else BQString()
+            )
+        right_text = self.process(right, **kw)
+        return f"JSON_QUERY({left}, {right_text})"
 
     def visit_table_valued_alias(self, element, **kw):
         # When using table-valued functions, like UNNEST, BigQuery requires a
@@ -533,6 +644,16 @@ def visit_bindparam(
         if literal_binds or isinstance(type_, NullType):
             return param
 
+        # SQLAlchemy's generic `JSONElementType` subclasses (`JSONIndexType`,
+        # `JSONIntIndexType`, `JSONStrIndexType`, `JSONPathType`) do not define
+        # `__visit_name__`. Because `bindparam.type` is not automatically
+        # adapted to the dialect before `visit_bindparam`, we resolve its
+        # dialect implementation via `BigQueryDialect.colspecs` (`BQJSON*Type`)
+        # so `self.dialect.type_compiler.process(type_)` can emit `:STRING` or
+        # `:INT64`.
+        if isinstance(type_, sqlalchemy.sql.sqltypes.JSON.JSONElementType):
+            type_ = type_.dialect_impl(self.dialect)
+
         if (
             isinstance(type_, Numeric)
             and (type_.precision is None or type_.scale is None)
@@ -637,6 +758,15 @@ def visit_BINARY(self, type_, **kw):
     def visit_JSON(self, type_, **kw):
         return "JSON"
 
+    def visit_json_int_index(self, type_, **kw):
+        return "INT64"
+
+    def visit_json_str_index(self, type_, **kw):
+        return "STRING"
+
+    visit_json_index = visit_json_str_index
+    visit_json_path = visit_json_str_index
+
     def visit_NUMERIC(self, type_, **kw):
         if (type_.precision is not None) and isinstance(
             kw.get("type_expression"), Column
@@ -1007,11 +1137,78 @@ def process_string_literal(value):
     return repr(value.replace("%", "%%"))
 
 
+def _escape_json_path_key(key: str) -> str:
+    # Always quote member keys in BigQuery JSONPath (`$."key"`) and escape
+    # backslashes and double quotes so keys containing dots, brackets, quotes,
+    # spaces, or `$` cannot alter the JSONPath structure.
+    escaped = key.replace("\\", "\\\\").replace('"', '\\"')
+    return f'."{escaped}"'
+
+
+def _format_json_path_step(step: Any) -> str:
+    # Check `not isinstance(step, bool)` because `bool` is a subclass of `int`
+    # in Python, and boolean keys in a path tuple should be treated as string
+    # member names rather than `[1]` / `[0]` array offsets.
+    if isinstance(step, int) and not isinstance(step, bool):
+        return f"[{step}]"
+    return _escape_json_path_key(str(step))
+
+
+def _is_json_path_sequence(value: Any) -> bool:
+    return isinstance(value, (tuple, list))
+
+
 class BQString(String):
     def literal_processor(self, dialect):
         return process_string_literal
 
 
+# SQLAlchemy's built-in `JSON.JSONIndexType`, `JSON.JSONIntIndexType`, and
+# `JSON.JSONStrIndexType` do not define `__visit_name__`, which causes
+# `BigQueryTypeCompiler.process()` to fail when `visit_bindparam` formats the
+# `%(name:TYPE)s` parameter placeholder. These dialect subclasses supply
+# `__visit_name__` and `process_string_literal` (to escape `%` as `%%` when
+# `literal_binds=True`) and are registered in `BigQueryDialect.colspecs`.
+class BQJSONIndexType(sqlalchemy.sql.sqltypes.JSON.JSONIndexType):
+    __visit_name__ = "json_index"
+
+    def string_literal_processor(self, dialect):
+        return process_string_literal
+
+
+class BQJSONIntIndexType(sqlalchemy.sql.sqltypes.JSON.JSONIntIndexType):
+    __visit_name__ = "json_int_index"
+
+
+class BQJSONStrIndexType(sqlalchemy.sql.sqltypes.JSON.JSONStrIndexType):
+    __visit_name__ = "json_str_index"
+
+    def string_literal_processor(self, dialect):
+        return process_string_literal
+
+
+class BQJSONPathType(sqlalchemy.sql.sqltypes.JSON.JSONPathType):
+    __visit_name__ = "json_path"
+
+    def _format_value(self, value):
+        # Convert a sequence of path steps like `("a", 0, 'b"c')` into a
+        # BigQuery JSONPath string `'$."a"[0]."b\\"c"'` for `JSON_QUERY`.
+        path = "".join(_format_json_path_step(elem) for elem in value)
+        return f"${path}"
+
+    def bind_processor(self, dialect):
+        def process(value):
+            return self._format_value(value)
+
+        return process
+
+    def literal_processor(self, dialect):
+        def process(value):
+            return process_string_literal(self._format_value(value))
+
+        return process
+
+
 class BQBinary(sqlalchemy.sql.sqltypes._Binary):
     @staticmethod
     def __process_bytes_literal(value):
@@ -1079,6 +1276,10 @@ class BigQueryDialect(DefaultDialect):
     supports_native_decimal = True
     description_encoding = None
     supports_native_boolean = True
+    # google-cloud-bigquery's DB-API cursor already parses JSON fields into
+    # Python objects via CELL_DATA_PARSER.json_to_py, so SQLAlchemy's default
+    # JSON.result_processor must not call json.loads() a second time.
+    supports_native_json_deserialization = True
     supports_simple_order_by_label = True
     postfetch_lastrowid = False
     preexecute_autoincrement_sequences = False
@@ -1092,6 +1293,10 @@ class BigQueryDialect(DefaultDialect):
         sqlalchemy.sql.sqltypes.TIMESTAMP: BQTimestamp,
         sqlalchemy.sql.sqltypes.ARRAY: BQArray,
         sqlalchemy.sql.sqltypes.Enum: sqlalchemy.sql.sqltypes.Enum,
+        sqlalchemy.sql.sqltypes.JSON.JSONIndexType: BQJSONIndexType,
+        sqlalchemy.sql.sqltypes.JSON.JSONIntIndexType: BQJSONIntIndexType,
+        sqlalchemy.sql.sqltypes.JSON.JSONStrIndexType: BQJSONStrIndexType,
+        sqlalchemy.sql.sqltypes.JSON.JSONPathType: BQJSONPathType,
     }
 
     def __init__(
diff --git a/packages/sqlalchemy-bigquery/tests/system/test__struct.py b/packages/sqlalchemy-bigquery/tests/system/test__struct.py
index 1fd9fc80fd12..afcc6f6cc3d2 100644
--- a/packages/sqlalchemy-bigquery/tests/system/test__struct.py
+++ b/packages/sqlalchemy-bigquery/tests/system/test__struct.py
@@ -22,7 +22,6 @@
 import packaging.version
 import pytest
 import sqlalchemy
-
 import sqlalchemy_bigquery
 
 
@@ -117,7 +116,7 @@ def test_complex_literals_pr_67(engine, bigquery_dataset, metadata):
         sqlalchemy.select((table.c.dimensions.x__count + 5).label("c")).compile(engine)
     )
     want = (
-        f"SELECT `{table_name}`.`dimensions`.x__count + %(param_1:INT64)s AS `c` \n"
+        f"SELECT `{table_name}`.`dimensions`.`x__count` + %(param_1:INT64)s AS `c` \n"
         f"FROM `{table_name}`"
     )
 
@@ -136,7 +135,6 @@ def test_unnest_and_struct_access_233(engine, bigquery_dataset, metadata):
 
     from sqlalchemy import ARRAY, Column, String, Table, func, select
     from sqlalchemy.orm import sessionmaker
-
     from sqlalchemy_bigquery import STRUCT
 
     conn = engine.connect()
@@ -185,7 +183,7 @@ def test_unnest_and_struct_access_233(engine, bigquery_dataset, metadata):
         f" AS `another_mock_objects`"
         f") AS `anon_1` "
         f"ON "
-        f"`anon_1`.`another_mock_objects`.object_id = "
+        f"`anon_1`.`another_mock_objects`.`object_id` = "
         f"`{bigquery_dataset}.Mock`.`mock_id`"
     )
     assert got == want
diff --git a/packages/sqlalchemy-bigquery/tests/system/test_sqlalchemy_bigquery.py b/packages/sqlalchemy-bigquery/tests/system/test_sqlalchemy_bigquery.py
index 6947e61707ae..280fe358c2e9 100644
--- a/packages/sqlalchemy-bigquery/tests/system/test_sqlalchemy_bigquery.py
+++ b/packages/sqlalchemy-bigquery/tests/system/test_sqlalchemy_bigquery.py
@@ -24,6 +24,7 @@
 import packaging.version
 import pytest
 import sqlalchemy
+import sqlalchemy_bigquery
 from google.cloud.bigquery import TimePartitioning
 from sqlalchemy import (
     Column,
@@ -43,8 +44,6 @@
 from sqlalchemy.orm import sessionmaker
 from sqlalchemy.sql import expression
 
-import sqlalchemy_bigquery
-
 SQLALCHEMY_VERSION = packaging.version.parse(sqlalchemy.__version__)
 
 ONE_ROW_CONTENTS = [
@@ -830,7 +829,6 @@ def test_huge_in():
 )
 def test_unnest(engine, bigquery_dataset):
     from sqlalchemy import String, func, select
-
     from sqlalchemy_bigquery import ARRAY
 
     conn = engine.connect()
@@ -861,7 +859,6 @@ def test_unnest(engine, bigquery_dataset):
 )
 def test_unnest_with_cte(engine, bigquery_dataset):
     from sqlalchemy import String, func, select
-
     from sqlalchemy_bigquery import ARRAY
 
     conn = engine.connect()
@@ -930,3 +927,62 @@ def test_not_regexp_match(session, table):
 
     number_of_non_barrowst_records = 993
     assert len(results) == number_of_non_barrowst_records
+
+
+def test_json_subscripts(engine, bigquery_dataset, metadata):
+    table_name = f"{bigquery_dataset}.test_json_subscripts"
+    table = Table(
+        table_name,
+        metadata,
+        Column("id", sqlalchemy.Integer),
+        Column("payload", sqlalchemy.JSON),
+    )
+    metadata.create_all(engine)
+
+    sample_json = (
+        '{"name": "alice", "items": [{"id": 10}, {"id": 20}], '
+        '"a`b": "tricky", "True": "bool_val"}'
+    )
+    with engine.connect() as conn:
+        conn.execute(
+            table.insert().from_select(
+                ["id", "payload"],
+                select(
+                    sqlalchemy.literal(1),
+                    func.parse_json(sample_json),
+                ),
+            )
+        )
+
+        stmt = select(
+            table.c.payload,
+            table.c.payload["name"],
+            table.c.payload["items"][1],
+            table.c.payload["items"][1]["id"],
+            table.c.payload[("items", 0, "id")],
+            table.c.payload["a`b"],
+            table.c.payload[True],
+            table.c.payload["a` FROM `events`; DROP TABLE `x`; --"],
+        ).where(func.json_value(table.c.payload["name"]) == "alice")
+
+        row = conn.execute(stmt).fetchone()
+        assert row == (
+            {
+                "name": "alice",
+                "items": [{"id": 10}, {"id": 20}],
+                "a`b": "tricky",
+                "True": "bool_val",
+            },
+            "alice",
+            {"id": 20},
+            20,
+            10,
+            "tricky",
+            "bool_val",
+            None,
+        )
+
+        # Also verify literal_binds=True execution against BigQuery
+        compiled_literal = stmt.compile(engine, compile_kwargs={"literal_binds": True})
+        literal_row = conn.exec_driver_sql(str(compiled_literal)).fetchone()
+        assert literal_row == row
diff --git a/packages/sqlalchemy-bigquery/tests/unit/test__struct.py b/packages/sqlalchemy-bigquery/tests/unit/test__struct.py
index 7e3a65d5d43c..13389136033f 100644
--- a/packages/sqlalchemy-bigquery/tests/unit/test__struct.py
+++ b/packages/sqlalchemy-bigquery/tests/unit/test__struct.py
@@ -61,24 +61,24 @@ def _col():
 @pytest.mark.parametrize(
     "expr,sql",
     [
-        (_col()["name"], "`t`.`person`.name"),
-        (_col()["Name"], "`t`.`person`.Name"),
-        (_col().NAME, "`t`.`person`.NAME"),
-        (_col().children, "`t`.`person`.children"),
+        (_col()["name"], "`t`.`person`.`name`"),
+        (_col()["Name"], "`t`.`person`.`Name`"),
+        (_col().NAME, "`t`.`person`.`NAME`"),
+        (_col().children, "`t`.`person`.`children`"),
         (
             # SQLAlchemy doesn't add the label in this case for some reason.
             # TODO: why?
             # https://github.com/googleapis/python-bigquery-sqlalchemy/issues/336
             _col().children[0].label("anon_1"),
-            "(`t`.`person`.children)[OFFSET(%(param_1:INT64)s)]",
+            "(`t`.`person`.`children`)[OFFSET(%(param_1:INT64)s)]",
         ),
         (
             _col().children[0]["bdate"],
-            "((`t`.`person`.children)[OFFSET(%(param_1:INT64)s)]).bdate",
+            "((`t`.`person`.`children`)[OFFSET(%(param_1:INT64)s)]).`bdate`",
         ),
         (
             _col().children[0].bdate,
-            "((`t`.`person`.children)[OFFSET(%(param_1:INT64)s)]).bdate",
+            "((`t`.`person`.`children`)[OFFSET(%(param_1:INT64)s)]).`bdate`",
         ),
     ],
 )
@@ -92,34 +92,34 @@ def test_struct_traversal_project(faux_conn, expr, sql):
     [
         (
             _col()["name"] == "bob",
-            "`t`.`person`.name = %(param_1:STRING)s",
+            "`t`.`person`.`name` = %(param_1:STRING)s",
         ),
         (
             _col()["Name"] == "bob",
-            "`t`.`person`.Name = %(param_1:STRING)s",
+            "`t`.`person`.`Name` = %(param_1:STRING)s",
         ),
         (
             _col()["NAME"] == "bob",
-            "`t`.`person`.NAME = %(param_1:STRING)s",
+            "`t`.`person`.`NAME` = %(param_1:STRING)s",
         ),
         (
             _col().children[0] == dict(name="foo", bdate=datetime.date(2020, 1, 1)),
-            "(`t`.`person`.children)[OFFSET(%(param_1:INT64)s)]"
+            "(`t`.`person`.`children`)[OFFSET(%(param_1:INT64)s)]"
             " = %(param_2:STRUCT<name STRING, bdate DATE>)s",
         ),
         (
             _col().children[0] == dict(name="foo", bdate=datetime.date(2020, 1, 1)),
-            "(`t`.`person`.children)[OFFSET(%(param_1:INT64)s)]"
+            "(`t`.`person`.`children`)[OFFSET(%(param_1:INT64)s)]"
             " = %(param_2:STRUCT<name STRING, bdate DATE>)s",
         ),
         (
             _col().children[0]["bdate"] == datetime.date(2021, 8, 30),
-            "((`t`.`person`.children)[OFFSET(%(param_1:INT64)s)]).bdate"
+            "((`t`.`person`.`children`)[OFFSET(%(param_1:INT64)s)]).`bdate`"
             " = %(param_2:DATE)s",
         ),
         (
             _col().children[0].bdate == datetime.date(2021, 8, 30),
-            "((`t`.`person`.children)[OFFSET(%(param_1:INT64)s)]).bdate"
+            "((`t`.`person`.`children`)[OFFSET(%(param_1:INT64)s)]).`bdate`"
             " = %(param_2:DATE)s",
         ),
     ],
@@ -161,3 +161,49 @@ def test_struct_non_string_field_access(faux_conn):
 def test_struct_bad_name(faux_conn):
     with pytest.raises(KeyError, match="42"):
         _col()["42"]
+
+
+def test_struct_compiler_rejects_invalid_direct_json_getitem_op(faux_conn):
+    from sqlalchemy.sql import operators
+    from sqlalchemy.sql.elements import BinaryExpression
+
+    col = _col()
+    bad_field_expr = BinaryExpression(
+        col, sqlalchemy.bindparam("field", "not_a_field"), operators.json_getitem_op
+    )
+    with pytest.raises(
+        sqlalchemy.exc.CompileError,
+        match="Invalid STRUCT field name: 'not_a_field'",
+    ):
+        sqlalchemy.select(bad_field_expr).compile(faux_conn.engine)
+
+    non_str_expr = BinaryExpression(
+        col, sqlalchemy.bindparam("field", 123), operators.json_getitem_op
+    )
+    with pytest.raises(
+        sqlalchemy.exc.CompileError,
+        match="Invalid STRUCT field name: 123",
+    ):
+        sqlalchemy.select(non_str_expr).compile(faux_conn.engine)
+
+
+def test_struct_type_decorator_and_identifier_escaping(faux_conn):
+    from sqlalchemy.sql import operators
+    from sqlalchemy.sql.elements import BinaryExpression
+    from sqlalchemy_bigquery import STRUCT
+
+    class WrappedStruct(sqlalchemy.types.TypeDecorator):
+        impl = STRUCT(("a`b\\c%d", sqlalchemy.String))
+        cache_ok = True
+
+    t = sqlalchemy.Table(
+        "t",
+        sqlalchemy.MetaData(),
+        sqlalchemy.Column("s", WrappedStruct()),
+    )
+
+    expr = BinaryExpression(
+        t.c.s, sqlalchemy.bindparam("field", "a`b\\c%d"), operators.json_getitem_op
+    )
+    got = str(sqlalchemy.select(expr).compile(faux_conn.engine))
+    assert got == "SELECT `t`.`s`.`a\\`b\\\\c%%d` AS `anon_1` \nFROM `t`"
diff --git a/packages/sqlalchemy-bigquery/tests/unit/test_compiler.py b/packages/sqlalchemy-bigquery/tests/unit/test_compiler.py
index 5341f8058d1c..8326237cc288 100644
--- a/packages/sqlalchemy-bigquery/tests/unit/test_compiler.py
+++ b/packages/sqlalchemy-bigquery/tests/unit/test_compiler.py
@@ -20,7 +20,6 @@
 import pytest
 import sqlalchemy.exc
 from sqlalchemy.sql.functions import cube, grouping_sets, rollup
-
 from sqlalchemy_bigquery import SafeCast, safe_cast
 
 from .conftest import setup_table, sqlalchemy_2_0_or_higher, sqlalchemy_before_2_0
@@ -571,3 +570,249 @@ def test_sqlalchemy_try_cast_compiles_to_safe_cast(
         "FROM `table1`"
     )
     assert found_sql == expected_sql
+
+
+@pytest.fixture
+def json_events_table(metadata):
+    return sqlalchemy.Table(
+        "events",
+        metadata,
+        sqlalchemy.Column("id", sqlalchemy.Integer),
+        sqlalchemy.Column("payload", sqlalchemy.JSON),
+        sqlalchemy.Column("k", sqlalchemy.String),
+        sqlalchemy.Column("idx", sqlalchemy.Integer),
+    )
+
+
+def test_json_getitem_string_key(faux_conn, json_events_table):
+    stmt = sqlalchemy.select(json_events_table.c.payload["key"].label("val"))
+    compiled = stmt.compile(faux_conn)
+    assert (
+        compiled.string
+        == "SELECT `events`.`payload`[%(payload_1:STRING)s] AS `val` \nFROM `events`"
+    )
+    assert compiled.construct_params() == {"payload_1": "key"}
+
+    literal_sql = stmt.compile(faux_conn, compile_kwargs={"literal_binds": True}).string
+    assert literal_sql == "SELECT `events`.`payload`['key'] AS `val` \nFROM `events`"
+
+
+def test_json_getitem_integer_index(faux_conn, json_events_table):
+    stmt = sqlalchemy.select(json_events_table.c.payload[0].label("val"))
+    compiled = stmt.compile(faux_conn)
+    assert (
+        compiled.string
+        == "SELECT `events`.`payload`[%(payload_1:INT64)s] AS `val` \nFROM `events`"
+    )
+    assert compiled.construct_params() == {"payload_1": 0}
+
+    literal_sql = stmt.compile(faux_conn, compile_kwargs={"literal_binds": True}).string
+    assert literal_sql == "SELECT `events`.`payload`[0] AS `val` \nFROM `events`"
+
+
+def test_json_getitem_chained_subscripts(faux_conn, json_events_table):
+    stmt = sqlalchemy.select(json_events_table.c.payload["a"][0]["b"].label("val"))
+    compiled = stmt.compile(faux_conn)
+    assert compiled.string in (
+        "SELECT `events`.`payload`[%(payload_1:STRING)s]"
+        "[%(param_1:INT64)s][%(param_2:STRING)s] AS `val` \n"
+        "FROM `events`",
+        "SELECT ((`events`.`payload`[%(payload_1:STRING)s])"
+        "[%(param_1:INT64)s])[%(param_2:STRING)s] AS `val` \n"
+        "FROM `events`",
+    )
+    assert compiled.construct_params() == {
+        "payload_1": "a",
+        "param_1": 0,
+        "param_2": "b",
+    }
+
+    literal_sql = stmt.compile(faux_conn, compile_kwargs={"literal_binds": True}).string
+    assert literal_sql in (
+        "SELECT `events`.`payload`['a'][0]['b'] AS `val` \nFROM `events`",
+        "SELECT ((`events`.`payload`['a'])[0])['b'] AS `val` \nFROM `events`",
+    )
+
+
+@pytest.mark.parametrize(
+    "malicious_key",
+    [
+        "a` FROM `events`; DROP TABLE `x`; --",
+        "a') FROM `events`; DROP TABLE `x`; --",
+        'a" FROM `events`; --',
+        "a\\') OR 1=1 --",
+        "$[0]",
+        "a.b[0]",
+        "pct%s",
+    ],
+)
+def test_json_getitem_sql_injection_prevented(
+    faux_conn, json_events_table, malicious_key
+):
+    stmt = sqlalchemy.select(json_events_table.c.payload[malicious_key])
+    compiled = stmt.compile(faux_conn)
+    assert (
+        compiled.string
+        == "SELECT `events`.`payload`[%(payload_1:STRING)s] AS `anon_1` \nFROM `events`"
+    )
+    assert "DROP TABLE" not in compiled.string
+    assert compiled.construct_params() == {"payload_1": malicious_key}
+
+    literal_sql = stmt.compile(faux_conn, compile_kwargs={"literal_binds": True}).string
+    expected_literal = repr(malicious_key.replace("%", "%%"))
+    assert (
+        literal_sql
+        == f"SELECT `events`.`payload`[{expected_literal}] AS `anon_1` \nFROM `events`"
+    )
+
+
+def test_json_getitem_boolean_key(faux_conn, json_events_table):
+    stmt = sqlalchemy.select(json_events_table.c.payload[True])
+    compiled = stmt.compile(faux_conn)
+    assert (
+        compiled.string
+        == "SELECT `events`.`payload`[%(payload_1:STRING)s] AS `anon_1` \nFROM `events`"
+    )
+    assert compiled.construct_params() == {"payload_1": "True"}
+
+    literal_sql = stmt.compile(faux_conn, compile_kwargs={"literal_binds": True}).string
+    assert (
+        literal_sql == "SELECT `events`.`payload`['True'] AS `anon_1` \nFROM `events`"
+    )
+
+
+def test_json_getitem_bindparam_and_column_expressions(faux_conn, json_events_table):
+    # Untyped bindparam (SQLAlchemy sets bindparam.type = JSONStrIndexType class)
+    stmt_untyped = sqlalchemy.select(
+        json_events_table.c.payload[sqlalchemy.bindparam("k")]
+    )
+    compiled_untyped = stmt_untyped.compile(faux_conn)
+    assert (
+        compiled_untyped.string
+        == "SELECT `events`.`payload`[%(k:STRING)s] AS `anon_1` \nFROM `events`"
+    )
+
+    # JSONIndexType (base class)
+    bp_idx = sqlalchemy.bindparam(
+        "idx_k", "v", type_=sqlalchemy.sql.sqltypes.JSON.JSONIndexType()
+    )
+    assert (
+        sqlalchemy.select(bp_idx).compile(faux_conn).string
+        == "SELECT %(idx_k:STRING)s AS `anon_1`"
+    )
+    assert (
+        sqlalchemy.select(bp_idx)
+        .compile(faux_conn, compile_kwargs={"literal_binds": True})
+        .string
+        == "SELECT 'v' AS `anon_1`"
+    )
+
+    # Column expressions
+    stmt_col = sqlalchemy.select(
+        json_events_table.c.payload[json_events_table.c.k],
+        json_events_table.c.payload[json_events_table.c.idx],
+    )
+    assert stmt_col.compile(faux_conn).string == (
+        "SELECT `events`.`payload`[`events`.`k`] AS `anon_1`, "
+        "`events`.`payload`[`events`.`idx`] AS `anon_2` \n"
+        "FROM `events`"
+    )
+
+
+def test_json_path_getitem(faux_conn, json_events_table):
+    stmt = sqlalchemy.select(
+        json_events_table.c.payload[("a", 0, 'b"c\\d', True)].label("val")
+    )
+    compiled = stmt.compile(faux_conn)
+    assert (
+        compiled.string
+        == "SELECT JSON_QUERY(`events`.`payload`, %(payload_1:STRING)s) AS `val` \nFROM `events`"
+    )
+    raw_params = compiled.construct_params()
+    assert (
+        compiled._bind_processors["payload_1"](raw_params["payload_1"])
+        == '$."a"[0]."b\\"c\\\\d"."True"'
+    )
+
+    literal_sql = stmt.compile(faux_conn, compile_kwargs={"literal_binds": True}).string
+    assert literal_sql == (
+        "SELECT JSON_QUERY(`events`.`payload`, "
+        + repr('$."a"[0]."b\\"c\\\\d"."True"')
+        + ") AS `val` \nFROM `events`"
+    )
+
+
+def test_json_getitem_direct_operator_edge_cases(faux_conn, json_events_table):
+    from sqlalchemy.sql import operators
+    from sqlalchemy.sql.elements import BinaryExpression
+
+    # 1. json_getitem_op with sequence value routes to json_path_getitem_op
+    bp_seq = sqlalchemy.sql.elements.BindParameter(
+        "p", value=["a", 1], type_=sqlalchemy.sql.sqltypes.NullType()
+    )
+    expr_seq = BinaryExpression(
+        json_events_table.c.payload, bp_seq, operators.json_getitem_op
+    )
+    compiled_seq = sqlalchemy.select(expr_seq).compile(faux_conn)
+    assert (
+        compiled_seq.string
+        == "SELECT JSON_QUERY(`events`.`payload`, %(p:STRING)s) AS `anon_1` \nFROM `events`"
+    )
+    raw_seq_params = compiled_seq.construct_params()
+    assert compiled_seq._bind_processors["p"](raw_seq_params["p"]) == '$."a"[1]'
+
+    # 2. json_getitem_op with NullType bindparam (int and str)
+    bp_null_int = sqlalchemy.sql.elements.BindParameter(
+        "p_int", value=2, type_=sqlalchemy.sql.sqltypes.NullType()
+    )
+    expr_null_int = BinaryExpression(
+        json_events_table.c.payload, bp_null_int, operators.json_getitem_op
+    )
+    assert (
+        sqlalchemy.select(expr_null_int).compile(faux_conn).string
+        == "SELECT `events`.`payload`[%(p_int:INT64)s] AS `anon_1` \nFROM `events`"
+    )
+
+    bp_null_str = sqlalchemy.sql.elements.BindParameter(
+        "p_str", value="k", type_=sqlalchemy.sql.sqltypes.NullType()
+    )
+    expr_null_str = BinaryExpression(
+        json_events_table.c.payload, bp_null_str, operators.json_getitem_op
+    )
+    assert (
+        sqlalchemy.select(expr_null_str).compile(faux_conn).string
+        == "SELECT `events`.`payload`[%(p_str:STRING)s] AS `anon_1` \nFROM `events`"
+    )
+
+    # 3. json_path_getitem_op with non-JSONPathType BindParameter (str) and ColumnElement
+    bp_str_path = sqlalchemy.bindparam("p_path", "$.a")
+    expr_str_path = BinaryExpression(
+        json_events_table.c.payload, bp_str_path, operators.json_path_getitem_op
+    )
+    assert (
+        sqlalchemy.select(expr_str_path).compile(faux_conn).string
+        == "SELECT JSON_QUERY(`events`.`payload`, %(p_path:STRING)s) AS `anon_1` \nFROM `events`"
+    )
+
+    expr_col_path = BinaryExpression(
+        json_events_table.c.payload,
+        json_events_table.c.k,
+        operators.json_path_getitem_op,
+    )
+    assert (
+        sqlalchemy.select(expr_col_path).compile(faux_conn).string
+        == "SELECT JSON_QUERY(`events`.`payload`, `events`.`k`) AS `anon_1` \nFROM `events`"
+    )
+
+
+def test_identifier_preparer_escape_unescape():
+    import sqlalchemy_bigquery
+
+    dialect_pyformat = sqlalchemy_bigquery.BigQueryDialect(paramstyle="pyformat")
+    prep_pyformat = dialect_pyformat.identifier_preparer
+    assert prep_pyformat.quote_identifier("a`b\\c%d") == "`a\\`b\\\\c%%d`"
+    assert prep_pyformat._unescape_identifier("a\\`b\\\\c") == "a`b\\c"
+
+    dialect_named = sqlalchemy_bigquery.BigQueryDialect(paramstyle="named")
+    prep_named = dialect_named.identifier_preparer
+    assert prep_named.quote_identifier("a`b\\c%d") == "`a\\`b\\\\c%d`"

From 307581f30a5bbdbb5dc1a3bf1c93a370e61159f9 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tim=20Swe=C3=B1a=20=28Swast=29?= <swast@google.com>
Date: Thu, 12 Nov 2026 09:09:49 -0600
Subject: [PATCH 2/4] Update
 packages/sqlalchemy-bigquery/sqlalchemy_bigquery/base.py

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
---
 .../sqlalchemy-bigquery/sqlalchemy_bigquery/base.py    | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/packages/sqlalchemy-bigquery/sqlalchemy_bigquery/base.py b/packages/sqlalchemy-bigquery/sqlalchemy_bigquery/base.py
index 09b1ebe1a6e3..b5c57968d870 100644
--- a/packages/sqlalchemy-bigquery/sqlalchemy_bigquery/base.py
+++ b/packages/sqlalchemy-bigquery/sqlalchemy_bigquery/base.py
@@ -230,15 +230,15 @@ def visit_insert(self, insert_stmt, asfrom=False, **kw):
             insert_stmt, asfrom=False, **kw
         )
 
+    @staticmethod
     @staticmethod
     def _get_struct_type(type_):
         # Unwrap TypeDecorator.impl so custom types wrapping STRUCT are still
         # recognized as STRUCT columns during compilation.
-        if isinstance(type_, _struct.STRUCT):
-            return type_
-        impl = getattr(type_, "impl", None)
-        if isinstance(impl, _struct.STRUCT):
-            return impl
+        while type_ is not None:
+            if isinstance(type_, _struct.STRUCT):
+                return type_
+            type_ = getattr(type_, "impl", None)
         return None
 
     def visit_json_getitem_op_binary(self, binary, operator_, **kw):

From aff550537619f018393a9d191f17561e4a99a334 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tim=20Swe=C3=B1a?= <swast@google.com>
Date: Thu, 12 Nov 2026 16:29:54 +0000
Subject: [PATCH 3/4] apply feedback about missed BQJSONPathType

---
 .../sqlalchemy_bigquery/base.py               | 36 +++++++++++--------
 .../tests/unit/test_compiler.py               | 25 +++++++++++++
 2 files changed, 47 insertions(+), 14 deletions(-)

diff --git a/packages/sqlalchemy-bigquery/sqlalchemy_bigquery/base.py b/packages/sqlalchemy-bigquery/sqlalchemy_bigquery/base.py
index b5c57968d870..20caba1f4a0f 100644
--- a/packages/sqlalchemy-bigquery/sqlalchemy_bigquery/base.py
+++ b/packages/sqlalchemy-bigquery/sqlalchemy_bigquery/base.py
@@ -230,7 +230,6 @@ def visit_insert(self, insert_stmt, asfrom=False, **kw):
             insert_stmt, asfrom=False, **kw
         )
 
-    @staticmethod
     @staticmethod
     def _get_struct_type(type_):
         # Unwrap TypeDecorator.impl so custom types wrapping STRUCT are still
@@ -265,12 +264,17 @@ def visit_json_getitem_op_binary(self, binary, operator_, **kw):
             left = self.process(binary.left, **kw)
             return f"{left}.{self.preparer.quote_identifier(field_name)}"
 
-        # 2. For `JSON`: if a sequence (tuple/list) was passed to
-        #    `json_getitem_op`, delegate to `visit_json_path_getitem_op_binary`
-        #    so it compiles as a multi-step `JSON_QUERY` path lookup.
-        if isinstance(
-            binary.right, sqlalchemy.sql.elements.BindParameter
-        ) and _is_json_path_sequence(binary.right.value):
+        # 2. For `JSON`: if a sequence (tuple/list) or `JSONPathType` parameter
+        #    was passed to `json_getitem_op`, delegate to
+        #    `visit_json_path_getitem_op_binary` so it compiles as a multi-step
+        #    `JSON_QUERY` path lookup.
+        if isinstance(binary.right, sqlalchemy.sql.elements.BindParameter) and (
+            _is_json_path_sequence(binary.right.value)
+            or isinstance(
+                sqlalchemy.sql.type_api.to_instance(binary.right.type),
+                sqlalchemy.sql.sqltypes.JSON.JSONPathType,
+            )
+        ):
             return self.visit_json_path_getitem_op_binary(binary, operator_, **kw)
 
         # Otherwise, compile single-step `JSON` subscripting using BigQuery's
@@ -321,13 +325,17 @@ def visit_json_path_getitem_op_binary(self, binary, operator_, **kw):
         # parameterized JSONPath string formatted by `BQJSONPathType`.
         left = self.process(binary.left, **kw)
         right = binary.right
-        if isinstance(right, sqlalchemy.sql.elements.BindParameter) and not isinstance(
-            right.type, sqlalchemy.sql.sqltypes.JSON.JSONPathType
-        ):
-            right = right._clone(maintain_key=True)
-            right.type = (
-                BQJSONPathType() if _is_json_path_sequence(right.value) else BQString()
-            )
+        if isinstance(right, sqlalchemy.sql.elements.BindParameter):
+            right_type = sqlalchemy.sql.type_api.to_instance(right.type)
+            if not isinstance(right_type, sqlalchemy.sql.sqltypes.JSON.JSONPathType):
+                right_type = (
+                    BQJSONPathType()
+                    if _is_json_path_sequence(right.value)
+                    else BQString()
+                )
+            if right_type is not right.type:
+                right = right._clone(maintain_key=True)
+                right.type = right_type
         right_text = self.process(right, **kw)
         return f"JSON_QUERY({left}, {right_text})"
 
diff --git a/packages/sqlalchemy-bigquery/tests/unit/test_compiler.py b/packages/sqlalchemy-bigquery/tests/unit/test_compiler.py
index 8326237cc288..83d025b461a0 100644
--- a/packages/sqlalchemy-bigquery/tests/unit/test_compiler.py
+++ b/packages/sqlalchemy-bigquery/tests/unit/test_compiler.py
@@ -804,6 +804,31 @@ def test_json_getitem_direct_operator_edge_cases(faux_conn, json_events_table):
         == "SELECT JSON_QUERY(`events`.`payload`, `events`.`k`) AS `anon_1` \nFROM `events`"
     )
 
+    # 4. Unbound bindparam with JSONPathType (both as class on .type and via subscript)
+    bp_unbound_cls = sqlalchemy.bindparam("p_cls")
+    bp_unbound_cls.type = sqlalchemy.sql.sqltypes.JSON.JSONPathType
+    expr_unbound_cls = BinaryExpression(
+        json_events_table.c.payload, bp_unbound_cls, operators.json_path_getitem_op
+    )
+    compiled_unbound_cls = sqlalchemy.select(expr_unbound_cls).compile(faux_conn)
+    assert (
+        compiled_unbound_cls.string
+        == "SELECT JSON_QUERY(`events`.`payload`, %(p_cls:STRING)s) AS `anon_1` \nFROM `events`"
+    )
+    assert compiled_unbound_cls._bind_processors["p_cls"](("a", 0)) == '$."a"[0]'
+
+    stmt_unbound_sub = sqlalchemy.select(
+        json_events_table.c.payload[
+            sqlalchemy.bindparam("p_sub", type_=sqlalchemy.JSON.JSONPathType)
+        ]
+    )
+    compiled_unbound_sub = stmt_unbound_sub.compile(faux_conn)
+    assert (
+        compiled_unbound_sub.string
+        == "SELECT JSON_QUERY(`events`.`payload`, %(p_sub:STRING)s) AS `anon_1` \nFROM `events`"
+    )
+    assert compiled_unbound_sub._bind_processors["p_sub"](("a", 0)) == '$."a"[0]'
+
 
 def test_identifier_preparer_escape_unescape():
     import sqlalchemy_bigquery

From 429656b3db64057ce48481d300f32d09bba7250d Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tim=20Swe=C3=B1a?= <swast@google.com>
Date: Thu, 12 Nov 2026 16:33:17 +0000
Subject: [PATCH 4/4] lint

---
 packages/sqlalchemy-bigquery/tests/system/test__struct.py    | 2 ++
 .../tests/system/test_sqlalchemy_bigquery.py                 | 5 ++++-
 packages/sqlalchemy-bigquery/tests/unit/test__struct.py      | 1 +
 packages/sqlalchemy-bigquery/tests/unit/test_compiler.py     | 1 +
 4 files changed, 8 insertions(+), 1 deletion(-)

diff --git a/packages/sqlalchemy-bigquery/tests/system/test__struct.py b/packages/sqlalchemy-bigquery/tests/system/test__struct.py
index afcc6f6cc3d2..8d21ddd88315 100644
--- a/packages/sqlalchemy-bigquery/tests/system/test__struct.py
+++ b/packages/sqlalchemy-bigquery/tests/system/test__struct.py
@@ -22,6 +22,7 @@
 import packaging.version
 import pytest
 import sqlalchemy
+
 import sqlalchemy_bigquery
 
 
@@ -135,6 +136,7 @@ def test_unnest_and_struct_access_233(engine, bigquery_dataset, metadata):
 
     from sqlalchemy import ARRAY, Column, String, Table, func, select
     from sqlalchemy.orm import sessionmaker
+
     from sqlalchemy_bigquery import STRUCT
 
     conn = engine.connect()
diff --git a/packages/sqlalchemy-bigquery/tests/system/test_sqlalchemy_bigquery.py b/packages/sqlalchemy-bigquery/tests/system/test_sqlalchemy_bigquery.py
index 280fe358c2e9..57bca04865ab 100644
--- a/packages/sqlalchemy-bigquery/tests/system/test_sqlalchemy_bigquery.py
+++ b/packages/sqlalchemy-bigquery/tests/system/test_sqlalchemy_bigquery.py
@@ -24,7 +24,6 @@
 import packaging.version
 import pytest
 import sqlalchemy
-import sqlalchemy_bigquery
 from google.cloud.bigquery import TimePartitioning
 from sqlalchemy import (
     Column,
@@ -44,6 +43,8 @@
 from sqlalchemy.orm import sessionmaker
 from sqlalchemy.sql import expression
 
+import sqlalchemy_bigquery
+
 SQLALCHEMY_VERSION = packaging.version.parse(sqlalchemy.__version__)
 
 ONE_ROW_CONTENTS = [
@@ -829,6 +830,7 @@ def test_huge_in():
 )
 def test_unnest(engine, bigquery_dataset):
     from sqlalchemy import String, func, select
+
     from sqlalchemy_bigquery import ARRAY
 
     conn = engine.connect()
@@ -859,6 +861,7 @@ def test_unnest(engine, bigquery_dataset):
 )
 def test_unnest_with_cte(engine, bigquery_dataset):
     from sqlalchemy import String, func, select
+
     from sqlalchemy_bigquery import ARRAY
 
     conn = engine.connect()
diff --git a/packages/sqlalchemy-bigquery/tests/unit/test__struct.py b/packages/sqlalchemy-bigquery/tests/unit/test__struct.py
index 13389136033f..82f0bf6b7a90 100644
--- a/packages/sqlalchemy-bigquery/tests/unit/test__struct.py
+++ b/packages/sqlalchemy-bigquery/tests/unit/test__struct.py
@@ -190,6 +190,7 @@ def test_struct_compiler_rejects_invalid_direct_json_getitem_op(faux_conn):
 def test_struct_type_decorator_and_identifier_escaping(faux_conn):
     from sqlalchemy.sql import operators
     from sqlalchemy.sql.elements import BinaryExpression
+
     from sqlalchemy_bigquery import STRUCT
 
     class WrappedStruct(sqlalchemy.types.TypeDecorator):
diff --git a/packages/sqlalchemy-bigquery/tests/unit/test_compiler.py b/packages/sqlalchemy-bigquery/tests/unit/test_compiler.py
index 83d025b461a0..669494facf12 100644
--- a/packages/sqlalchemy-bigquery/tests/unit/test_compiler.py
+++ b/packages/sqlalchemy-bigquery/tests/unit/test_compiler.py
@@ -20,6 +20,7 @@
 import pytest
 import sqlalchemy.exc
 from sqlalchemy.sql.functions import cube, grouping_sets, rollup
+
 from sqlalchemy_bigquery import SafeCast, safe_cast
 
 from .conftest import setup_table, sqlalchemy_2_0_or_higher, sqlalchemy_before_2_0