datetime.datetime.utcnow() is deprecated, replace with a central utility (#658)

Signed-off-by: Will Rieger <[email protected]>

Author: wrieg123

csp/impl/pandas_accessor.py

@@ -11,6 +11,7 @@
 from csp.impl.pandas_ext_type import TsDtype, is_csp_type
 from csp.impl.struct import define_nested_struct
 from csp.impl.wiring.edge import Edge
+from csp.utils.datetime import utc_now
 
 T = TypeVar("T")
 
@@ -369,7 +370,7 @@ def snap(self, timeout: timedelta = timedelta(seconds=10), starttime: datetime =
         :returns: A new Series of type corresponding to the underlying edge type (TsDtype.subtype),
             with the same index as the original series, containing the snapped value of each Edge.
         """
-        starttime = starttime or datetime.utcnow()
+        starttime = starttime or utc_now()
         return self.run(starttime, timeout, True, tick_count=1, snap=True)
 
     def show_graph(self):
@@ -546,7 +547,7 @@ def snap(self, timeout: timedelta = timedelta(seconds=10), starttime: datetime =
             a time in the future (to schedule a snap at, i.e. the next minute)
         :returns: A new DataFrame with the same indices where every Edge is replaced by it's snapped value.
         """
-        starttime = starttime or datetime.utcnow()
+        starttime = starttime or utc_now()
         return self.run(starttime, timeout, True, tick_count=1, snap=True)
 
     def sample(self, trigger: Union[timedelta, np.timedelta64, pd.Timedelta, ts[object]], inplace: bool = False):

csp/impl/pandas_perspective.py

@@ -16,6 +16,7 @@
     perspective,
     perspective_type_map,
 )
+from csp.utils.datetime import utc_now
 
 _ = csp.impl.pandas_accessor
 
@@ -240,7 +241,7 @@ def run(self, starttime=None, endtime=timedelta(seconds=60), realtime=True, clea
         """Run a graph that sends data to the table on the current thread.
         Normally, this is only useful for debugging.
         """
-        starttime = starttime or datetime.utcnow()
+        starttime = starttime or utc_now()
         if clear:
             self.clear()
         csp.run(self.graph, starttime=starttime, endtime=endtime, realtime=realtime)
@@ -250,7 +251,7 @@ def start(self, starttime=None, endtime=timedelta(seconds=60), *, realtime=True,
         If clear=True, will clear any data from the data before writing to it.
         If auto_shutdown=True, will stop the engine if the table gets garbage collected.
         """
-        starttime = starttime or datetime.utcnow()
+        starttime = starttime or utc_now()
         if clear:
             self.clear()
         self._runner = csp.run_on_thread(

csp/tests/adapters/test_kafka.py

@@ -14,6 +14,7 @@
     RawBytesMessageMapper,
     RawTextMessageMapper,
 )
+from csp.utils.datetime import utc_now
 
 from .kafka_utils import _precreate_topic
 
@@ -117,14 +118,14 @@ def graph(count: int):
             csp.stop_engine(stop)
 
         count = 5
-        results = csp.run(graph, count, starttime=datetime.utcnow(), endtime=timedelta(seconds=30), realtime=True)
+        results = csp.run(graph, count, starttime=utc_now(), endtime=timedelta(seconds=30), realtime=True)
         assert len(results["sub_data"]) >= 5
         print(results)
         for result in results["sub_data"]:
             assert result[1].mapped_partition >= 0
             assert result[1].mapped_offset >= 0
             assert result[1].mapped_live is not None
-            assert result[1].mapped_timestamp < datetime.utcnow()
+            assert result[1].mapped_timestamp < utc_now()
         assert results["sub_data"][-1][1].mapped_live
 
     @pytest.mark.skipif(not os.environ.get("CSP_TEST_KAFKA"), reason="Skipping kafka adapter tests")
@@ -204,9 +205,7 @@ def graph(symbols: list, count: int):
 
         symbols = ["AAPL", "MSFT"]
         count = 100
-        results = csp.run(
-            graph, symbols, count, starttime=datetime.utcnow(), endtime=timedelta(seconds=30), realtime=True
-        )
+        results = csp.run(graph, symbols, count, starttime=utc_now(), endtime=timedelta(seconds=30), realtime=True)
         for symbol in symbols:
             pub = results[f"pall_{symbol}"]
             sub = results[f"sall_{symbol}"]
@@ -231,7 +230,7 @@ def pub_graph():
             csp.stop_engine(stop)
             # csp.print('pub', struct)
 
-        csp.run(pub_graph, starttime=datetime.utcnow(), endtime=timedelta(seconds=30), realtime=True)
+        csp.run(pub_graph, starttime=utc_now(), endtime=timedelta(seconds=30), realtime=True)
 
         # grab start/end times
         def get_times_graph():
@@ -251,9 +250,7 @@ def get_times_graph():
             # csp.print('sub', data)
             # csp.print('status', kafkaadapter.status())
 
-        all_data = csp.run(get_times_graph, starttime=datetime.utcnow(), endtime=timedelta(seconds=30), realtime=True)[
-            "data"
-        ]
+        all_data = csp.run(get_times_graph, starttime=utc_now(), endtime=timedelta(seconds=30), realtime=True)["data"]
         min_time = all_data[0][1].dt
 
         def get_data(start_offset, expected_count):
@@ -276,7 +273,7 @@ def get_data(start_offset, expected_count):
             get_data,
             KafkaStartOffset.EARLIEST,
             10,
-            starttime=datetime.utcnow(),
+            starttime=utc_now(),
             endtime=timedelta(seconds=30),
             realtime=True,
         )["data"]
@@ -288,7 +285,7 @@ def get_data(start_offset, expected_count):
             get_data,
             KafkaStartOffset.LATEST,
             1,
-            starttime=datetime.utcnow(),
+            starttime=utc_now(),
             endtime=timedelta(seconds=1),
             realtime=True,
         )["data"]
@@ -306,7 +303,7 @@ def get_data(start_offset, expected_count):
         stime = all_data[2][1].dt + timedelta(milliseconds=1)
         expected = [x for x in all_data if x[1].dt >= stime]
         res = csp.run(
-            get_data, stime, len(expected), starttime=datetime.utcnow(), endtime=timedelta(seconds=30), realtime=True
+            get_data, stime, len(expected), starttime=utc_now(), endtime=timedelta(seconds=30), realtime=True
         )["data"]
         assert len(res) == len(expected)
 
@@ -376,9 +373,7 @@ def graph(symbols: list, count: int):
 
         symbols = ["AAPL", "MSFT"]
         count = 10
-        results = csp.run(
-            graph, symbols, count, starttime=datetime.utcnow(), endtime=timedelta(seconds=30), realtime=True
-        )
+        results = csp.run(graph, symbols, count, starttime=utc_now(), endtime=timedelta(seconds=30), realtime=True)
         # print(results)
         for symbol in symbols:
             pub = results[f"pub_{symbol}"]
@@ -405,7 +400,7 @@ def graph_sub():
 
         # With bug this would deadlock
         with pytest.raises(RuntimeError):
-            csp.run(graph_sub, starttime=datetime.utcnow(), endtime=timedelta(seconds=2), realtime=True)
+            csp.run(graph_sub, starttime=utc_now(), endtime=timedelta(seconds=2), realtime=True)
         kafkaadapter2 = KafkaAdapterManager(**kafkaadapterkwargs)
 
         def graph_pub():
@@ -414,7 +409,7 @@ def graph_pub():
 
         # With bug this would deadlock
         with pytest.raises(RuntimeError):
-            csp.run(graph_pub, starttime=datetime.utcnow(), endtime=timedelta(seconds=2), realtime=True)
+            csp.run(graph_pub, starttime=utc_now(), endtime=timedelta(seconds=2), realtime=True)
 
     @pytest.mark.skipif(not os.environ.get("CSP_TEST_KAFKA"), reason="Skipping kafka adapter tests")
     def test_invalid_broker(self, kafkaadapterkwargs):
@@ -434,7 +429,7 @@ def graph_sub():
 
         # With bug this would deadlock
         with pytest.raises(RuntimeError):
-            csp.run(graph_sub, starttime=datetime.utcnow(), endtime=timedelta(seconds=2), realtime=True)
+            csp.run(graph_sub, starttime=utc_now(), endtime=timedelta(seconds=2), realtime=True)
 
         kafkaadapter2 = KafkaAdapterManager(**dict_with_broker)
 
@@ -444,7 +439,7 @@ def graph_pub():
 
         # With bug this would deadlock
         with pytest.raises(RuntimeError):
-            csp.run(graph_pub, starttime=datetime.utcnow(), endtime=timedelta(seconds=2), realtime=True)
+            csp.run(graph_pub, starttime=utc_now(), endtime=timedelta(seconds=2), realtime=True)
 
     @pytest.mark.skipif(not os.environ.get("CSP_TEST_KAFKA"), reason="Skipping kafka adapter tests")
     def test_meta_field_map_tick_timestamp_from_field(self, kafkaadapterkwargs):
@@ -464,7 +459,7 @@ def graph_sub():
             )
 
         with pytest.raises(ValueError):
-            csp.run(graph_sub, starttime=datetime.utcnow(), endtime=timedelta(seconds=2), realtime=True)
+            csp.run(graph_sub, starttime=utc_now(), endtime=timedelta(seconds=2), realtime=True)
 
     @pytest.mark.skipif(not os.environ.get("CSP_TEST_KAFKA"), reason="Skipping kafka adapter tests")
     def test_conf_options(self):
@@ -499,7 +494,7 @@ def pub_graph():
             stop = csp.filter(stop, stop)
             csp.stop_engine(stop)
 
-        csp.run(pub_graph, starttime=datetime.utcnow(), endtime=timedelta(seconds=5), realtime=True)
+        csp.run(pub_graph, starttime=utc_now(), endtime=timedelta(seconds=5), realtime=True)
 
         def sub_graph():
             kafkaadapter = KafkaAdapterManager(broker=kafkabroker, start_offset=KafkaStartOffset.EARLIEST)
@@ -526,7 +521,7 @@ def sub_graph():
             stop = csp.and_(*stop_flags)
             csp.stop_engine(csp.filter(stop, stop))
 
-        res = csp.run(sub_graph, starttime=datetime.utcnow(), endtime=timedelta(seconds=5), realtime=True)
+        res = csp.run(sub_graph, starttime=utc_now(), endtime=timedelta(seconds=5), realtime=True)
         burst = res["burst"]
         assert len(burst) == 1
         assert isinstance(burst[0][1], list)

csp/tests/adapters/test_parquet.py

@@ -14,6 +14,7 @@
 import csp
 from csp.adapters.output_adapters.parquet import ParquetOutputConfig
 from csp.adapters.parquet import ParquetReader, ParquetWriter
+from csp.utils.datetime import utc_now
 
 
 class PriceQuantity(csp.Struct):
@@ -855,7 +856,7 @@ def write_data(file_name: str, binary_arrow: bool) -> csp.ts[AllTypes]:
                 b=True,
                 i=123,
                 d=123.456,
-                dt=datetime.utcnow(),
+                dt=utc_now(),
                 dte=date.today(),
                 t=time(1, 2, 3),
                 s="hello hello",

csp/tests/adapters/test_perspective.py

@@ -2,6 +2,7 @@
 from datetime import date, datetime, timedelta
 
 import csp
+from csp.utils.datetime import utc_now
 
 try:
     from csp.adapters.perspective import PerspectiveAdapter
@@ -35,4 +36,4 @@ class TestPerspectiveAdapter(unittest.TestCase):
     @unittest.skipIf(not HAS_PERSPECTIVE, "Test requires perspective")
     def test_adapter(self):
         output = {}
-        csp.run(my_graph, output, starttime=datetime.utcnow(), endtime=timedelta(seconds=1))
+        csp.run(my_graph, output, starttime=utc_now(), endtime=timedelta(seconds=1))

csp/tests/adapters/test_status.py

@@ -7,6 +7,7 @@
 from csp import ts
 from csp.adapters.kafka import DateTimeType, JSONTextMessageMapper, KafkaStatusMessageType
 from csp.adapters.status import Level
+from csp.utils.datetime import utc_now
 
 from .kafka_utils import _precreate_topic
 
@@ -43,7 +44,7 @@ def graph():
             csp.stop_engine(done_flag)
 
         _precreate_topic(topic)
-        results = csp.run(graph, starttime=datetime.utcnow(), endtime=timedelta(seconds=10), realtime=True)
+        results = csp.run(graph, starttime=utc_now(), endtime=timedelta(seconds=10), realtime=True)
         status = results["status"][0][1]
         assert status.status_code == KafkaStatusMessageType.MSG_RECV_ERROR
         assert status.level == Level.ERROR

csp/tests/impl/test_dateframe.py

@@ -2,6 +2,7 @@
 from datetime import datetime, timedelta
 
 import csp
+from csp.utils.datetime import utc_now
 
 
 class TestCspDataFrame(unittest.TestCase):
@@ -137,7 +138,7 @@ def test_perspective(self):
         _ = df.to_perspective(starttime, endtime)
 
         # realtime
-        widget = df.to_perspective(datetime.utcnow(), endtime=timedelta(seconds=30), realtime=True)
+        widget = df.to_perspective(utc_now(), endtime=timedelta(seconds=30), realtime=True)
         import time
 
         time.sleep(1)

csp/tests/impl/test_enum.py

@@ -9,6 +9,7 @@
 
 import csp
 from csp import ts
+from csp.utils.datetime import utc_now
 
 
 class MyEnum(csp.Enum):
@@ -129,7 +130,7 @@ def graph():
             csp.add_graph_output("eout", eout)
             csp.add_graph_output("ein", ein)
 
-        result = csp.run(graph, starttime=datetime.utcnow())
+        result = csp.run(graph, starttime=utc_now())
         self.assertEqual([v[1] for v in result["eout"]], [MyEnum(3), MyEnum(20), MyEnum(1)])
         self.assertEqual([v[1] for v in result["ein"]], [3, 20, 1])
 

csp/tests/impl/test_genericpushadapter.py

@@ -4,6 +4,7 @@
 from datetime import datetime, timedelta
 
 import csp
+from csp.utils.datetime import utc_now
 
 
 class Driver:
@@ -48,7 +49,7 @@ def graph():
 
         res = csp.run(
             graph,
-            starttime=datetime.utcnow(),
+            starttime=utc_now(),
             endtime=timedelta(1),
             realtime=True,
             queue_wait_time=timedelta(seconds=0),

csp/tests/impl/test_outputadapter.py

@@ -13,6 +13,7 @@
 from csp.impl.outputadapter import OutputAdapter
 from csp.impl.pushadapter import PushInputAdapter
 from csp.impl.wiring import py_output_adapter_def, py_push_adapter_def
+from csp.utils.datetime import utc_now
 
 
 class MyData(csp.Struct):
@@ -178,13 +179,13 @@ def my_graph_with_manager():
 
 class TestPythonOutputAdapter(unittest.TestCase):
     def test_basic(self):
-        csp.run(my_graph_basic, starttime=datetime.utcnow(), endtime=timedelta(seconds=5), realtime=False)
+        csp.run(my_graph_basic, starttime=utc_now(), endtime=timedelta(seconds=5), realtime=False)
         self.assertEqual(
             output_buffer[0], '[{"a": 1, "b": 2, "c": 3}, {"a": 1, "b": 2, "c": 3}, {"a": 1, "b": 2, "c": 3}]'
         )
 
     def test_with_manager(self):
-        csp.run(my_graph_with_manager, starttime=datetime.utcnow(), endtime=timedelta(seconds=1), realtime=True)
+        csp.run(my_graph_with_manager, starttime=utc_now(), endtime=timedelta(seconds=1), realtime=True)
 
         # assert that the adapter manager put the right things in the right place,
         # e.g. data_1 and data_2 into data_1 output, and data_3 into data_3 output