Fix handling of None as parameter for a datetimeoffset column

Fixed issue where the :class:`.mssql.DATETIMEOFFSET` type would not
accommodate for the ``None`` value, introduced as part of the series of
fixes for this type first introduced in :ticket:`4983`, :ticket:`5045`.
Additionally, added support for passing a backend-specific date formatted
string through this type, as is typically allowed for date/time types on
most other DBAPIs.

Fixes: #5132
Change-Id: Iab05d67382e0f550474d50e0c3c1c888521b678a
(cherry picked from commit 532ddb0fa717bc10c3a95b7ddf3a60bbfad7ac56)

diff --git a/doc/build/changelog/unreleased_13/5132.rst b/doc/build/changelog/unreleased_13/5132.rst
new file mode 100644 (file)
index 0000000..d4e68f2
--- /dev/null
+++ b/doc/build/changelog/unreleased_13/5132.rst
@@ -0,0 +1,10 @@
+.. change::
+    :tags: bug, mssql
+    :tickets: 5132
+
+    Fixed issue where the :class:`.mssql.DATETIMEOFFSET` type would not
+    accommodate for the ``None`` value, introduced as part of the series of
+    fixes for this type first introduced in :ticket:`4983`, :ticket:`5045`.
+    Additionally, added support for passing a backend-specific date formatted
+    string through this type, as is typically allowed for date/time types on
+    most other DBAPIs.

diff --git a/lib/sqlalchemy/dialects/mssql/pyodbc.py b/lib/sqlalchemy/dialects/mssql/pyodbc.py
index 879fe9a2ae53c679bf8c40a5319c0c7ac5c3f62c..4ba3a0dfac198c7e27de6ca56ac72e80b4cf91af 100644 (file)
--- a/lib/sqlalchemy/dialects/mssql/pyodbc.py
+++ b/lib/sqlalchemy/dialects/mssql/pyodbc.py
@@ -234,10 +234,21 @@ class _ms_binary_pyodbc(object):
 class _ODBCDateTimeOffset(DATETIMEOFFSET):
     def bind_processor(self, dialect):
         def process(value):
-            """Convert to string format required by T-SQL."""
-            dto_string = value.strftime("%Y-%m-%d %H:%M:%S.%f %z")
-            # offset needs a colon, e.g., -0700 -> -07:00
-            return dto_string[:30] + ":" + dto_string[30:]
+            if value is None:
+                return None
+            elif isinstance(value, util.string_types):
+                # if a string was passed directly, allow it through
+                return value
+            else:
+                # Convert to string format required by T-SQL
+                dto_string = value.strftime("%Y-%m-%d %H:%M:%S.%f %z")
+                # offset needs a colon, e.g., -0700 -> -07:00
+                # "UTC offset in the form (+-)HHMM[SS[.ffffff]]"
+                # backend currently rejects seconds / fractional seconds
+                dto_string = re.sub(
+                    r"([\+\-]\d{2})([\d\.]+)$", r"\1:\2", dto_string
+                )
+                return dto_string
 
         return process
 

diff --git a/test/dialect/mssql/test_types.py b/test/dialect/mssql/test_types.py
index 3cf9d9036dd91bd1e14a9f03ff713ae2fc679ac1..d7fbccac268bd82299d1da5a0e34aaaa14a5fba9 100644 (file)
--- a/test/dialect/mssql/test_types.py
+++ b/test/dialect/mssql/test_types.py
@@ -41,6 +41,7 @@ from sqlalchemy.dialects.mssql.base import MS_2005_VERSION
 from sqlalchemy.dialects.mssql.base import MS_2008_VERSION
 from sqlalchemy.dialects.mssql.base import TIME
 from sqlalchemy.sql import sqltypes
+from sqlalchemy.testing import assert_raises
 from sqlalchemy.testing import assert_raises_message
 from sqlalchemy.testing import AssertsCompiledSQL
 from sqlalchemy.testing import AssertsExecutionResults
@@ -709,69 +710,152 @@ class TypeRoundTripTest(
         for col in reflected_dates.c:
             self.assert_types_base(col, dates_table.c[col.key])
 
-    def test_date_roundtrip(self):
+    @testing.metadata_fixture()
+    def date_fixture(self, metadata):
         t = Table(
             "test_dates",
             metadata,
-            Column(
-                "id",
-                Integer,
-                Sequence("datetest_id_seq", optional=True),
-                primary_key=True,
-            ),
             Column("adate", Date),
-            Column("atime", Time),
+            Column("atime1", Time),
+            Column("atime2", Time),
             Column("adatetime", DateTime),
             Column("adatetimeoffset", DATETIMEOFFSET),
         )
-        metadata.create_all()
+
         d1 = datetime.date(2007, 10, 30)
         t1 = datetime.time(11, 2, 32)
         d2 = datetime.datetime(2007, 10, 30, 11, 2, 32)
-        dto = datetime.datetime(
-            2007,
-            10,
-            30,
-            11,
-            2,
-            32,
-            123456,
-            util.timezone(datetime.timedelta(hours=1)),
-        )
-        t.insert().execute(
-            adate=d1, adatetime=d2, atime=t1, adatetimeoffset=dto
-        )
-
-        # NOTE: this previously passed 'd2' for "adate" even though
-        # "adate" is a date column; we asserted that it truncated w/o issue.
-        # As of pyodbc 4.0.22, this is no longer accepted, was accepted
-        # in 4.0.21.  See also the new pyodbc assertions regarding numeric
-        # precision.
-        t.insert().execute(
-            adate=d1, adatetime=d2, atime=d2, adatetimeoffset=dto
-        )
+        return t, (d1, t1, d2)
 
-        x = t.select().execute().fetchall()[0]
-        self.assert_(x.adate.__class__ == datetime.date)
-        self.assert_(x.atime.__class__ == datetime.time)
-        self.assert_(x.adatetime.__class__ == datetime.datetime)
-        self.assert_(x.adatetimeoffset.__class__ == datetime.datetime)
+    def test_date_roundtrips(self, date_fixture):
+        t, (d1, t1, d2) = date_fixture
+        with testing.db.begin() as conn:
+            conn.execute(
+                t.insert(), adate=d1, adatetime=d2, atime1=t1, atime2=d2
+            )
 
-        t.delete().execute()
+            row = conn.execute(t.select()).first()
+            eq_(
+                (row.adate, row.adatetime, row.atime1, row.atime2),
+                (d1, d2, t1, d2.time()),
+            )
 
-        t.insert().execute(
-            adate=d1, adatetime=d2, atime=t1, adatetimeoffset=dto
+    @testing.metadata_fixture()
+    def datetimeoffset_fixture(self, metadata):
+        t = Table(
+            "test_dates", metadata, Column("adatetimeoffset", DATETIMEOFFSET),
         )
 
-        eq_(
-            select(
-                [t.c.adate, t.c.atime, t.c.adatetime, t.c.adatetimeoffset],
-                t.c.adate == d1,
+        return t
+
+    @testing.combinations(
+        ("dto_param_none", None, None, False),
+        (
+            "dto_param_datetime_aware_positive",
+            datetime.datetime(
+                2007,
+                10,
+                30,
+                11,
+                2,
+                32,
+                123456,
+                util.timezone(datetime.timedelta(hours=1)),
+            ),
+            1,
+            False,
+        ),
+        (
+            "dto_param_datetime_aware_negative",
+            datetime.datetime(
+                2007,
+                10,
+                30,
+                11,
+                2,
+                32,
+                123456,
+                util.timezone(datetime.timedelta(hours=-5)),
+            ),
+            -5,
+            False,
+        ),
+        (
+            "dto_param_datetime_aware_seconds_frac_fail",
+            datetime.datetime(
+                2007,
+                10,
+                30,
+                11,
+                2,
+                32,
+                123456,
+                util.timezone(datetime.timedelta(seconds=4000)),
+            ),
+            None,
+            True,
+            testing.requires.python3,
+        ),
+        (
+            "dto_param_datetime_naive",
+            datetime.datetime(2007, 10, 30, 11, 2, 32, 123456),
+            0,
+            False,
+        ),
+        (
+            "dto_param_string_one",
+            "2007-10-30 11:02:32.123456 +01:00",
+            1,
+            False,
+        ),
+        # wow
+        ("dto_param_string_two", "October 30, 2007 11:02:32.123456", 0, False),
+        ("dto_param_string_invalid", "this is not a date", 0, True),
+        id_="iaaa",
+        argnames="dto_param_value, expected_offset_hours, should_fail",
+    )
+    def test_datetime_offset(
+        self,
+        datetimeoffset_fixture,
+        dto_param_value,
+        expected_offset_hours,
+        should_fail,
+    ):
+        t = datetimeoffset_fixture
+        with testing.db.begin() as conn:
+            if should_fail:
+                assert_raises(
+                    sa.exc.DBAPIError,
+                    conn.execute,
+                    t.insert(),
+                    adatetimeoffset=dto_param_value,
+                )
+                return
+
+            conn.execute(
+                t.insert(), adatetimeoffset=dto_param_value,
             )
-            .execute()
-            .fetchall(),
-            [(d1, t1, d2, dto)],
-        )
+
+            row = conn.execute(t.select()).first()
+
+            if dto_param_value is None:
+                is_(row.adatetimeoffset, None)
+            else:
+                eq_(
+                    row.adatetimeoffset,
+                    datetime.datetime(
+                        2007,
+                        10,
+                        30,
+                        11,
+                        2,
+                        32,
+                        123456,
+                        util.timezone(
+                            datetime.timedelta(hours=expected_offset_hours)
+                        ),
+                    ),
+                )
 
     @emits_warning_on("mssql+mxodbc", r".*does not have any indexes.*")
     @testing.provide_metadata