refactor: replace deprecated datetime.utcnow with datetime.now(timezo…

invenio_accounts/cli.py

@@ -3,14 +3,14 @@
 # This file is part of Invenio.
 # Copyright (C) 2015-2023 CERN.
 # Copyright (C) 2024 Graz University of Technology.
+# Copyright (C) 2024      KTH Royal Institute of Technology.
 #
 # Invenio is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
 
 """Command Line Interface for accounts."""
 
 import json
-from datetime import datetime
 from functools import wraps
 
 import click
@@ -23,6 +23,7 @@
 from werkzeug.local import LocalProxy
 
 from .models import DomainCategory
+from .utils import get_utc_now
 
 _datastore = LocalProxy(lambda: current_app.extensions["security"].datastore)
 
@@ -71,7 +72,7 @@ def users_create(email, password, active, confirm, profile):
         kwargs["password"] = hash_password(kwargs["password"])
         kwargs["active"] = active
         if confirm:
-            kwargs["confirmed_at"] = datetime.utcnow()
+            kwargs["confirmed_at"] = get_utc_now()
         if profile:
             kwargs["user_profile"] = json.loads(profile)
         _datastore.create_user(**kwargs)

invenio_accounts/datastore.py

@@ -2,13 +2,13 @@
 #
 # This file is part of Invenio.
 # Copyright (C) 2015-2024 CERN.
+# Copyright (C) 2024      KTH Royal Institute of Technology.
 #
 # Invenio is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
 
 """Session-aware datastore."""
 
-from datetime import datetime
 
 from flask import current_app
 from flask_security import SQLAlchemyUserDatastore, user_confirmed
@@ -18,14 +18,15 @@
 from .proxies import current_db_change_history
 from .sessions import delete_user_sessions
 from .signals import datastore_post_commit, datastore_pre_commit
+from .utils import get_utc_now
 
 
 class SessionAwareSQLAlchemyUserDatastore(SQLAlchemyUserDatastore):
     """Datastore which deletes active session when a user is deactivated."""
 
     def verify_user(self, user):
         """Verify a user."""
-        now = datetime.utcnow()
+        now = get_utc_now()
         user.blocked_at = None
         user.verified_at = now
         user.active = True
@@ -35,7 +36,7 @@ def verify_user(self, user):
 
     def block_user(self, user):
         """Verify a user."""
-        now = datetime.utcnow()
+        now = get_utc_now()
         user.blocked_at = now
         user.verified_at = None
         user.active = False
@@ -47,7 +48,7 @@ def activate_user(self, user):
         res = super().activate_user(user)
         user.blocked_at = None
         if user.confirmed_at is None:
-            user.confirmed_at = datetime.utcnow()
+            user.confirmed_at = get_utc_now()
             user_confirmed.send(current_app._get_current_object(), user=user)
         return res
 

invenio_accounts/models.py

@@ -2,15 +2,14 @@
 #
 # This file is part of Invenio.
 # Copyright (C) 2015-2024 CERN.
-# Copyright (C) 2022 KTH Royal Institute of Technology
+# Copyright (C) 2022-2024 KTH Royal Institute of Technology
 #
 # Invenio is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
 
 """Database models for accounts."""
 
 import uuid
-from datetime import datetime
 
 from flask import current_app, session
 from flask_babel import refresh
@@ -26,7 +25,7 @@
 
 from .errors import AlreadyLinkedError
 from .profiles import UserPreferenceDict, UserProfileDict
-from .utils import DomainStatus, split_emailaddr, validate_username
+from .utils import DomainStatus, get_utc_now, split_emailaddr, validate_username
 
 json_field = (
     db.JSON()
@@ -168,7 +167,7 @@ class User(db.Model, Timestamp, UserMixin):
     def __init__(self, *args, **kwargs):
         """Constructor."""
         self.verified_at = (
-            datetime.utcnow()
+            get_utc_now()
             if current_app.config.get("ACCOUNTS_DEFAULT_USERS_VERIFIED")
             else None
         )
@@ -414,7 +413,7 @@ class SessionActivity(db.Model, Timestamp):
     def query_by_expired(cls):
         """Query to select all expired sessions."""
         lifetime = current_app.permanent_session_lifetime
-        expired_moment = datetime.utcnow() - lifetime
+        expired_moment = get_utc_now() - lifetime
         return cls.query.filter(cls.created < expired_moment)
 
     @classmethod

invenio_accounts/tasks.py

@@ -2,13 +2,13 @@
 #
 # This file is part of Invenio.
 # Copyright (C) 2015-2018 CERN.
+# Copyright (C) 2024      KTH Royal Institute of Technology.
 #
 # Invenio is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
 
 """Background tasks for accounts."""
 
-from datetime import datetime
 
 from celery import shared_task
 from flask import current_app
@@ -18,6 +18,7 @@
 
 from .models import Domain, LoginInformation, SessionActivity, User
 from .sessions import delete_session
+from .utils import get_utc_now
 
 
 @shared_task
@@ -60,9 +61,7 @@ def clean_session_table():
 @shared_task
 def delete_ips():
     """Automatically remove login_info.last_login_ip older than 30 days."""
-    expiration_date = (
-        datetime.utcnow() - current_app.config["ACCOUNTS_RETENTION_PERIOD"]
-    )
+    expiration_date = get_utc_now() - current_app.config["ACCOUNTS_RETENTION_PERIOD"]
 
     LoginInformation.query.filter(
         LoginInformation.last_login_ip.isnot(None),
@@ -126,7 +125,7 @@ def update_domain_status():
 
     # Commit batches of 500 updates
     batch_size = 500
-    now = datetime.utcnow()
+    now = get_utc_now()
 
     # Process updates in batches
     for i in range(0, len(domain_updates), batch_size):

invenio_accounts/utils.py

@@ -2,6 +2,7 @@
 #
 # This file is part of Invenio.
 # Copyright (C) 2017-2024 CERN.
+# Copyright (C) 2024      KTH Royal Institute of Technology.
 #
 # Invenio is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
@@ -11,7 +12,7 @@
 import enum
 import re
 import uuid
-from datetime import datetime
+from datetime import datetime, timezone
 from urllib.parse import parse_qs, urlencode, urlsplit, urlunsplit
 
 from flask import current_app, request, session, url_for
@@ -69,7 +70,7 @@ def jwt_create_token(user_id=None, additional_data=None):
     # Create an ID
     uid = str(uuid.uuid4())
     # The time in UTC now
-    now = datetime.utcnow()
+    now = get_utc_now()
     # Build the token data
     token_data = {
         "exp": now + current_app.config["ACCOUNTS_JWT_EXPIRATION_DELTA"],
@@ -269,3 +270,8 @@ def split_emailaddr(email):
     if domain[-1] == ".":
         domain = domain[:-1]
     return prefix, domain
+
+
+def get_utc_now():
+    """Get the current time in UTC without timezone information for backwards compatibility."""
+    return datetime.now(timezone.utc).replace(tzinfo=None)

tests/test_tasks.py

@@ -2,14 +2,15 @@
 #
 # This file is part of Invenio.
 # Copyright (C) 2015-2018 CERN.
+# Copyright (C) 2024      KTH Royal Institute of Technology.
 #
 # Invenio is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
 
 
 """Module tests."""
 
-from datetime import datetime, timedelta
+from datetime import timedelta
 from time import sleep
 
 from flask import url_for
@@ -20,6 +21,7 @@
 from invenio_accounts.models import SessionActivity, User
 from invenio_accounts.tasks import clean_session_table, delete_ips, send_security_email
 from invenio_accounts.testutils import create_test_user
+from invenio_accounts.utils import get_utc_now
 
 
 def test_send_message_outbox(task_app):
@@ -113,11 +115,9 @@ def test():
 def test_delete_ips(task_app):
     """Test if ips are deleted after 30 days."""
     last_login_at1 = (
-        datetime.utcnow()
-        - task_app.config["ACCOUNTS_RETENTION_PERIOD"]
-        - timedelta(days=1)
+        get_utc_now() - task_app.config["ACCOUNTS_RETENTION_PERIOD"] - timedelta(days=1)
     )
-    last_login_at2 = datetime.utcnow()
+    last_login_at2 = get_utc_now()
 
     with task_app.app_context():
         user1 = create_test_user(