diff --git a/packages/django-google-spanner/django_spanner/features.py b/packages/django-google-spanner/django_spanner/features.py
index 8b182ebdb4e0..b424486b7e4a 100644
--- a/packages/django-google-spanner/django_spanner/features.py
+++ b/packages/django-google-spanner/django_spanner/features.py
@@ -121,15 +121,11 @@ def introspected_field_types(self):
         "db_functions.comparison.test_cast.CastTests.test_cast_to_decimal_field",
         "db_functions.datetime.test_extract_trunc.DateFunctionTests.test_extract_func",
         "db_functions.datetime.test_extract_trunc.DateFunctionTests.test_extract_iso_weekday_func",
-        "db_functions.datetime.test_extract_trunc.DateFunctionTests.test_extract_lookup_name_sql_injection",
-        "db_functions.datetime.test_extract_trunc.DateFunctionTests.test_trunc_lookup_name_sql_injection",
         "db_functions.datetime.test_extract_trunc.DateFunctionTests.test_trunc_time_comparison",
         "db_functions.datetime.test_extract_trunc.DateFunctionWithTimeZoneTests.test_extract_func",
         "db_functions.datetime.test_extract_trunc.DateFunctionWithTimeZoneTests.test_extract_func_with_timezone",
         "db_functions.datetime.test_extract_trunc.DateFunctionWithTimeZoneTests.test_extract_iso_weekday_func",
-        "db_functions.datetime.test_extract_trunc.DateFunctionWithTimeZoneTests.test_extract_lookup_name_sql_injection",
         "db_functions.datetime.test_extract_trunc.DateFunctionWithTimeZoneTests.test_trunc_func_with_timezone",
-        "db_functions.datetime.test_extract_trunc.DateFunctionWithTimeZoneTests.test_trunc_lookup_name_sql_injection",
         "db_functions.datetime.test_extract_trunc.DateFunctionWithTimeZoneTests.test_trunc_time_comparison",
         "db_functions.datetime.test_extract_trunc.DateFunctionWithTimeZoneTests.test_trunc_timezone_applied_before_truncation",
         "db_functions.json.test_json_object.JSONObjectTests.test_basic",
diff --git a/packages/django-google-spanner/django_spanner/operations.py b/packages/django-google-spanner/django_spanner/operations.py
index 5a5b8e3e4612..8cf6c466cf07 100644
--- a/packages/django-google-spanner/django_spanner/operations.py
+++ b/packages/django-google-spanner/django_spanner/operations.py
@@ -12,9 +12,11 @@
 
 from django.conf import settings
 from django.db.backends.base.operations import BaseDatabaseOperations
-from django.db.utils import DatabaseError
+from django.db.backends.utils import split_tzname_delta
+from django.db.utils import DatabaseError, OperationalError
 from django.utils import timezone
 from django.utils.duration import duration_microseconds
+from django.utils.regex_helper import _lazy_re_compile
 from google.cloud.spanner_dbapi.parse_utils import (
     DateStr,
     TimestampStr,
@@ -25,6 +27,12 @@
 class DatabaseOperations(BaseDatabaseOperations):
     """A Spanner-specific version of Django database operations."""
 
+    # EXTRACT and TRUNC date parts cannot be passed as query parameters in SQL.
+    # Following the practice of other Django database backends (PostgreSQL,
+    # MySQL, Oracle), validate lookup types with an identifier regex to prevent
+    # SQL injection while allowing custom user-registered lookups.
+    extract_trunc_lookup_pattern = _lazy_re_compile(r"^[a-zA-Z_]+$")
+
     # Cloud Spanner only supports INT64, so all integer fields can hold
     # 64-bit values.
     integer_field_ranges = {
@@ -406,6 +414,48 @@ def convert_uuidfield_value(self, value, expression, connection):
             value = UUID(value)
         return value
 
+    _well_formatted_offset_pattern = _lazy_re_compile(r"^[+-]\d{2}:\d{2}$")
+    _four_digit_offset_pattern = _lazy_re_compile(r"([+-])(\d{2})(\d{2})$")
+    _two_digit_offset_pattern = _lazy_re_compile(r"([+-])(\d{2})$")
+
+    def _prepare_tzname_delta(self, tzname):
+        if not tzname or tzname == "UTC":
+            return "UTC"
+        if "+" not in tzname and "-" not in tzname:
+            return tzname
+        if self._well_formatted_offset_pattern.fullmatch(tzname):
+            return tzname
+        if tzname.startswith("UTC") and self._well_formatted_offset_pattern.fullmatch(
+            tzname[3:]
+        ):
+            return tzname[3:]
+
+        tzname = self._four_digit_offset_pattern.sub(r"\1\2:\3", tzname)
+        tzname = self._two_digit_offset_pattern.sub(r"\1\2:00", tzname)
+        tzname, sign, offset = split_tzname_delta(tzname)
+        if offset:
+            if len(offset) in (7, 8) and offset.endswith(":00"):
+                offset = offset[:-3]
+            if len(offset) == 4 and offset.isdigit():
+                offset = f"{offset[:2]}:{offset[2:]}"
+            return f"{sign}{offset}"
+        return tzname
+
+    def _validate_lookup_type(self, lookup_type):
+        if not isinstance(
+            lookup_type, str
+        ) or not self.extract_trunc_lookup_pattern.fullmatch(lookup_type):
+            raise ValueError(f"Invalid lookup type: {lookup_type!r}")
+
+    # Django's test_trunc_lookup_name_sql_injection catches only (DataError, OperationalError),
+    # unlike test_extract_lookup_name_sql_injection which also catches ValueError.
+    # OperationalError is raised here to satisfy Django's trunc exception contract.
+    def _validate_trunc_lookup_type(self, lookup_type):
+        if not isinstance(
+            lookup_type, str
+        ) or not self.extract_trunc_lookup_pattern.fullmatch(lookup_type):
+            raise OperationalError(f"Invalid trunc lookup type: {lookup_type!r}")
+
     def date_extract_sql(self, lookup_type, field_name, params=None):
         """Extract date from the lookup.
 
@@ -422,7 +472,8 @@ def date_extract_sql(self, lookup_type, field_name, params=None):
         :returns: A SQL statement for extracting.
         """
         lookup_type = self.extract_names.get(lookup_type, lookup_type)
-        sql = "EXTRACT(%s FROM %s)" % (lookup_type, field_name)
+        self._validate_lookup_type(lookup_type)
+        sql = f"EXTRACT({lookup_type} FROM {field_name})"
         return sql, params
 
     def datetime_extract_sql(self, lookup_type, field_name, params, tzname):
@@ -442,15 +493,13 @@ def datetime_extract_sql(self, lookup_type, field_name, params, tzname):
         :returns: A SQL statement for extracting.
         """
         tzname = tzname if settings.USE_TZ and tzname else "UTC"
+        tzname_param = self._prepare_tzname_delta(tzname)
         lookup_type = self.extract_names.get(lookup_type, lookup_type)
+        self._validate_lookup_type(lookup_type)
+        params = () if params is None else tuple(params)
         return (
-            'EXTRACT(%s FROM %s AT TIME ZONE "%s")'
-            % (
-                lookup_type,
-                field_name,
-                tzname,
-            ),
-            params,
+            f"EXTRACT({lookup_type} FROM {field_name} AT TIME ZONE %s)",
+            (*params, tzname_param),
         )
 
     def time_extract_sql(self, lookup_type, field_name, params=None):
@@ -468,12 +517,12 @@ def time_extract_sql(self, lookup_type, field_name, params=None):
         :rtype: str
         :returns: A SQL statement for extracting.
         """
+        lookup_type = self.extract_names.get(lookup_type, lookup_type)
+        self._validate_lookup_type(lookup_type)
+        params = () if params is None else tuple(params)
         # Time is stored as TIMESTAMP with UTC time zone.
-        sql = 'EXTRACT(%s FROM %s AT TIME ZONE "UTC")' % (
-            lookup_type,
-            field_name,
-        )
-        return sql, params
+        sql = f"EXTRACT({lookup_type} FROM {field_name} AT TIME ZONE %s)"
+        return sql, (*params, "UTC")
 
     def date_trunc_sql(self, lookup_type, field_name, params, tzname=None):
         """Truncate date in the lookup.
@@ -494,16 +543,14 @@ def date_trunc_sql(self, lookup_type, field_name, params, tzname=None):
         :rtype: str
         :returns: A SQL statement for truncating.
         """
+        self._validate_trunc_lookup_type(lookup_type)
         # https://cloud.google.com/spanner/docs/functions-and-operators#date_trunc
         if lookup_type == "week":
             # Spanner truncates to Sunday but Django expects Monday. First,
             # subtract a day so that a Sunday will be truncated to the previous
             # week...
             field_name = "DATE_SUB(CAST(" + field_name + " AS DATE), INTERVAL 1 DAY)"
-        sql = "DATE_TRUNC(CAST(%s AS DATE), %s)" % (
-            field_name,
-            lookup_type,
-        )
+        sql = f"DATE_TRUNC(CAST({field_name} AS DATE), {lookup_type})"
         if lookup_type == "week":
             # ...then add a day to get from Sunday to Monday.
             sql = "DATE_ADD(CAST(" + sql + " AS DATE), INTERVAL 1 DAY)"
@@ -529,20 +576,19 @@ def datetime_trunc_sql(self, lookup_type, field_name, params, tzname="UTC"):
         """
         # https://cloud.google.com/spanner/docs/functions-and-operators#timestamp_trunc
         tzname = tzname if settings.USE_TZ and tzname else "UTC"
+        tzname_param = self._prepare_tzname_delta(tzname)
+        self._validate_trunc_lookup_type(lookup_type)
+        params = () if params is None else tuple(params)
         if lookup_type == "week":
             # Spanner truncates to Sunday but Django expects Monday. First,
             # subtract a day so that a Sunday will be truncated to the previous
             # week...
             field_name = "TIMESTAMP_SUB(" + field_name + ", INTERVAL 1 DAY)"
-        sql = 'TIMESTAMP_TRUNC(%s, %s, "%s")' % (
-            field_name,
-            lookup_type,
-            tzname,
-        )
+        sql = f"TIMESTAMP_TRUNC({field_name}, {lookup_type}, %s)"
         if lookup_type == "week":
             # ...then add a day to get from Sunday to Monday.
             sql = "TIMESTAMP_ADD(" + sql + ", INTERVAL 1 DAY)"
-        return sql, params
+        return sql, (*params, tzname_param)
 
     def time_trunc_sql(self, lookup_type, field_name, params, tzname="UTC"):
         """Truncate time in the lookup.
@@ -564,14 +610,12 @@ def time_trunc_sql(self, lookup_type, field_name, params, tzname="UTC"):
         """
         # https://cloud.google.com/spanner/docs/functions-and-operators#timestamp_trunc
         tzname = tzname if settings.USE_TZ and tzname else "UTC"
+        tzname_param = self._prepare_tzname_delta(tzname)
+        self._validate_trunc_lookup_type(lookup_type)
+        params = () if params is None else tuple(params)
         return (
-            'TIMESTAMP_TRUNC(%s, %s, "%s")'
-            % (
-                field_name,
-                lookup_type,
-                tzname,
-            ),
-            params,
+            f"TIMESTAMP_TRUNC({field_name}, {lookup_type}, %s)",
+            (*params, tzname_param),
         )
 
     def datetime_cast_date_sql(self, field_name, params, tzname):
@@ -592,7 +636,9 @@ def datetime_cast_date_sql(self, field_name, params, tzname):
         """
         # https://cloud.google.com/spanner/docs/functions-and-operators#date
         tzname = tzname if settings.USE_TZ and tzname else "UTC"
-        return 'DATE(%s, "%s")' % (field_name, tzname), params
+        tzname_param = self._prepare_tzname_delta(tzname)
+        params = () if params is None else tuple(params)
+        return f"DATE({field_name}, %s)", (*params, tzname_param)
 
     def datetime_cast_time_sql(self, field_name, params, tzname):
         """Cast time in the lookup.
@@ -611,12 +657,14 @@ def datetime_cast_time_sql(self, field_name, params, tzname):
         :returns: A SQL statement for casting.
         """
         tzname = tzname if settings.USE_TZ and tzname else "UTC"
+        tzname_param = self._prepare_tzname_delta(tzname)
+        params = () if params is None else tuple(params)
         # Cloud Spanner doesn't have a function for converting
         # TIMESTAMP to another time zone.
         return (
-            "TIMESTAMP(FORMAT_TIMESTAMP("
-            "'%%Y-%%m-%%d %%R:%%E9S %%Z', %s, '%s'))" % (field_name, tzname)
-        ), params
+            f"TIMESTAMP(FORMAT_TIMESTAMP('%Y-%m-%d %R:%E9S %Z', {field_name}, %s))",
+            (*params, tzname_param),
+        )
 
     def date_interval_sql(self, timedelta):
         """Get a date interval in microseconds.
diff --git a/packages/django-google-spanner/tests/mockserver_tests/models.py b/packages/django-google-spanner/tests/mockserver_tests/models.py
index 66cafc60412f..4d3e8db68de6 100644
--- a/packages/django-google-spanner/tests/mockserver_tests/models.py
+++ b/packages/django-google-spanner/tests/mockserver_tests/models.py
@@ -18,3 +18,8 @@
 class Singer(models.Model):
     first_name = models.CharField(max_length=200)
     last_name = models.CharField(max_length=200)
+
+
+class Event(models.Model):
+    name = models.CharField(max_length=200)
+    start_time = models.DateTimeField()
diff --git a/packages/django-google-spanner/tests/mockserver_tests/test_datetime_expressions.py b/packages/django-google-spanner/tests/mockserver_tests/test_datetime_expressions.py
new file mode 100644
index 000000000000..6bde1bd4a61a
--- /dev/null
+++ b/packages/django-google-spanner/tests/mockserver_tests/test_datetime_expressions.py
@@ -0,0 +1,287 @@
+# Copyright 2026 Google LLC All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import zoneinfo
+from datetime import datetime, timedelta, tzinfo
+from datetime import timezone as dt_timezone
+
+from django.db.models import Value
+from django.db.models.functions import (
+    Coalesce,
+    Extract,
+    ExtractHour,
+    Trunc,
+    TruncDate,
+    TruncDay,
+    TruncTime,
+    TruncWeek,
+)
+from django.db.utils import OperationalError
+from django.test import override_settings
+from django.utils import timezone
+from google.cloud.spanner_v1 import ExecuteSqlRequest, TypeCode
+
+from tests.mockserver_tests.mock_server_test_base import (
+    MockServerTestBase,
+    add_single_result,
+)
+from tests.mockserver_tests.models import Event
+
+
+class TestDatetimeExpressions(MockServerTestBase):
+    def get_last_execute_sql_request(self) -> ExecuteSqlRequest:
+        execute_requests = [
+            request
+            for request in self.spanner_service.requests
+            if isinstance(request, ExecuteSqlRequest)
+        ]
+        self.assertTrue(
+            len(execute_requests) > 0,
+            "Expected at least one ExecuteSqlRequest in recorded requests",
+        )
+        return execute_requests[-1]
+
+    def test_extract_hour_parameterized(self):
+        sql = (
+            "SELECT EXTRACT(hour FROM tests_event.start_time AT TIME ZONE @a0) "
+            "AS extracted_hour FROM tests_event"
+        )
+        add_single_result(sql, "extracted_hour", TypeCode.INT64, [("12",)])
+        results = list(
+            Event.objects.annotate(
+                extracted_hour=ExtractHour("start_time")
+            ).values_list("extracted_hour", flat=True)
+        )
+        self.assertEqual(results, [12])
+        execute_request = self.get_last_execute_sql_request()
+        self.assertEqual(execute_request.sql, sql)
+        self.assertEqual(execute_request.params["a0"], "UTC")
+        self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING)
+
+    def test_extract_named_timezone_parameterized(self):
+        new_york_timezone = zoneinfo.ZoneInfo("America/New_York")
+        sql = (
+            "SELECT EXTRACT(hour FROM tests_event.start_time AT TIME ZONE @a0) "
+            "AS extracted_hour FROM tests_event"
+        )
+        add_single_result(sql, "extracted_hour", TypeCode.INT64, [("8",)])
+        results = list(
+            Event.objects.annotate(
+                extracted_hour=ExtractHour("start_time", tzinfo=new_york_timezone)
+            ).values_list("extracted_hour", flat=True)
+        )
+        self.assertEqual(results, [8])
+        execute_request = self.get_last_execute_sql_request()
+        self.assertEqual(execute_request.sql, sql)
+        self.assertEqual(execute_request.params["a0"], "America/New_York")
+        self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING)
+
+    def test_extract_fixed_offset_timezone_normalized(self):
+        fixed_offset_timezone = timezone.get_fixed_timezone(-283)
+        sql = (
+            "SELECT EXTRACT(hour FROM tests_event.start_time AT TIME ZONE @a0) "
+            "AS extracted_hour FROM tests_event"
+        )
+        add_single_result(sql, "extracted_hour", TypeCode.INT64, [("7",)])
+        results = list(
+            Event.objects.annotate(
+                extracted_hour=ExtractHour("start_time", tzinfo=fixed_offset_timezone)
+            ).values_list("extracted_hour", flat=True)
+        )
+        self.assertEqual(results, [7])
+        execute_request = self.get_last_execute_sql_request()
+        self.assertEqual(execute_request.sql, sql)
+        self.assertEqual(execute_request.params["a0"], "-04:43")
+        self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING)
+
+    def test_trunc_day_parameterized(self):
+        tokyo_timezone = zoneinfo.ZoneInfo("Asia/Tokyo")
+        sql = (
+            "SELECT TIMESTAMP_TRUNC(tests_event.start_time, day, @a0) "
+            "AS truncated_day FROM tests_event"
+        )
+        add_single_result(
+            sql, "truncated_day", TypeCode.TIMESTAMP, [("2026-11-10T00:00:00Z",)]
+        )
+        results = list(
+            Event.objects.annotate(
+                truncated_day=TruncDay("start_time", tzinfo=tokyo_timezone)
+            ).values_list("truncated_day", flat=True)
+        )
+        self.assertEqual(len(results), 1)
+        execute_request = self.get_last_execute_sql_request()
+        self.assertEqual(execute_request.sql, sql)
+        self.assertEqual(execute_request.params["a0"], "Asia/Tokyo")
+        self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING)
+
+    def test_trunc_week_parameterized(self):
+        new_york_timezone = zoneinfo.ZoneInfo("America/New_York")
+        sql = (
+            "SELECT TIMESTAMP_ADD(TIMESTAMP_TRUNC(TIMESTAMP_SUB(tests_event.start_time, INTERVAL 1 DAY), "
+            "week, @a0), INTERVAL 1 DAY) AS truncated_week FROM tests_event"
+        )
+        add_single_result(
+            sql, "truncated_week", TypeCode.TIMESTAMP, [("2026-11-09T00:00:00Z",)]
+        )
+        results = list(
+            Event.objects.annotate(
+                truncated_week=TruncWeek("start_time", tzinfo=new_york_timezone)
+            ).values_list("truncated_week", flat=True)
+        )
+        self.assertEqual(len(results), 1)
+        execute_request = self.get_last_execute_sql_request()
+        self.assertEqual(execute_request.sql, sql)
+        self.assertEqual(execute_request.params["a0"], "America/New_York")
+        self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING)
+
+    def test_trunc_date_and_time_cast_parameterized(self):
+        new_york_timezone = zoneinfo.ZoneInfo("America/New_York")
+        sql_date = "SELECT DATE(tests_event.start_time, @a0) AS truncated_date FROM tests_event"
+        add_single_result(sql_date, "truncated_date", TypeCode.DATE, [("2026-11-10",)])
+        results_date = list(
+            Event.objects.annotate(
+                truncated_date=TruncDate("start_time", tzinfo=new_york_timezone)
+            ).values_list("truncated_date", flat=True)
+        )
+        self.assertEqual(len(results_date), 1)
+        execute_request_date = self.get_last_execute_sql_request()
+        self.assertEqual(execute_request_date.sql, sql_date)
+        self.assertEqual(execute_request_date.params["a0"], "America/New_York")
+
+        sql_time = (
+            "SELECT TIMESTAMP(FORMAT_TIMESTAMP('%Y-%m-%d %R:%E9S %Z', "
+            "tests_event.start_time, @a0)) AS truncated_time FROM tests_event"
+        )
+        add_single_result(
+            sql_time, "truncated_time", TypeCode.TIMESTAMP, [("2026-11-10T12:00:00Z",)]
+        )
+        results_time = list(
+            Event.objects.annotate(
+                truncated_time=TruncTime("start_time", tzinfo=new_york_timezone)
+            ).values_list("truncated_time", flat=True)
+        )
+        self.assertEqual(len(results_time), 1)
+        execute_request_time = self.get_last_execute_sql_request()
+        self.assertEqual(execute_request_time.sql, sql_time)
+        self.assertEqual(execute_request_time.params["a0"], "America/New_York")
+
+    def test_extract_compound_expression_preserves_parameter_order(self):
+        new_york_timezone = zoneinfo.ZoneInfo("America/New_York")
+        fallback_datetime = datetime(2026, 1, 1, 15, 0, tzinfo=dt_timezone.utc)
+        sql = (
+            "SELECT EXTRACT(hour FROM COALESCE(tests_event.start_time, @a0) "
+            "AT TIME ZONE @a1) AS extracted_hour FROM tests_event"
+        )
+        add_single_result(sql, "extracted_hour", TypeCode.INT64, [("10",)])
+        results = list(
+            Event.objects.annotate(
+                extracted_hour=ExtractHour(
+                    Coalesce("start_time", Value(fallback_datetime)),
+                    tzinfo=new_york_timezone,
+                )
+            ).values_list("extracted_hour", flat=True)
+        )
+        self.assertEqual(results, [10])
+        execute_request = self.get_last_execute_sql_request()
+        self.assertEqual(execute_request.sql, sql)
+        self.assertEqual(execute_request.params["a0"], "2026-01-01T15:00:00.000000Z")
+        self.assertEqual(execute_request.params["a1"], "America/New_York")
+        self.assertEqual(execute_request.param_types["a0"].code, TypeCode.TIMESTAMP)
+        self.assertEqual(execute_request.param_types["a1"].code, TypeCode.STRING)
+
+    @override_settings(USE_TZ=False)
+    def test_extract_use_tz_false_defaults_to_utc(self):
+        new_york_timezone = zoneinfo.ZoneInfo("America/New_York")
+        sql = (
+            "SELECT EXTRACT(hour FROM tests_event.start_time AT TIME ZONE @a0) "
+            "AS extracted_hour FROM tests_event"
+        )
+        add_single_result(sql, "extracted_hour", TypeCode.INT64, [("12",)])
+        results = list(
+            Event.objects.annotate(
+                extracted_hour=ExtractHour("start_time", tzinfo=new_york_timezone)
+            ).values_list("extracted_hour", flat=True)
+        )
+        self.assertEqual(results, [12])
+        execute_request = self.get_last_execute_sql_request()
+        self.assertEqual(execute_request.sql, sql)
+        self.assertEqual(execute_request.params["a0"], "UTC")
+        self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING)
+
+    def test_tzname_sql_injection_travels_safely_as_parameter(self):
+        class MaliciousTimezone(tzinfo):
+            def utcoffset(self, dt):
+                return timedelta(0)
+
+            def tzname(self, dt):
+                return "UTC'; DROP TABLE tests_event; --"
+
+            def dst(self, dt):
+                return timedelta(0)
+
+        malicious_timezone = MaliciousTimezone()
+        sql = (
+            "SELECT EXTRACT(hour FROM tests_event.start_time AT TIME ZONE @a0) "
+            "AS extracted_hour FROM tests_event"
+        )
+        add_single_result(sql, "extracted_hour", TypeCode.INT64, [("1",)])
+        results = list(
+            Event.objects.annotate(
+                extracted_hour=ExtractHour("start_time", tzinfo=malicious_timezone)
+            ).values_list("extracted_hour", flat=True)
+        )
+        self.assertEqual(results, [1])
+        execute_request = self.get_last_execute_sql_request()
+        self.assertEqual(execute_request.sql, sql)
+        self.assertNotIn("DROP TABLE", execute_request.sql)
+        self.assertEqual(
+            execute_request.params["a0"], "UTC'; DROP TABLE tests_event; --"
+        )
+        self.assertEqual(execute_request.param_types["a0"].code, TypeCode.STRING)
+
+    def test_extract_rejects_malicious_lookup_type_before_rpc(self):
+        with self.assertRaises(ValueError):
+            list(
+                Event.objects.annotate(
+                    injected=Extract("start_time", "hour'; DROP TABLE tests_event; --")
+                ).values_list("injected", flat=True)
+            )
+        execute_requests = [
+            request
+            for request in self.spanner_service.requests
+            if isinstance(request, ExecuteSqlRequest)
+        ]
+        self.assertEqual(
+            len(execute_requests),
+            0,
+            "Expected zero ExecuteSqlRequest when lookup_type is invalid",
+        )
+
+    def test_trunc_rejects_malicious_lookup_type_before_rpc(self):
+        with self.assertRaises(OperationalError):
+            list(
+                Event.objects.annotate(
+                    injected=Trunc("start_time", "day'; DROP TABLE tests_event; --")
+                ).values_list("injected", flat=True)
+            )
+        execute_requests = [
+            request
+            for request in self.spanner_service.requests
+            if isinstance(request, ExecuteSqlRequest)
+        ]
+        self.assertEqual(
+            len(execute_requests),
+            0,
+            "Expected zero ExecuteSqlRequest when lookup_type is invalid",
+        )
diff --git a/packages/django-google-spanner/tests/unit/django_spanner/test_operations.py b/packages/django-google-spanner/tests/unit/django_spanner/test_operations.py
index 520d268bce10..cd2c19ee0dc6 100644
--- a/packages/django-google-spanner/tests/unit/django_spanner/test_operations.py
+++ b/packages/django-google-spanner/tests/unit/django_spanner/test_operations.py
@@ -12,7 +12,7 @@
 
 from django.conf import settings
 from django.core.management.color import no_style
-from django.db.utils import DatabaseError
+from django.db.utils import DatabaseError, OperationalError
 from google.cloud.spanner_dbapi.types import DateStr
 
 from tests.unit.django_spanner.simple_test import SpannerSimpleTestClass
@@ -130,8 +130,8 @@ def test_datetime_extract_sql(self):
                 "dayofweek", "dummy_field", None, "IST"
             ),
             (
-                'EXTRACT(dayofweek FROM dummy_field AT TIME ZONE "IST")',
-                None,
+                "EXTRACT(dayofweek FROM dummy_field AT TIME ZONE %s)",
+                ("IST",),
             ),
         )
 
@@ -142,8 +142,8 @@ def test_datetime_extract_sql_use_tz_false(self):
                 "dayofweek", "dummy_field", None, "IST"
             ),
             (
-                'EXTRACT(dayofweek FROM dummy_field AT TIME ZONE "UTC")',
-                None,
+                "EXTRACT(dayofweek FROM dummy_field AT TIME ZONE %s)",
+                ("UTC",),
             ),
         )
         settings.USE_TZ = True  # reset changes.
@@ -152,21 +152,21 @@ def test_time_extract_sql(self):
         self.assertEqual(
             self.db_operations.time_extract_sql("dayofweek", "dummy_field"),
             (
-                'EXTRACT(dayofweek FROM dummy_field AT TIME ZONE "UTC")',
-                None,
+                "EXTRACT(dayofweek FROM dummy_field AT TIME ZONE %s)",
+                ("UTC",),
             ),
         )
 
     def test_time_trunc_sql(self):
         self.assertEqual(
             self.db_operations.time_trunc_sql("dayofweek", "dummy_field", None),
-            ('TIMESTAMP_TRUNC(dummy_field, dayofweek, "UTC")', None),
+            ("TIMESTAMP_TRUNC(dummy_field, dayofweek, %s)", ("UTC",)),
         )
 
     def test_datetime_cast_date_sql(self):
         self.assertEqual(
             self.db_operations.datetime_cast_date_sql("dummy_field", None, "IST"),
-            ('DATE(dummy_field, "IST")', None),
+            ("DATE(dummy_field, %s)", ("IST",)),
         )
 
     def test_datetime_cast_time_sql(self):
@@ -174,8 +174,8 @@ def test_datetime_cast_time_sql(self):
         self.assertEqual(
             self.db_operations.datetime_cast_time_sql("dummy_field", None, "IST"),
             (
-                "TIMESTAMP(FORMAT_TIMESTAMP('%Y-%m-%d %R:%E9S %Z', dummy_field, 'IST'))",
-                None,
+                "TIMESTAMP(FORMAT_TIMESTAMP('%Y-%m-%d %R:%E9S %Z', dummy_field, %s))",
+                ("IST",),
             ),
         )
 
@@ -184,8 +184,8 @@ def test_datetime_cast_time_sql_use_tz_false(self):
         self.assertEqual(
             self.db_operations.datetime_cast_time_sql("dummy_field", None, "IST"),
             (
-                "TIMESTAMP(FORMAT_TIMESTAMP('%Y-%m-%d %R:%E9S %Z', dummy_field, 'UTC'))",
-                None,
+                "TIMESTAMP(FORMAT_TIMESTAMP('%Y-%m-%d %R:%E9S %Z', dummy_field, %s))",
+                ("UTC",),
             ),
         )
         settings.USE_TZ = True  # reset changes.
@@ -327,23 +327,200 @@ def test_date_and_time_trunc_sql(self):
         self.assertIsNone(params)
 
         sql, params = self.db_operations.time_trunc_sql("hour", "field", None)
-        self.assertEqual(sql, 'TIMESTAMP_TRUNC(field, hour, "UTC")')
-        self.assertIsNone(params)
+        self.assertEqual(sql, "TIMESTAMP_TRUNC(field, hour, %s)")
+        self.assertEqual(params, ("UTC",))
 
         sql, params = self.db_operations.datetime_trunc_sql(
             "day", "field", None, tzname="UTC"
         )
-        self.assertEqual(sql, 'TIMESTAMP_TRUNC(field, day, "UTC")')
+        self.assertEqual(sql, "TIMESTAMP_TRUNC(field, day, %s)")
+        self.assertEqual(params, ("UTC",))
+
+        sql, params = self.db_operations.date_trunc_sql("week", "field", None)
+        self.assertEqual(
+            sql,
+            "DATE_ADD(CAST(DATE_TRUNC(CAST(DATE_SUB(CAST(field AS DATE), INTERVAL 1 DAY) AS DATE), week) AS DATE), INTERVAL 1 DAY)",
+        )
         self.assertIsNone(params)
 
+        sql, params = self.db_operations.datetime_trunc_sql(
+            "week", "field", None, tzname="UTC"
+        )
+        self.assertEqual(
+            sql,
+            "TIMESTAMP_ADD(TIMESTAMP_TRUNC(TIMESTAMP_SUB(field, INTERVAL 1 DAY), week, %s), INTERVAL 1 DAY)",
+        )
+        self.assertEqual(params, ("UTC",))
+
     def test_datetime_cast_sql(self):
         sql, params = self.db_operations.datetime_cast_date_sql("field", None, "UTC")
-        self.assertEqual(sql, 'DATE(field, "UTC")')
-        self.assertIsNone(params)
+        self.assertEqual(sql, "DATE(field, %s)")
+        self.assertEqual(params, ("UTC",))
 
         sql, params = self.db_operations.datetime_cast_time_sql("field", None, "UTC")
         self.assertEqual(
             sql,
-            "TIMESTAMP(FORMAT_TIMESTAMP('%Y-%m-%d %R:%E9S %Z', field, 'UTC'))",
+            "TIMESTAMP(FORMAT_TIMESTAMP('%Y-%m-%d %R:%E9S %Z', field, %s))",
         )
-        self.assertIsNone(params)
+        self.assertEqual(params, ("UTC",))
+
+    def test_prepare_tzname_delta(self):
+        self.assertEqual(
+            self.db_operations._prepare_tzname_delta("UTC-04:43"), "-04:43"
+        )
+        self.assertEqual(
+            self.db_operations._prepare_tzname_delta("UTC+05:00"), "+05:00"
+        )
+        self.assertEqual(
+            self.db_operations._prepare_tzname_delta("UTC+05:30:00"), "+05:30"
+        )
+        self.assertEqual(self.db_operations._prepare_tzname_delta("-0443"), "-04:43")
+        self.assertEqual(self.db_operations._prepare_tzname_delta("+0500"), "+05:00")
+        self.assertEqual(self.db_operations._prepare_tzname_delta("+03:00"), "+03:00")
+        self.assertEqual(
+            self.db_operations._prepare_tzname_delta("+05:30:00"), "+05:30"
+        )
+        self.assertEqual(
+            self.db_operations._prepare_tzname_delta("-05:00:00"), "-05:00"
+        )
+        self.assertEqual(self.db_operations._prepare_tzname_delta("+05:30"), "+05:30")
+        self.assertEqual(self.db_operations._prepare_tzname_delta("-04:43"), "-04:43")
+        self.assertEqual(self.db_operations._prepare_tzname_delta("UTC-0443"), "-04:43")
+        self.assertEqual(self.db_operations._prepare_tzname_delta("UTC+0500"), "+05:00")
+        self.assertEqual(self.db_operations._prepare_tzname_delta("+05"), "+05:00")
+        self.assertEqual(self.db_operations._prepare_tzname_delta("-08"), "-08:00")
+        self.assertEqual(
+            self.db_operations._prepare_tzname_delta("America/New_York"),
+            "America/New_York",
+        )
+        self.assertEqual(self.db_operations._prepare_tzname_delta("UTC"), "UTC")
+        self.assertEqual(self.db_operations._prepare_tzname_delta(""), "UTC")
+        self.assertEqual(self.db_operations._prepare_tzname_delta(None), "UTC")
+
+    def test_extract_lookup_type_validation(self):
+        valid_lookups = ["year", "month", "day", "hour", "minute", "second", "isoweek"]
+        for lookup_type in valid_lookups:
+            self.db_operations._validate_lookup_type(lookup_type)
+
+        invalid_lookups = [
+            "year' OR 1=1--",
+            "day; DROP TABLE students;",
+            "foo bar",
+            "",
+            "year\x00",
+            None,
+            123,
+        ]
+        for lookup_type in invalid_lookups:
+            with self.assertRaises(ValueError):
+                self.db_operations._validate_lookup_type(lookup_type)
+            with self.assertRaises(ValueError):
+                self.db_operations.date_extract_sql(lookup_type, "field")
+            with self.assertRaises(ValueError):
+                self.db_operations.datetime_extract_sql(
+                    lookup_type, "field", None, "UTC"
+                )
+            with self.assertRaises(ValueError):
+                self.db_operations.time_extract_sql(lookup_type, "field")
+
+    def test_trunc_lookup_type_validation(self):
+        valid_lookups = [
+            "year",
+            "quarter",
+            "month",
+            "week",
+            "day",
+            "hour",
+            "minute",
+            "second",
+        ]
+        for lookup_type in valid_lookups:
+            self.db_operations._validate_trunc_lookup_type(lookup_type)
+
+        invalid_lookups = [
+            "year' OR 1=1--",
+            "day; DROP TABLE students;",
+            "foo bar",
+            "",
+            "year\x00",
+            None,
+            123,
+        ]
+        for lookup_type in invalid_lookups:
+            with self.assertRaises(OperationalError):
+                self.db_operations._validate_trunc_lookup_type(lookup_type)
+            with self.assertRaises(OperationalError):
+                self.db_operations.date_trunc_sql(lookup_type, "field", None)
+            with self.assertRaises(OperationalError):
+                self.db_operations.datetime_trunc_sql(
+                    lookup_type, "field", None, tzname="UTC"
+                )
+            with self.assertRaises(OperationalError):
+                self.db_operations.time_trunc_sql(lookup_type, "field", None)
+
+    def test_timezone_methods_preserve_existing_params(self):
+        existing_params = ["param1", "param2"]
+
+        sql, params = self.db_operations.datetime_extract_sql(
+            "hour", "field", existing_params, "UTC"
+        )
+        self.assertEqual(params, ("param1", "param2", "UTC"))
+
+        sql, params = self.db_operations.datetime_trunc_sql(
+            "hour", "field", existing_params, tzname="UTC"
+        )
+        self.assertEqual(params, ("param1", "param2", "UTC"))
+
+        sql, params = self.db_operations.time_trunc_sql(
+            "hour", "field", existing_params, tzname="UTC"
+        )
+        self.assertEqual(params, ("param1", "param2", "UTC"))
+
+        sql, params = self.db_operations.datetime_cast_date_sql(
+            "field", existing_params, "UTC"
+        )
+        self.assertEqual(params, ("param1", "param2", "UTC"))
+
+        sql, params = self.db_operations.datetime_cast_time_sql(
+            "field", existing_params, "UTC"
+        )
+        self.assertEqual(params, ("param1", "param2", "UTC"))
+
+    def test_tzname_sql_injection_defense(self):
+        settings.USE_TZ = True
+        malicious_tzname = "UTC'; DROP TABLE students; --"
+
+        sql, params = self.db_operations.datetime_extract_sql(
+            "hour", "field", None, malicious_tzname
+        )
+        self.assertNotIn(malicious_tzname, sql)
+        self.assertIn("%s", sql)
+        self.assertEqual(params, (malicious_tzname,))
+
+        sql, params = self.db_operations.datetime_trunc_sql(
+            "hour", "field", None, tzname=malicious_tzname
+        )
+        self.assertNotIn(malicious_tzname, sql)
+        self.assertIn("%s", sql)
+        self.assertEqual(params, (malicious_tzname,))
+
+        sql, params = self.db_operations.time_trunc_sql(
+            "hour", "field", None, tzname=malicious_tzname
+        )
+        self.assertNotIn(malicious_tzname, sql)
+        self.assertIn("%s", sql)
+        self.assertEqual(params, (malicious_tzname,))
+
+        sql, params = self.db_operations.datetime_cast_date_sql(
+            "field", None, malicious_tzname
+        )
+        self.assertNotIn(malicious_tzname, sql)
+        self.assertIn("%s", sql)
+        self.assertEqual(params, (malicious_tzname,))
+
+        sql, params = self.db_operations.datetime_cast_time_sql(
+            "field", None, malicious_tzname
+        )
+        self.assertNotIn(malicious_tzname, sql)
+        self.assertIn("%s", sql)
+        self.assertEqual(params, (malicious_tzname,))