Use timezone aware datetime objects and use local time in webapp

.github/workflows/pytest.yml

@@ -10,7 +10,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.8", "3.9", "3.10", "3.11"]
+        python-version: ["3.9", "3.10", "3.11"]
 
     steps:
     - uses: actions/checkout@v2

alembic/versions/43f8a46a10f4_add_login_token_expire_date.py

@@ -6,7 +6,7 @@
 
 """
 
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from alembic import op
 import sqlalchemy as sa
 
@@ -26,7 +26,7 @@ def upgrade():
     # To avoid having to reset all tokens, we set the default value to the current date + 30 days
     # when running this migration. This ensures users will continue to receive notifications with
     # their current login token.
-    expire = datetime.utcnow() + timedelta(days=30)
+    expire = datetime.now(timezone.utc) + timedelta(days=30)
     users = sa.sql.table("users", sa.sql.column("login_token_expire_date"))
     op.execute(users.update().values(login_token_expire_date=expire))
     op.alter_column("users", "login_token_expire_date", nullable=False)

app/api/login.py

@@ -1,4 +1,4 @@
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from fastapi import APIRouter, Depends, HTTPException, Response, status
 from fastapi.security import OAuth2PasswordRequestForm
 from fastapi.logger import logger
@@ -26,7 +26,7 @@ def login(
     if db_user is None:
         db_user = crud.create_user(db, form_data.username.lower())
         response.status_code = status.HTTP_201_CREATED
-    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
     access_token = utils.create_access_token(db_user.username, expire=expire)
     crud.update_user_login_token_expire_date(db, db_user, expire)
     return {"access_token": access_token, "token_type": "bearer"}

app/crud.py

@@ -240,7 +240,9 @@ def update_user_notifications(
 
 def delete_notifications(db: Session, keep_days: int) -> None:
     """Delete notifications older than X days"""
-    date_limit = datetime.datetime.utcnow() - datetime.timedelta(days=keep_days)
+    date_limit = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(
+        days=keep_days
+    )
     # First retrieve all notifications id to delete
     old_notification_ids = db.query(models.Notification.id).filter(
         models.Notification.timestamp < date_limit

app/models.py

@@ -1,4 +1,5 @@
 from __future__ import annotations
+import datetime
 import uuid
 from typing import List
 from sqlalchemy import (
@@ -12,14 +13,41 @@
     func,
 )
 from sqlalchemy.orm import relationship, backref, Session
-from datetime import datetime
 from sqlalchemy.types import TypeDecorator, CHAR
 from sqlalchemy.dialects.postgresql import UUID
 from sqlalchemy.ext.associationproxy import association_proxy
 from .database import Base
 from . import schemas
 
 
+def utcnow():
+    # datetime.utcnow() is deprecated in 3.12
+    # as it returns a naive datetime
+    return datetime.datetime.now(datetime.timezone.utc)
+
+
+class TZDateTime(TypeDecorator):
+    """DateTime with TimeZone
+
+    From https://docs.sqlalchemy.org/en/20/core/custom_types.html#store-timezone-aware-timestamps-as-timezone-naive-utc
+    """
+
+    impl = DateTime
+    cache_ok = True
+
+    def process_bind_param(self, value, dialect):
+        if value is not None:
+            if not value.tzinfo or value.tzinfo.utcoffset(value) is None:
+                raise TypeError("tzinfo is required")
+            value = value.astimezone(datetime.timezone.utc).replace(tzinfo=None)
+        return value
+
+    def process_result_value(self, value, dialect):
+        if value is not None:
+            value = value.replace(tzinfo=datetime.timezone.utc)
+        return value
+
+
 class GUID(TypeDecorator):
     """Platform-independent GUID type.
 
@@ -75,7 +103,7 @@ class User(Base):
     _device_tokens = Column(String, default="")
     is_active = Column(Boolean, default=True, nullable=False)
     is_admin = Column(Boolean, default=False, nullable=False)
-    login_token_expire_date = Column(DateTime, default=datetime.utcnow, nullable=False)
+    login_token_expire_date = Column(TZDateTime, default=utcnow, nullable=False)
 
     services = relationship(
         "Service",
@@ -119,7 +147,9 @@ def android_tokens(self):
 
     @property
     def is_logged_in(self):
-        return datetime.utcnow() < self.login_token_expire_date
+        return (
+            datetime.datetime.now(datetime.timezone.utc) < self.login_token_expire_date
+        )
 
     def add_device_token(self, value: str):
         if value in self.device_tokens:
@@ -180,7 +210,7 @@ class Notification(Base):
     __tablename__ = "notifications"
 
     id = Column(Integer, primary_key=True, index=True)
-    timestamp = Column(DateTime, index=True, default=datetime.utcnow, nullable=False)
+    timestamp = Column(TZDateTime, index=True, default=utcnow, nullable=False)
     title = Column(String, nullable=False)
     subtitle = Column(String)
     url = Column(String)

app/schemas.py

@@ -7,6 +7,7 @@
 from typing_extensions import Annotated
 from pydantic import ConfigDict, BaseModel
 from pydantic.functional_validators import AfterValidator
+from pydantic.functional_serializers import PlainSerializer
 
 RE_COLOR = re.compile(r"^[0-9a-fA-F]{6}$")
 
@@ -18,6 +19,16 @@ def validate_color(color: str) -> str:
 
 
 Color = Annotated[str, AfterValidator(validate_color)]
+# Pydantic serialization adds "Z" to timestamp in UTC timezone.
+# To not brake current mobile clients, we should continue returning
+# timestamp without timezone info and use a custom PlainSerializer.
+# Note that python isoformat() doesn't return "Z" but "+00:00".
+NoTZDateTime = Annotated[
+    datetime.datetime,
+    PlainSerializer(
+        lambda x: x.isoformat().replace("+00:00", ""), return_type=str, when_used="json"
+    ),
+]
 
 
 class SortOrder(str, Enum):
@@ -105,7 +116,7 @@ class NotificationCreate(NotificationBase):
 
 class Notification(NotificationBase):
     id: int
-    timestamp: datetime.datetime
+    timestamp: NoTZDateTime
     service_id: uuid.UUID
     model_config = ConfigDict(from_attributes=True)
 

app/settings.py

@@ -78,3 +78,6 @@
 )
 
 APP_NAME = config("APP_NAME", cast=str, default="ESS Notify")
+
+# Local timezone used for timestamp in the web app
+LOCAL_TIMEZONE = config("LOCAL_TIMEZONE", cast=str, default="Europe/Stockholm")

app/templates/_helpers.html

@@ -9,10 +9,6 @@
 </pre>
 {%- endmacro %}
 
-{% macro format_datetime(dt) -%}
-{{ dt.strftime("%Y-%m-%d %H:%M:%S") }}
-{%- endmacro %}
-
 {% macro is_checked(value, expected) -%}
 {% if value == expected %}checked{% endif -%}
-{%- endmacro %}
+{%- endmacro %}

app/templates/notifications_table.html

@@ -1,4 +1,4 @@
-{% from "_helpers.html" import format_notification, format_datetime %}
+{% from "_helpers.html" import format_notification %}
 
 <table class="table table-striped table-sm" id="notifications-table">
   <thead>
@@ -12,7 +12,7 @@
     {% for notification in notifications %}
     <tr>
       <td>{{ categories[notification.service_id] }}</td>
-      <td>{{ format_datetime(notification.timestamp) }}</td>
+      <td>{{ notification.timestamp | format_datetime }}</td>
       <td>{{ format_notification(notification) }}</td>
     </tr>
     {% endfor %}

app/utils.py

@@ -3,7 +3,7 @@
 import ipaddress
 import uuid
 import jwt
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import List, Optional, Dict
 from sqlalchemy.orm import Session
 from . import models, ios, firebase
@@ -75,7 +75,7 @@ async def sem_task(task):
 async def send_notification(db: Session, notification: models.Notification) -> None:
     """Send the notification to all subscribers"""
     tasks = []
-    ios_headers = ios.create_headers(datetime.utcnow())
+    ios_headers = ios.create_headers(datetime.now(timezone.utc))
     ios_client = httpx.AsyncClient(http2=True, headers=ios_headers)
     android_headers = await firebase.create_headers(str(uuid.uuid4()))
     android_client = httpx.AsyncClient(headers=android_headers)

app/views/__init__.py

@@ -1,5 +1,14 @@
 from pathlib import Path
+from zoneinfo import ZoneInfo
 from starlette.templating import Jinja2Templates
+from ..settings import LOCAL_TIMEZONE
 
 app_dir = Path(__file__).parents[1].resolve()
 templates = Jinja2Templates(directory=str(app_dir / "templates"))
+
+
+def format_datetime(dt):
+    return dt.astimezone(ZoneInfo(LOCAL_TIMEZONE)).strftime("%Y-%m-%d %H:%M:%S")
+
+
+templates.env.filters["format_datetime"] = format_datetime

pyproject.toml

@@ -1,6 +1,3 @@
-[tool.black]
-target_version = ['py37']
-
 [build-system]
 requires = ["setuptools >= 42", "wheel", "setuptools_scm[toml]>=3.4"]
 build-backend = "setuptools.build_meta"

setup.py

@@ -60,5 +60,5 @@
     ],
     entry_points={"console_scripts": ["notify-server=app.command:cli"]},
     extras_require={"postgres": postgres_requires, "tests": tests_requires},
-    python_requires=">=3.8",
+    python_requires=">=3.9",
 )

tests/api/test_services.py

@@ -5,6 +5,7 @@
 import packaging.version
 from fastapi.testclient import TestClient
 from app import models, schemas
+from ..utils import no_tz_isoformat
 
 
 @pytest.fixture(scope="module")
@@ -155,15 +156,15 @@ def test_read_service_notifications(
             "id": notification1.id,
             "service_id": str(notification1.service_id),
             "subtitle": notification1.subtitle,
-            "timestamp": notification1.timestamp.isoformat(),
+            "timestamp": no_tz_isoformat(notification1.timestamp),
             "title": notification1.title,
             "url": notification1.url,
         },
         {
             "id": notification2.id,
             "service_id": str(notification2.service_id),
             "subtitle": notification2.subtitle,
-            "timestamp": notification2.timestamp.isoformat(),
+            "timestamp": no_tz_isoformat(notification2.timestamp),
             "title": notification2.title,
             "url": notification2.url,
         },
@@ -262,7 +263,7 @@ def test_create_notification_for_service(
         "id": db_notification.id,
         "service_id": str(service.id),
         "subtitle": sample_notification["subtitle"],
-        "timestamp": db_notification.timestamp.isoformat(),
+        "timestamp": no_tz_isoformat(db_notification.timestamp),
         "title": sample_notification["title"],
         "url": sample_notification["url"],
     }

tests/api/test_users.py

@@ -1,12 +1,13 @@
 import pytest
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from fastapi.testclient import TestClient
 from app import schemas, models, utils
+from ..utils import no_tz_isoformat
 
 
 def user_authorization_headers(username):
     token = utils.create_access_token(
-        username, expire=datetime.utcnow() + timedelta(minutes=60)
+        username, expire=datetime.now(timezone.utc) + timedelta(minutes=60)
     )
     return {"Authorization": f"Bearer {token}"}
 
@@ -55,7 +56,7 @@ def test_read_current_user_profile_invalid_username(client: TestClient, api_vers
 
 def test_read_current_user_profile_expired_token(client: TestClient, user, api_version):
     token = utils.create_access_token(
-        user.username, expire=datetime.utcnow() + timedelta(minutes=-5)
+        user.username, expire=datetime.now(timezone.utc) + timedelta(minutes=-5)
     )
     response = client.get(
         f"/api/{api_version}/users/user/profile",
@@ -257,7 +258,7 @@ def test_read_current_user_notifications(
             "is_read": False,
             "service_id": str(notification1.service_id),
             "subtitle": notification1.subtitle,
-            "timestamp": notification1.timestamp.isoformat(),
+            "timestamp": no_tz_isoformat(notification1.timestamp),
             "title": notification1.title,
             "url": notification1.url,
         },
@@ -266,7 +267,7 @@ def test_read_current_user_notifications(
             "is_read": False,
             "service_id": str(notification2.service_id),
             "subtitle": notification2.subtitle,
-            "timestamp": notification2.timestamp.isoformat(),
+            "timestamp": no_tz_isoformat(notification2.timestamp),
             "title": notification2.title,
             "url": notification2.url,
         },

tests/conftest.py

@@ -71,7 +71,8 @@ def client() -> Generator:
 def user_token_headers(user):
     token = create_access_token(
         user.username,
-        expire=datetime.datetime.utcnow() + datetime.timedelta(minutes=60),
+        expire=datetime.datetime.now(datetime.timezone.utc)
+        + datetime.timedelta(minutes=60),
     )
     return {"Authorization": f"Bearer {token}"}
 
@@ -81,7 +82,8 @@ def admin_token_headers(user_factory):
     admin = user_factory(is_admin=True)
     token = create_access_token(
         admin.username,
-        expire=datetime.datetime.utcnow() + datetime.timedelta(minutes=60),
+        expire=datetime.datetime.now(datetime.timezone.utc)
+        + datetime.timedelta(minutes=60),
     )
     return {"Authorization": f"Bearer {token}"}
 
@@ -102,6 +104,8 @@ def _make_device_token(length):
 @pytest.fixture
 def notification_date():
     def _notification_date(days_old):
-        return datetime.datetime.utcnow() - datetime.timedelta(days=days_old)
+        return datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(
+            days=days_old
+        )
 
     return _notification_date

tests/test_crud.py

@@ -210,7 +210,7 @@ def test_get_user_notifications_limit(db, user, service, limit, sort):
     user.subscribe(service)
     db.commit()
     notifications = []
-    now = datetime.datetime.now()
+    now = datetime.datetime.now(datetime.timezone.utc)
     for nb in range(20):
         notification = crud.create_service_notification(
             db, schemas.NotificationCreate(title=f"message{nb}"), service
@@ -253,7 +253,7 @@ def test_get_user_notifications_filter_services_id(db, user, service_factory):
     user.subscribe(service3)
     db.commit()
     notifications = []
-    now = datetime.datetime.now()
+    now = datetime.datetime.now(datetime.timezone.utc)
     for service_nb, service in enumerate((service1, service2, service3), start=1):
         for nb in range(10):
             notification = crud.create_service_notification(

tests/test_models.py

@@ -1,5 +1,5 @@
 import uuid
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from sqlalchemy.orm import Session
 from app import schemas
 
@@ -19,7 +19,7 @@ def test_user_is_logged_in(db: Session, user_factory) -> None:
     username = "johndoe"
     user = user_factory(username=username)
     assert not user.is_logged_in
-    user.login_token_expire_date = datetime.utcnow() + timedelta(minutes=5)
+    user.login_token_expire_date = datetime.now(timezone.utc) + timedelta(minutes=5)
     assert user.is_logged_in