gitea_admin/sientia-dataops-orchestrator_temporal
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix: satisfy mypy in put_last_data_timestamp after utc=True normalization

Browse Source 
CI mypy caught what my local environment couldn't (Python 3.14 has no
compatible numpy stub for the project's python_version=3.11 mypy target,
so mypy couldn't run locally at all):

  slot_manager.py:353: error: Incompatible types in assignment
    (expression has type "str", variable has type "Timestamp")
  slot_manager.py:368: error: Incompatible return value type
    (got "Timestamp", expected "str | None")

Reassigning `last_data_timestamp` from a Timestamp to its strftime() string
trips mypy's inference (it fixes an unannotated local's type from its first
assignment). Split into a separate `last_data_timestamp_str` for the string
form instead of reusing the name.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Eduardo Rios
2026-09-15 08:18:17 -03:00
parent 14b854148a
commit ed666c260d
1 changed files with 8 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
orchestrator/activities/slot_manager.py
View File

@@ -349,11 +349,15 @@ class SlotManager(SientiaMonitoring):
# The Redis repository's plain ``json.dumps`` cannot serialize a # The Redis repository's plain ``json.dumps`` cannot serialize a
# ``Timestamp``/``datetime`` directly, so store it as a string # ``Timestamp``/``datetime`` directly, so store it as a string
# formatted per ``DATETIME_FORMAT_MS_WITH_TZ`` instead. # formatted per ``DATETIME_FORMAT_MS_WITH_TZ`` instead. Kept as a
last_data_timestamp = last_data_timestamp.strftime(DATETIME_FORMAT_MS_WITH_TZ) # separate variable (rather than reassigning ``last_data_timestamp``)
# since it's a different type — mypy infers a local's type from its
# first assignment and flags a same-name reassignment to an
# incompatible type as an error.
last_data_timestamp_str = last_data_timestamp.strftime(DATETIME_FORMAT_MS_WITH_TZ)
try: try:
self.redis_repository.set(key, last_data_timestamp, ttl=60 * 60 * 5) self.redis_repository.set(key, last_data_timestamp_str, ttl=60 * 60 * 5)
except Exception as e: except Exception as e:
self.send_notification( self.send_notification(
metadata=metadata, metadata=metadata,
@@ -365,7 +369,7 @@ class SlotManager(SientiaMonitoring):
) )
raise e raise e
return last_data_timestamp return last_data_timestamp_str
@activity.defn(name='filter_notification_alerts') @activity.defn(name='filter_notification_alerts')
def filter_notification_alerts(self, input_data: dict[str, Any]) -> dict[str, Any]: def filter_notification_alerts(self, input_data: dict[str, Any]) -> dict[str, Any]: