Use timezone-aware datetimes everywhere (#237)

* Use library function

* Use timezone-aware datetimes everywhere

Author: NMertsch

prod-config.toml

@@ -31,7 +31,6 @@ pretix_cache_file = "pretix_cache.json"
 
 [program_notifications]
 # UTC offset in hours (e.g. 2 for CEST)
-timezone_offset = 2
 api_url = "https://static.europython.eu/programme/ep2025/releases/current/schedule.json"
 schedule_cache_file = "schedule_cache.json"
 livestream_url_file = "livestreams.toml"

src/europython_discord/program_notifications/cog.py

@@ -20,7 +20,6 @@ def __init__(self, bot: Client, config: ProgramNotificationsConfig) -> None:
         self.config = config
         self.program_connector = ProgramConnector(
             api_url=self.config.api_url,
-            timezone_offset=self.config.timezone_offset,
             cache_file=self.config.schedule_cache_file,
             simulated_start_time=self.config.simulated_start_time,
             fast_mode=self.config.fast_mode,

src/europython_discord/program_notifications/config.py

@@ -1,19 +1,17 @@
 from __future__ import annotations
 
 from collections.abc import Mapping
-from datetime import datetime
 from pathlib import Path
 
-from pydantic import BaseModel
+from pydantic import AwareDatetime, BaseModel
 
 
 class ProgramNotificationsConfig(BaseModel):
-    timezone_offset: int
     api_url: str
     schedule_cache_file: Path
     livestream_url_file: Path
     main_notification_channel_name: str
     rooms_to_channel_names: Mapping[str, str]
 
-    simulated_start_time: datetime | None = None
+    simulated_start_time: AwareDatetime | None = None
     fast_mode: bool = False

src/europython_discord/program_notifications/models.py

@@ -1,8 +1,8 @@
 from __future__ import annotations
 
-from datetime import date, datetime
+from datetime import date
 
-from pydantic import BaseModel
+from pydantic import AwareDatetime, BaseModel
 
 
 class DaySchedule(BaseModel):
@@ -25,7 +25,7 @@ class Break(BaseModel):
     title: str
     duration: int
     rooms: list[str]
-    start: datetime
+    start: AwareDatetime
 
 
 class Session(BaseModel):
@@ -41,7 +41,7 @@ class Session(BaseModel):
     level: str
     track: str | None
     rooms: list[str]
-    start: datetime
+    start: AwareDatetime
     website_url: str
     duration: int
 

src/europython_discord/program_notifications/program_connector.py

@@ -3,7 +3,7 @@
 import asyncio
 import json
 import logging
-from datetime import date, datetime, timedelta, timezone
+from datetime import UTC, date, datetime, timedelta
 from pathlib import Path
 
 import aiofiles
@@ -18,21 +18,19 @@ class ProgramConnector:
     def __init__(
         self,
         api_url: str,
-        timezone_offset: int,
         cache_file: Path,
         simulated_start_time: datetime | None = None,
         *,
         fast_mode: bool = False,
     ) -> None:
         self._api_url = api_url
-        self._timezone = timezone(timedelta(hours=timezone_offset))
         self._cache_file = cache_file
 
         # time travel parameters for testing
         self._simulated_start_time = simulated_start_time
         if self._simulated_start_time:
             self._time_multiplier = 60 if fast_mode else 1
-            self._real_start_time = datetime.now(tz=self._timezone)
+            self._real_start_time = datetime.now(tz=UTC)
 
         self._fetch_lock = asyncio.Lock()
         self.sessions_by_day: dict[date, list[Session]] | None = None
@@ -102,11 +100,11 @@ async def _get_schedule_from_cache(self) -> dict[date, list[Session]] | None:
     async def _get_now(self) -> datetime:
         """Get the current time in the conference timezone."""
         if self._simulated_start_time:
-            elapsed = datetime.now(tz=self._timezone) - self._real_start_time
+            elapsed = datetime.now(tz=UTC) - self._real_start_time
             simulated_now = self._simulated_start_time + elapsed * self._time_multiplier
-            return simulated_now.astimezone(self._timezone)
+            return simulated_now.astimezone(UTC)
 
-        return datetime.now(tz=self._timezone)
+        return datetime.now(tz=UTC)
 
     async def get_sessions_by_date(self, date_now: date) -> list[Session]:
         if self.sessions_by_day is None:

src/europython_discord/program_notifications/session_to_embed.py

@@ -6,6 +6,7 @@
 from typing import Final
 
 from discord import Embed
+from discord.utils import format_dt
 
 from europython_discord.program_notifications.models import Session, Speaker
 
@@ -34,7 +35,7 @@ def create_session_embed(session: Session, livestream_url: str | None) -> Embed:
         color=_get_color(session.level),
     )
 
-    embed.add_field(name="Start Time", value=_format_start_time(session.start), inline=True)
+    embed.add_field(name="Start Time", value=format_dt(session.start), inline=True)
     embed.add_field(name="Room", value=_format_room(session.rooms), inline=True)
     embed.add_field(name="Track", value=_format_track(session.track), inline=True)
     embed.add_field(name="Duration", value=_format_duration(session.duration), inline=True)
@@ -91,16 +92,6 @@ def _format_title(title: str) -> str:
     return textwrap.shorten(title, width=_TITLE_WIDTH)
 
 
-def _format_start_time(start_time: datetime) -> str:
-    """Format the start time to a Discord timestamp string.
-
-    :param start_time: The start time
-    :return: A start time value for the embed.
-    """
-    start_time_timestamp = int(start_time.timestamp())
-    return f"<t:{start_time_timestamp}:f>"
-
-
 def _format_footer(start_time: datetime) -> str:
     """Create a footer with the local conference time.
 

src/europython_discord/registration/pretix_connector.py

@@ -5,7 +5,7 @@
 import logging
 import time
 from collections import defaultdict
-from datetime import datetime, timedelta, timezone
+from datetime import UTC, datetime, timedelta
 from pathlib import Path
 
 import aiofiles
@@ -57,7 +57,7 @@ async def fetch_pretix_data(self) -> None:
         # if called during an ongoing fetch, the caller waits until the fetch is done...
         async with self._fetch_lock:
             # ... but does not trigger a second fetch
-            now = datetime.now(tz=timezone.utc)
+            now = datetime.now(tz=UTC)
             if self._last_fetch and now - self._last_fetch < timedelta(minutes=2):
                 _logger.info(f"Skipping pretix fetch (last fetch was at {self._last_fetch})")
                 return

test-config.toml

@@ -31,7 +31,6 @@ pretix_cache_file = "pretix_cache.json"
 
 [program_notifications]
 # UTC offset in hours (e.g. 2 for CEST)
-timezone_offset = 2
 api_url = "https://static.europython.eu/programme/ep2025/releases/current/schedule.json"
 schedule_cache_file = "schedule_cache.json"
 livestream_url_file = "test-livestreams.toml"

tests/program_notifications/test_program_connector.py

@@ -27,9 +27,7 @@ def cache_file(tmp_path):
 
 @pytest.fixture
 async def program_connector(cache_file):
-    return ProgramConnector(
-        api_url="http://test.api/schedule", timezone_offset=0, cache_file=cache_file
-    )
+    return ProgramConnector(api_url="http://test.api/schedule", cache_file=cache_file)
 
 
 @pytest.fixture

tests/program_notifications/test_session_to_embed.py

@@ -1,4 +1,4 @@
-from datetime import UTC, datetime
+from datetime import datetime, timedelta, timezone
 
 import pytest
 
@@ -126,10 +126,12 @@ def test_embed_color(session: Session, level: str, expected_color: int) -> None:
 
 def test_embed_fields_start_time(session: Session) -> None:
     """Test the 'Start Time' field of the embed."""
+    start_time = datetime(2024, 1, 2, 3, 4, 5, tzinfo=timezone(timedelta(hours=6)))
+    session = session.model_copy(update={"start": start_time})
+
     embed = session_to_embed.create_session_embed(session, None)
     assert embed.fields[0].name == "Start Time"
-    assert embed.fields[0].value.startswith("<t:")
-    assert embed.fields[0].value.endswith(":f>")
+    assert embed.fields[0].value == f"<t:{int(start_time.timestamp())}>"
 
 
 def test_embed_fields_room(session: Session) -> None:
@@ -252,19 +254,6 @@ def test_embed_footer(session: Session) -> None:
     assert embed.footer.text == "This session starts at 08:00:00 (local conference time)"
 
 
-def test_format_start_time(session: Session) -> None:
-    """Test the _format_start_time function."""
-    formatted_start_time = session_to_embed._format_start_time(session.start)
-    assert formatted_start_time.startswith("<t:")
-    assert formatted_start_time.endswith(":f>")
-
-    # The following code assumes that the start time in the mock data is in UTC.
-    datetime_obj = datetime.fromtimestamp(
-        int(formatted_start_time.replace("<t:", "").replace(":f>", "")), tz=UTC
-    )
-    assert datetime_obj == session.start
-
-
 def test_format_duration(session: Session) -> None:
     """Test the _format_duration function."""
     formatted_duration = session_to_embed._format_duration(session.duration)