Fix incorrect assert for active_since when creating a Subscription

docker-app/qfieldcloud/authentication/auth_backends.py

@@ -30,7 +30,7 @@ def get_user(self, user_id):
         """Almost the same as `contrib.auth.backends.ModelBackend`, but not using the default manager, but the normal `objects` manager
 
         Returns:
-            Optional[Union[Person, Organization, Team]]: In theory it can return any of this three types, however it will always be a Person or None
+            Person | Organization | Team | None: In theory it can return any of these types, however it will always be a `Person` or `None`
         """
         UserModel = get_user_model()
 

docker-app/qfieldcloud/core/models.py

@@ -5,7 +5,7 @@
 import uuid
 from datetime import datetime, timedelta
 from enum import Enum
-from typing import List, Optional, cast
+from typing import cast
 
 import django_cryptography.fields
 from deprecated import deprecated
@@ -1079,7 +1079,7 @@ def name_with_owner(self) -> str:
         return f"{self.owner.username}/{self.name}"
 
     @property
-    def attachment_dirs(self) -> List[str]:
+    def attachment_dirs(self) -> list[str]:
         """Returns a list of configured attachment dirs for the project.
 
         Attachment dir is a special directory in the QField infrastructure that holds attachment files
@@ -1089,7 +1089,7 @@ def attachment_dirs(self) -> List[str]:
         neither the extraction from the projectfile, nor the configuration in QFieldSync are implemented.
 
         Returns:
-            List[str]: A list configured attachment dirs for the project.
+            list[str]: A list configured attachment dirs for the project.
         """
         attachment_dirs = []
 
@@ -1107,7 +1107,7 @@ def private(self) -> bool:
         return not self.is_public
 
     @cached_property
-    def files(self) -> List[utils.S3ObjectWithVersions]:
+    def files(self) -> list[utils.S3ObjectWithVersions]:
         """Gets all the files from S3 storage. This is potentially slow. Results are cached on the instance."""
         return list(utils.get_project_files_with_versions(self.id))
 
@@ -1121,7 +1121,7 @@ def users(self):
         return User.objects.for_project(self)
 
     @property
-    def has_online_vector_data(self) -> Optional[bool]:
+    def has_online_vector_data(self) -> bool | None:
         """Returns None if project details or layers details are not available"""
 
         if not self.project_details:

docker-app/qfieldcloud/core/permission_check.py

@@ -1,13 +1,11 @@
-from typing import Any, Callable, List, Union
+from typing import Any, Callable
 
 from django.http.request import HttpRequest
 from django.http.response import HttpResponse, HttpResponseForbidden
 from qfieldcloud.core import permissions_utils
 
 
-def permission_check(
-    perm: str, check_args: List[Union[str, Callable]] = []
-) -> Callable:
+def permission_check(perm: str, check_args: list[str | Callable] = []) -> Callable:
     perm_check = getattr(permissions_utils, perm)
 
     def decorator_wrapper(func: Callable):

docker-app/qfieldcloud/core/permissions_utils.py

@@ -1,4 +1,4 @@
-from typing import List, Literal, Union
+from typing import Literal
 
 from django.utils.translation import gettext as _
 from qfieldcloud.authentication.models import AuthToken
@@ -76,7 +76,7 @@ def _organization_of_owner(user: QfcUser, organization: Organization):
 def user_has_project_roles(
     user: QfcUser,
     project: Project,
-    roles: List[ProjectCollaborator.Roles],
+    roles: list[ProjectCollaborator.Roles],
     skip_invalid: bool = False,
 ):
     return (
@@ -87,7 +87,7 @@ def user_has_project_roles(
 
 
 def check_user_has_project_role_origins(
-    user: QfcUser, project: Project, origins: List[ProjectQueryset.RoleOrigins]
+    user: QfcUser, project: Project, origins: list[ProjectQueryset.RoleOrigins]
 ) -> Literal[True]:
     if (
         _project_for_owner(user, project, skip_invalid=False)
@@ -106,7 +106,7 @@ def check_user_has_project_role_origins(
 
 
 def user_has_project_role_origins(
-    user: QfcUser, project: Project, origins: List[ProjectQueryset.RoleOrigins]
+    user: QfcUser, project: Project, origins: list[ProjectQueryset.RoleOrigins]
 ) -> bool:
     try:
         return check_user_has_project_role_origins(user, project, origins)
@@ -115,7 +115,7 @@ def user_has_project_role_origins(
 
 
 def check_user_has_organization_roles(
-    user: QfcUser, organization: Organization, roles: List[OrganizationMember.Roles]
+    user: QfcUser, organization: Organization, roles: list[OrganizationMember.Roles]
 ) -> Literal[True]:
     if (
         _organization_of_owner(user, organization)
@@ -134,7 +134,7 @@ def check_user_has_organization_roles(
 
 
 def user_has_organization_roles(
-    user: QfcUser, organization: Organization, roles: List[OrganizationMember.Roles]
+    user: QfcUser, organization: Organization, roles: list[OrganizationMember.Roles]
 ) -> bool:
     try:
         return check_user_has_organization_roles(user, organization, roles)
@@ -145,7 +145,7 @@ def user_has_organization_roles(
 def user_has_organization_role_origins(
     user: QfcUser,
     organization: Organization,
-    origins: List[OrganizationQueryset.RoleOrigins],
+    origins: list[OrganizationQueryset.RoleOrigins],
 ):
     return (
         _organization_of_owner(user, organization)
@@ -165,7 +165,7 @@ def get_param_from_request(request, param):
 
 
 def can_create_project(
-    user: QfcUser, organization: Union[QfcUser, Organization] = None
+    user: QfcUser, organization: QfcUser | Organization = None
 ) -> bool:
     """Return True if the `user` can create a project. Accepts additional
     `organization` to check whether the user has permissions to do so on

docker-app/qfieldcloud/core/serializers.py

@@ -1,5 +1,4 @@
 import os
-from typing import Optional
 
 from django.contrib.sites.models import Site
 from qfieldcloud.authentication.models import AuthToken
@@ -361,7 +360,7 @@ def to_internal_value(self, data):
 
         return internal_data
 
-    def get_lastest_not_finished_job(self) -> Optional[Job]:
+    def get_lastest_not_finished_job(self) -> Job | None:
         ModelClass: Job = self.Meta.model
         last_active_job = (
             ModelClass.objects.filter(
@@ -405,7 +404,7 @@ class Meta:
 
 
 class PackageJobSerializer(JobMixin, serializers.ModelSerializer):
-    def get_lastest_not_finished_job(self) -> Optional[Job]:
+    def get_lastest_not_finished_job(self) -> Job | None:
         job = super().get_lastest_not_finished_job()
         if job:
             return job

docker-app/qfieldcloud/core/tests/test_packages.py

@@ -3,7 +3,6 @@
 import os
 import tempfile
 import time
-from typing import List, Tuple
 
 import psycopg2
 from django.http import FileResponse
@@ -69,7 +68,7 @@ def upload_files(
         self,
         token: str,
         project: Project,
-        files: List[Tuple[str, str]],
+        files: list[tuple[str, str]],
     ):
         self.client.credentials(HTTP_AUTHORIZATION=f"Token {token}")
         for local_filename, remote_filename in files:
@@ -88,11 +87,11 @@ def upload_files_and_check_package(
         self,
         token: str,
         project: Project,
-        files: List[Tuple[str, str]],
-        expected_files: List[str],
-        job_create_error: Tuple[int, str] = None,
+        files: list[tuple[str, str]],
+        expected_files: list[str],
+        job_create_error: tuple[int, str] = None,
         tempdir: str = None,
-        invalid_layers: List[str] = [],
+        invalid_layers: list[str] = [],
     ):
         self.upload_files(token, project, files)
         self.check_package(
@@ -103,10 +102,10 @@ def check_package(
         self,
         token: str,
         project: Project,
-        expected_files: List[str],
-        job_create_error: Tuple[int, str] = None,
+        expected_files: list[str],
+        job_create_error: tuple[int, str] = None,
         tempdir: str = None,
-        invalid_layers: List[str] = [],
+        invalid_layers: list[str] = [],
     ):
         self.client.credentials(HTTP_AUTHORIZATION=f"Token {token}")
 

docker-app/qfieldcloud/core/utils.py

@@ -6,7 +6,7 @@
 import posixpath
 from datetime import datetime
 from pathlib import PurePath
-from typing import IO, Generator, NamedTuple, Optional, Union
+from typing import IO, Generator, NamedTuple
 
 import boto3
 import jsonschema
@@ -146,9 +146,7 @@ def get_sha256(file: IO) -> str:
         return _get_sha256_file(file)
 
 
-def _get_sha256_memory_file(
-    file: Union[InMemoryUploadedFile, TemporaryUploadedFile]
-) -> str:
+def _get_sha256_memory_file(file: InMemoryUploadedFile | TemporaryUploadedFile) -> str:
     BLOCKSIZE = 65536
     hasher = hashlib.sha256()
 
@@ -178,9 +176,7 @@ def get_md5sum(file: IO) -> str:
         return _get_md5sum_file(file)
 
 
-def _get_md5sum_memory_file(
-    file: Union[InMemoryUploadedFile, TemporaryUploadedFile]
-) -> str:
+def _get_md5sum_memory_file(file: InMemoryUploadedFile | TemporaryUploadedFile) -> str:
     BLOCKSIZE = 65536
     hasher = hashlib.md5()
 
@@ -258,7 +254,7 @@ def is_qgis_project_file(filename: str) -> bool:
     return False
 
 
-def get_qgis_project_file(project_id: str) -> Optional[str]:
+def get_qgis_project_file(project_id: str) -> str | None:
     """Return the relative path inside the project of the qgs/qgz file or
     None if no qgs/qgz file is present"""
 
@@ -274,7 +270,7 @@ def get_qgis_project_file(project_id: str) -> Optional[str]:
     return None
 
 
-def check_s3_key(key: str) -> Optional[str]:
+def check_s3_key(key: str) -> str | None:
     """Check to see if an object exists on S3. It it exists, the function
     returns the sha256 of the file from the metadata"""
 
@@ -349,7 +345,7 @@ def get_project_files_with_versions(
 
 def get_project_file_with_versions(
     project_id: str, filename: str
-) -> Optional[S3ObjectWithVersions]:
+) -> S3ObjectWithVersions | None:
     """Returns a list of files and their versions.
 
     Args:
@@ -480,7 +476,7 @@ def list_files_with_versions(
     """
     last_key = None
     versions: list[S3ObjectVersion] = []
-    latest: Optional[S3ObjectVersion] = None
+    latest: S3ObjectVersion | None = None
 
     for v in list_versions(bucket, prefix, strip_prefix):
         if last_key != v.key:

docker-app/qfieldcloud/core/utils2/audit.py

@@ -1,5 +1,5 @@
 import json
-from typing import Any, Dict, List, Union
+from typing import Any
 
 from auditlog.models import LogEntry
 from django.contrib.auth.models import AnonymousUser, User
@@ -9,7 +9,7 @@
 def audit(
     instance,
     action: LogEntry.Action,
-    changes: Union[Dict[str, Any], List[Any], str] = None,
+    changes: dict[str, Any] | list[Any] | str | None = None,
     actor: User = None,
     remote_addr: str = None,
     additional_data: Any = None,

docker-app/qfieldcloud/core/utils2/jobs.py

@@ -1,5 +1,4 @@
 import logging
-from typing import List
 
 import qfieldcloud.core.models as models
 from django.conf import settings
@@ -16,8 +15,8 @@ def apply_deltas(
     user: "models.User",
     project_file: str,
     overwrite_conflicts: bool,
-    delta_ids: List[str] = [],
-) -> List["models.ApplyJob"]:
+    delta_ids: list[str] = [],
+) -> list["models.ApplyJob"]:
     """Apply a deltas"""
 
     logger.info(

docker-app/qfieldcloud/core/utils2/projects.py

@@ -82,8 +82,8 @@ def create_collaborator_by_username_or_email(
         )
     elif users[0].is_organization:
         message = _(
-            f'Organization "{username}" cannot be added. Only users and teams can be collaborators.'
-        )
+            'Organization "{}" cannot be added. Only users and teams can be collaborators.'
+        ).format(username)
     else:
         success, message = create_collaborator(project, users[0], created_by)
 

docker-app/qfieldcloud/subscription/models.py

@@ -2,7 +2,7 @@
 import uuid
 from datetime import datetime, timedelta
 from functools import lru_cache
-from typing import Optional, TypedDict, cast
+from typing import Type, TypedDict, cast
 
 from constance import config
 from deprecated import deprecated
@@ -794,15 +794,15 @@ def create_subscription(
         account: UserAccount,
         plan: Plan,
         created_by: Person,
-        active_since: Optional[datetime] = None,
-    ) -> tuple[Optional["AbstractSubscription"], "AbstractSubscription"]:
+        active_since: datetime | None = None,
+    ) -> tuple[Type["AbstractSubscription"] | None, "AbstractSubscription"]:
         """Creates a subscription for a given account to a given plan. If the plan is a trial, create the default subscription in the end of the period.
 
         Args:
             account (UserAccount): the account the subscription belongs to.
             plan (Plan): the plan to subscribe to. Note if the the plan is a trial, the first return value would be the trial subscription, otherwise it would be None.
             created_by (Person): created by.
-            active_since (Optional[datetime]): active since for the subscription.
+            active_since (datetime | None): active since for the subscription.
 
         Returns:
             tuple[AbstractSubscription | None, AbstractSubscription]: the created trial subscription if the given plan was a trial and the regular subscription.
@@ -813,8 +813,7 @@ def create_subscription(
             # remove microseconds as there will be slight shift with the remote system data
             active_since = active_since.replace(microsecond=0)
 
-        assert active_since
-
+        regular_active_since: datetime | None = None
         if plan.is_trial:
             assert isinstance(
                 active_since, datetime