adjusted the code according to latest Python and FastAPI annotation improvements

Author: kamikaze

setup.cfg

@@ -35,6 +35,7 @@ install_requires =
     rawpy==0.17.3
     Pillow==9.4.0
     pydantic==1.10.4
+    tzdata==2022.7
 # The usage of test_requires is discouraged, see `Dependency Management` docs
 # tests_require = pytest; pytest-cov
 # Require a specific Python version, e.g. Python 2.7 or >= 3.4
@@ -51,17 +52,17 @@ exclude =
 # PDF = ReportLab; RXP
 # Add here test requirements (semicolon/line-separated)
 desktop =
-    PySide6-Essentials==6.4.1
+    PySide6-Essentials==6.4.2
 web =
     alembic==1.9.1
     asyncpg==0.27.0
     Authlib==1.2.0
     passlib==1.7.4
     databases[postgresql]==0.7.0
-    fastapi==0.89.0
-    fastapi-pagination==0.11.1
+    fastapi==0.89.1
+    fastapi-pagination==0.11.2
     fastapi-users[sqlalchemy]==10.2.1
-    orjson==3.8.4
+    orjson==3.8.5
     PyJWT==2.6.0
     python-dateutil==2.8.2
     SQLAlchemy==1.4.46

src/dataset_image_annotator/__main__.py

@@ -3,7 +3,7 @@
 import sys
 from collections import defaultdict
 from pathlib import Path
-from typing import Sequence, Union, Optional, Mapping
+from typing import Sequence, Union, Mapping
 
 import rawpy
 from PySide6.QtCore import QFile, QIODevice, QDir, QFileInfo, QModelIndex, Qt, QStringListModel
@@ -37,7 +37,7 @@ def list_dir_images(path: Path) -> Sequence[Path]:
     return list_dir(path, '*.ARW')
 
 
-def list_dir_metadata(path: Path) -> Optional[Sequence[Path]]:
+def list_dir_metadata(path: Path) -> Sequence[Path] | None:
     metadata_dir_path = Path(path, '.metadata')
 
     if metadata_dir_path.exists():
@@ -116,7 +116,7 @@ class MainWindow:
     def __init__(self, data_root_path: Path):
         self.data_root_path = None
         self.metadata = defaultdict(dict)
-        self.selected_file_name: Optional[str] = None
+        self.selected_file_name: str | None = None
         self.types = set()
         self.makes = set()
         self.models = set()

src/dataset_image_annotator/api/users.py

@@ -1,7 +1,6 @@
 import contextlib
 import logging
 import uuid
-from typing import Optional
 
 from fastapi import Request, Depends
 from fastapi_users import BaseUserManager, UUIDIDMixin
@@ -16,16 +15,16 @@
 
 
 class UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]):
-    reset_password_token_secret = settings.auth_secret
-    verification_token_secret = settings.auth_secret
+    reset_password_token_secret = settings.auth_secret.get_secret_value()
+    verification_token_secret = settings.auth_secret.get_secret_value()
 
-    async def on_after_register(self, user: User, request: Optional[Request] = None):
+    async def on_after_register(self, user: User, request: Request | None = None):
         logger.info(f'User {user.id} has registered.')
 
-    async def on_after_forgot_password(self, user: User, token: str, request: Optional[Request] = None):
+    async def on_after_forgot_password(self, user: User, token: str, request: Request | None = None):
         logger.info(f'User {user.id} has forgot their password. Reset token: {token}')
 
-    async def on_after_request_verify(self, user: User, token: str, request: Optional[Request] = None):
+    async def on_after_request_verify(self, user: User, token: str, request: Request | None = None):
         logger.info(f'Verification requested for user {user.id}. Verification token: {token}')
 
 

src/dataset_image_annotator/api/v1/endpoints.py

@@ -2,7 +2,7 @@
 import uuid
 from functools import wraps
 from inspect import signature
-from typing import Optional, Any, List
+from typing import Sequence
 
 from fastapi import APIRouter, HTTPException, status, Depends, UploadFile, File
 from fastapi.responses import ORJSONResponse
@@ -29,7 +29,7 @@
 
 
 def get_jwt_strategy() -> JWTStrategy:
-    return JWTStrategy(secret=settings.auth_secret, lifetime_seconds=3600)
+    return JWTStrategy(secret=settings.auth_secret.get_secret_value(), lifetime_seconds=3600)
 
 
 auth_backend = AuthenticationBackend(name='cluserauth', transport=cookie_transport, get_strategy=get_jwt_strategy)
@@ -77,19 +77,19 @@ async def wrapper(*args, **kwargs):
     return wrapper
 
 
-@router.get('/users', response_model=Page[UserItem], response_class=ORJSONResponse, tags=['Admin'])
+@router.get('/users', response_class=ORJSONResponse, tags=['Admin'])
 @handle_exceptions
-async def get_users(search: Optional[Json] = None, order_by: Optional[str] = None,
-                    user=Depends(get_current_user)):
+async def get_users(search: Json | None = None, order_by: str | None = None,
+                    user=Depends(get_current_user)) -> Page[UserItem]:
     if user.is_superuser:
         return await core.get_users(database, search, order_by)
 
     raise HTTPException(status_code=403)
 
 
-@router.post('/users', response_model=UserItem, response_class=ORJSONResponse, tags=['Admin'])
+@router.post('/users', response_class=ORJSONResponse, tags=['Admin'])
 @handle_exceptions
-async def create_user(new_user: UserCreate, user=Depends(get_current_user)):
+async def create_user(new_user: UserCreate, user=Depends(get_current_user)) -> UserItem:
     if user.is_superuser:
         created_user = await users.create_user(new_user)
 
@@ -98,10 +98,10 @@ async def create_user(new_user: UserCreate, user=Depends(get_current_user)):
     raise HTTPException(status_code=403)
 
 
-@router.get('/user-groups', response_model=List[UserGroup], response_class=ORJSONResponse, tags=['Admin'])
+@router.get('/user-groups', response_class=ORJSONResponse, tags=['Admin'])
 @handle_exceptions
-async def get_user_groups(search: Optional[Json[Any]] = None, order_by: Optional[str] = None,
-                          user=Depends(get_current_user)):
+async def get_user_groups(search: Json | None = None, order_by: str | None = None,
+                          user=Depends(get_current_user)) -> Sequence[UserGroup]:
     if user.is_active:
         return await core.get_user_groups(database, search, order_by)
 
@@ -110,7 +110,7 @@ async def get_user_groups(search: Optional[Json[Any]] = None, order_by: Optional
 
 @router.post('/raw-file', response_class=ORJSONResponse, tags=['Admin'])
 @handle_exceptions
-async def upload_raw_file(image_file: UploadFile = File(...)):
+async def upload_raw_file(image_file: UploadFile = File(...)) -> bool:
     if not image_file.filename:
         raise HTTPException(status_code=400, detail='Missing file')
 

src/dataset_image_annotator/api/v1/schemas.py

@@ -1,24 +1,23 @@
 import uuid
-from typing import Optional
 
 from fastapi_users import schemas
 from pydantic import BaseModel, UUID4
 
 
 class UserRead(schemas.BaseUser[uuid.UUID]):
-    group_id: Optional[int] = None
+    group_id: int | None = None
 
 
 class UserCreate(schemas.BaseUserCreate):
-    group_id: Optional[int] = None
+    group_id: int | None = None
 
 
 class UserUpdate(schemas.BaseUserUpdate):
-    group_id: Optional[int] = None
+    group_id: int | None = None
 
 
 class UserGroupSearchParams(BaseModel):
-    name: Optional[str]
+    name: str | None
 
 
 class UserGroup(BaseModel):
@@ -29,7 +28,7 @@ class UserGroup(BaseModel):
 class UserItem(BaseModel):
     id: UUID4
     email: str
-    group_id: Optional[int]
+    group_id: int | None
     is_active: bool
     is_superuser: bool
     is_verified: bool

src/dataset_image_annotator/conf.py

@@ -1,18 +1,17 @@
-from typing import Optional
-
-from pydantic import BaseSettings
+from pydantic import BaseSettings, PostgresDsn, SecretStr
 
 
 class Settings(BaseSettings):
     logging_level: str = 'INFO'
     logging_format: str = '%(asctime)s [%(levelname)s] %(name)s: %(message)s'
-    db_dsn: Optional[str] = None
+    db_dsn: PostgresDsn = None
     service_addr: str = '127.0.0.1'
     service_port: int = 8080
     alembic_auto_upgrade: bool = False
     alembic_config: str = 'alembic.ini'
-    bootstrap_user_email: Optional[str] = None
-    auth_secret: str = 'TODO-REPLACE'
+    bootstrap_user_email: str | None = None
+    auth_secret: SecretStr = 'TODO-REPLACE'
+    timezone: str = 'UTC'
 
 
 settings = Settings()

src/dataset_image_annotator/core.py

@@ -1,22 +1,22 @@
 import gettext
 import logging
 from pathlib import Path
-from typing import Mapping, Optional, Sequence
+from typing import Mapping, Sequence
 
 import sqlalchemy as sa
 from databases.backends.postgres import Record
-from fastapi_pagination.bases import AbstractPage
+from fastapi_pagination import Page
 from fastapi_pagination.ext.databases import paginate
 
+from dataset_image_annotator.api.v1.schemas import UserItem
 from dataset_image_annotator.db.models import UserGroup, User
 
 logger = logging.getLogger(__name__)
 t = gettext.translation('base', Path(Path(__file__).parent, 'locale'), fallback=True, languages=['lv_LV'])
 _ = t.gettext
 
 
-async def get_users(database, search: Optional[Mapping[str, str]] = None,
-                    order_by: Optional[str] = None) -> AbstractPage[Record]:
+async def get_users(database, search: Mapping[str, str] | None = None, order_by: str | None = None) -> Page[UserItem]:
     query = sa.select([User])
     result = await paginate(database, query)
 
@@ -29,8 +29,8 @@ async def get_user(database, user_id: str) -> Record:
     return await database.fetch_row(query)
 
 
-async def get_user_groups(database, search: Optional[Mapping[str, str]] = None,
-                          order_by: Optional[str] = None) -> Sequence[Record]:
+async def get_user_groups(database, search: Mapping[str, str] | None = None,
+                          order_by: str | None = None) -> Sequence[UserGroup]:
     query = sa.select([UserGroup]).order_by(UserGroup.name)
     result = await database.fetch_all(query)