Merge pull request #749 from robsdedude/bolt-utc-datetimes-4.4

UTC encoding patch for Bolt 4.4 and 4.3

Author: robsdedude

neo4j/data.py

@@ -34,6 +34,7 @@
     hydrate_date, dehydrate_date,
     hydrate_time, dehydrate_time,
     hydrate_datetime, dehydrate_datetime,
+    hydrate_datetime_v2, dehydrate_datetime_v2,
     hydrate_duration, dehydrate_duration, dehydrate_timedelta,
 )
 
@@ -268,7 +269,7 @@ def transform(self, x):
 class DataHydrator:
     # TODO: extend DataTransformer
 
-    def __init__(self):
+    def __init__(self, patch_utc=False):
         super(DataHydrator, self).__init__()
         self.graph = Graph()
         self.graph_hydrator = Graph.Hydrator(self.graph)
@@ -282,11 +283,19 @@ def __init__(self):
             b"D": hydrate_date,
             b"T": hydrate_time,         # time zone offset
             b"t": hydrate_time,         # no time zone
-            b"F": hydrate_datetime,     # time zone offset
-            b"f": hydrate_datetime,     # time zone name
             b"d": hydrate_datetime,     # no time zone
             b"E": hydrate_duration,
         }
+        if not patch_utc:
+            self.hydration_functions.update({
+                b"F": hydrate_datetime,     # time zone offset
+                b"f": hydrate_datetime,     # time zone name
+            })
+        else:
+            self.hydration_functions.update({
+                b"I": hydrate_datetime_v2,     # time zone offset
+                b"i": hydrate_datetime_v2,     # time zone name
+            })
 
     def hydrate(self, values):
         """ Convert PackStream values into native values.
@@ -320,10 +329,10 @@ class DataDehydrator:
     # TODO: extend DataTransformer
 
     @classmethod
-    def fix_parameters(cls, parameters):
+    def fix_parameters(cls, parameters, patch_utc=False):
         if not parameters:
             return {}
-        dehydrator = cls()
+        dehydrator = cls(patch_utc=patch_utc)
         try:
             dehydrated, = dehydrator.dehydrate([parameters])
         except TypeError as error:
@@ -332,19 +341,28 @@ def fix_parameters(cls, parameters):
         else:
             return dehydrated
 
-    def __init__(self):
+    def __init__(self, patch_utc=False):
         self.dehydration_functions = {}
         self.dehydration_functions.update({
             Point: dehydrate_point,
             Date: dehydrate_date,
             date: dehydrate_date,
             Time: dehydrate_time,
             time: dehydrate_time,
-            DateTime: dehydrate_datetime,
-            datetime: dehydrate_datetime,
             Duration: dehydrate_duration,
             timedelta: dehydrate_timedelta,
         })
+        if not patch_utc:
+            self.dehydration_functions.update({
+                DateTime: dehydrate_datetime,
+                datetime: dehydrate_datetime,
+            })
+        else:
+            self.dehydration_functions.update({
+                DateTime: dehydrate_datetime_v2,
+                datetime: dehydrate_datetime_v2,
+            })
+
         # Allow dehydration from any direct Point subclass
         self.dehydration_functions.update({cls: dehydrate_point for cls in Point.__subclasses__()})
 

neo4j/io/__init__.py

@@ -147,6 +147,8 @@ def __init__(self, unresolved_address, sock, max_connection_lifetime, *, auth=No
         # configuration hint that exists. Therefore, all hints can be stored at
         # connection level. This might change in the future.
         self.configuration_hints = {}
+        # back ported protocol patches negotiated with the server
+        self.bolt_patches = set()
         self.outbox = Outbox()
         self.inbox = Inbox(self.socket, on_error=self._set_defunct_read)
         self.packer = Packer(self.outbox)

neo4j/io/_bolt4.py

@@ -368,6 +368,20 @@ class Bolt4x3(Bolt4x2):
 
     PROTOCOL_VERSION = Version(4, 3)
 
+    def get_base_headers(self):
+        """ Bolt 4.1 passes the routing context, originally taken from
+        the URI, into the connection initialisation message. This
+        enables server-side routing to propagate the same behaviour
+        through its driver.
+        """
+        headers = {
+            "user_agent": self.user_agent,
+            "patch_bolt": ["utc"]
+        }
+        if self.routing_context is not None:
+            headers["routing"] = self.routing_context
+        return headers
+
     def route(self, database=None, imp_user=None, bookmarks=None):
         if imp_user is not None:
             raise ConfigurationError(
@@ -394,6 +408,7 @@ def route(self, database=None, imp_user=None, bookmarks=None):
 
     def hello(self):
         def on_success(metadata):
+            # configuration hints
             self.configuration_hints.update(metadata.pop("hints", {}))
             self.server_info.update(metadata)
             if "connection.recv_timeout_seconds" in self.configuration_hints:
@@ -407,6 +422,8 @@ def on_success(metadata):
                              "connection.recv_timeout_seconds (%r). Make sure "
                              "the server and network is set up correctly.",
                              self.local_port, recv_timeout)
+            # bolt patch handshake
+            self.bolt_patches.update(set(metadata.pop("patch_bolt", ())))
 
         headers = self.get_base_headers()
         headers.update(self.auth_dict)

neo4j/time/hydration.py

@@ -132,6 +132,34 @@ def hydrate_datetime(seconds, nanoseconds, tz=None):
     return zone.localize(t)
 
 
+def hydrate_datetime_v2(seconds, nanoseconds, tz=None):
+    """ Hydrator for `DateTime` and `LocalDateTime` values.
+
+    :param seconds:
+    :param nanoseconds:
+    :param tz:
+    :return: datetime
+    """
+    import pytz
+
+    minutes, seconds = map(int, divmod(seconds, 60))
+    hours, minutes = map(int, divmod(minutes, 60))
+    days, hours = map(int, divmod(hours, 24))
+    t = DateTime.combine(
+        Date.from_ordinal(get_date_unix_epoch_ordinal() + days),
+        Time(hours, minutes, seconds, nanoseconds)
+    )
+    if tz is None:
+        return t
+    if isinstance(tz, int):
+        tz_offset_minutes, tz_offset_seconds = divmod(tz, 60)
+        zone = pytz.FixedOffset(tz_offset_minutes)
+    else:
+        zone = pytz.timezone(tz)
+    t = t.replace(tzinfo=pytz.UTC)
+    return t.as_timezone(zone)
+
+
 def dehydrate_datetime(value):
     """ Dehydrator for `datetime` values.
 
@@ -167,8 +195,57 @@ def seconds_and_nanoseconds(dt):
     else:
         # with time offset
         seconds, nanoseconds = seconds_and_nanoseconds(value)
-        return Structure(b"F", seconds, nanoseconds,
-                         int(tz.utcoffset(value).total_seconds()))
+        offset = tz.utcoffset(value)
+        if offset.microseconds:
+            raise ValueError("Bolt protocol does not support sub-second "
+                             "UTC offsets.")
+        offset_seconds = offset.days * 86400 + offset.seconds
+        return Structure(b"F", seconds, nanoseconds, offset_seconds)
+
+
+def dehydrate_datetime_v2(value):
+    """ Dehydrator for `datetime` values.
+
+    :param value:
+    :type value: datetime
+    :return:
+    """
+
+    import pytz
+
+    def seconds_and_nanoseconds(dt):
+        if isinstance(dt, datetime):
+            dt = DateTime.from_native(dt)
+        dt = dt.astimezone(pytz.UTC)
+        utc_epoch = DateTime(1970, 1, 1, tzinfo=pytz.UTC)
+        dt_clock_time = dt.to_clock_time()
+        utc_epoch_clock_time = utc_epoch.to_clock_time()
+        t = dt_clock_time - utc_epoch_clock_time
+        return t.seconds, t.nanoseconds
+
+    tz = value.tzinfo
+    if tz is None:
+        # without time zone
+        value = pytz.UTC.localize(value)
+        seconds, nanoseconds = seconds_and_nanoseconds(value)
+        return Structure(b"d", seconds, nanoseconds)
+    elif hasattr(tz, "zone") and tz.zone and isinstance(tz.zone, str):
+        # with named pytz time zone
+        seconds, nanoseconds = seconds_and_nanoseconds(value)
+        return Structure(b"i", seconds, nanoseconds, tz.zone)
+    elif hasattr(tz, "key") and tz.key and isinstance(tz.key, str):
+        # with named zoneinfo (Python 3.9+) time zone
+        seconds, nanoseconds = seconds_and_nanoseconds(value)
+        return Structure(b"i", seconds, nanoseconds, tz.key)
+    else:
+        # with time offset
+        seconds, nanoseconds = seconds_and_nanoseconds(value)
+        offset = tz.utcoffset(value)
+        if offset.microseconds:
+            raise ValueError("Bolt protocol does not support sub-second "
+                             "UTC offsets.")
+        offset_seconds = offset.days * 86400 + offset.seconds
+        return Structure(b"I", seconds, nanoseconds, offset_seconds)
 
 
 def hydrate_duration(months, days, seconds, nanoseconds):

neo4j/work/result.py

@@ -75,7 +75,10 @@ def _run(self, query, parameters, db, imp_user, access_mode, bookmarks,
         query_metadata = getattr(query, "metadata", None)
         query_timeout = getattr(query, "timeout", None)
 
-        parameters = DataDehydrator.fix_parameters(dict(parameters or {}, **kwparameters))
+        parameters = DataDehydrator.fix_parameters(
+            dict(parameters or {}, **kwparameters),
+            patch_utc="utc" in self._connection.bolt_patches
+        )
 
         self._metadata = {
             "query": query_text,

neo4j/work/simple.py

@@ -206,7 +206,7 @@ def run(self, query, parameters=None, **kwparameters):
         protocol_version = cx.PROTOCOL_VERSION
         server_info = cx.server_info
 
-        hydrant = DataHydrator()
+        hydrant = DataHydrator(patch_utc="utc" in cx.bolt_patches)
 
         self._autoResult = Result(
             cx, hydrant, self._config.fetch_size, self._result_closed,

neo4j/work/transaction.py

@@ -127,8 +127,12 @@ def run(self, query, parameters=None, **kwparameters):
             # have any qid to fetch in batches.
             self._results[-1]._buffer_all()
 
+        hydrant = DataHydrator(
+            patch_utc="utc" in self._connection.bolt_patches
+        )
+
         result = Result(
-            self._connection, DataHydrator(), self._fetch_size,
+            self._connection, hydrant, self._fetch_size,
             self._result_on_closed_handler,
             self._error_handler
         )

testkit/build.py

@@ -1,14 +1,20 @@
 """
-Executed in Go driver container.
+Executed in driver container.
 Responsible for building driver and test backend.
 """
+
+
 import subprocess
+import sys
 
 
 def run(args, env=None):
-    subprocess.run(args, universal_newlines=True, stderr=subprocess.STDOUT,
-                   check=True, env=env)
+    subprocess.run(args, universal_newlines=True, stdout=sys.stdout,
+                   stderr=sys.stderr, check=True, env=env)
 
 
 if __name__ == "__main__":
     run(["python", "setup.py", "build"])
+    run(["python", "-m", "pip", "install", "-U", "pip"])
+    run(["python", "-m", "pip", "install", "-Ur",
+         "testkitbackend/requirements.txt"])

testkitbackend/_driver_logger.py

@@ -0,0 +1,41 @@
+# Copyright (c) "Neo4j"
+# Neo4j Sweden AB [http://neo4j.com]
+#
+# This file is part of Neo4j.
+#
+# 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 io
+import logging
+import sys
+
+
+buffer_handler = logging.StreamHandler(io.StringIO())
+buffer_handler.setLevel(logging.DEBUG)
+
+handler = logging.StreamHandler(sys.stdout)
+handler.setLevel(logging.DEBUG)
+logging.getLogger("neo4j").addHandler(handler)
+logging.getLogger("neo4j").addHandler(buffer_handler)
+logging.getLogger("neo4j").setLevel(logging.DEBUG)
+
+log = logging.getLogger("testkitbackend")
+log.addHandler(handler)
+log.setLevel(logging.DEBUG)
+
+
+__all__ = [
+    "buffer_handler",
+    "log",
+]