Merge remote-tracking branch 'DragonQ/intelligent_octopus' into Yanso…

…n-intelligent_octopus

Dockerfile

@@ -15,11 +15,12 @@ ENV POETRY_VIRTUALENVS_CREATE=false \
 RUN mkdir -p "/opt/$APP_NAME"
 WORKDIR "/opt/$APP_NAME"
 
+RUN python -m pip install --no-cache-dir --upgrade pip poetry wheel
+
 COPY poetry.lock pyproject.toml ./
 COPY app app
 
-RUN python -m pip install --no-cache-dir --upgrade pip poetry wheel && \
-    poetry install --only main --no-root
+RUN poetry install --only main --no-root
 
 ######### TEST Image #########
 FROM base AS test

app/octopus_graphql_api_client.py

@@ -1,11 +1,19 @@
 import json
 import aiohttp
 import datetime as dt
+import re
+from typing import Any
 from intelligent_dispatches import IntelligentDispatchItem, IntelligentDispatches
 from contextlib import suppress
-import ciso8601
 import click
 
+DATETIME_RE = re.compile(
+    r"(?P<year>\d{4})-(?P<month>\d{1,2})-(?P<day>\d{1,2})"
+    r"[T ](?P<hour>\d{1,2}):(?P<minute>\d{1,2})"
+    r"(?::(?P<second>\d{1,2})(?:\.(?P<microsecond>\d{1,6})\d{0,6})?)?"
+    r"(?P<tzinfo>Z|[+-]\d{2}(?::?\d{2})?)?$"
+)
+
 intelligent_dispatches_query = '''query {{
   plannedDispatches(accountNumber: "{account_id}") {{
     startDt
@@ -36,7 +44,7 @@ def parse_datetime(dt_str: str) -> dt.datetime | None:
   Returns None if the input isn't well formatted.
   """
   with suppress(ValueError, IndexError):
-      return ciso8601.parse_datetime(dt_str)
+      return dt.datetime.fromisoformat(dt_str)
 
   if not (match := DATETIME_RE.match(dt_str)):
     return None
@@ -47,7 +55,7 @@ def parse_datetime(dt_str: str) -> dt.datetime | None:
 
   tzinfo: dt.tzinfo | None = None
   if tzinfo_str == "Z":
-    tzinfo = UTC
+    tzinfo = dt.UTC
   elif tzinfo_str is not None:
     offset_mins = int(tzinfo_str[-2:]) if len(tzinfo_str) > 3 else 0
     offset_hours = int(tzinfo_str[1:3])
@@ -59,18 +67,6 @@ def parse_datetime(dt_str: str) -> dt.datetime | None:
   kws["tzinfo"] = tzinfo
   return dt.datetime(**kws)
 
-def as_utc(dattim: dt.datetime) -> dt.datetime:
-  """Return a datetime as UTC time.
-
-  Assumes datetime without tzinfo to be in the DEFAULT_TIME_ZONE.
-  """
-  if dattim.tzinfo == UTC:
-    return dattim
-  if dattim.tzinfo is None:
-    dattim = dattim.replace(tzinfo=DEFAULT_TIME_ZONE)
-
-  return dattim.astimezone(UTC)
-
 class OctopusGraphQlApiClient:
   def __init__(self, api_key, timeout_in_seconds = 15):
     if (api_key is None):