Convert type annotations to Python 3.10 syntax

src/ewoksserver/__main__.py

@@ -6,8 +6,6 @@
 """
 
 import sys
-from typing import List
-from typing import Optional
 
 import click
 from uvicorn.main import main as uvicorn_main
@@ -66,7 +64,7 @@ def _ewoks_main(**cli_args):
 uvicorn_main.callback = _ewoks_main
 
 
-def main(argv: Optional[List[str]] = None) -> None:
+def main(argv: list[str] | None = None) -> None:
     """Exposes the uvicorn CLI with a default APP factory"""
     if argv is None:
         argv = sys.argv[1:]

src/ewoksserver/app/backends/binary_backend.py

@@ -3,7 +3,6 @@
 import mimetypes
 from pathlib import Path
 from typing import Iterator
-from typing import Union
 from urllib import request
 
 ResourceIdentifierType = str
@@ -13,7 +12,7 @@
 _logger = logging.getLogger(__name__)
 
 
-def root_url(root_url: Union[str, Path, None], category: str) -> ResourceUrlType:
+def root_url(root_url: str | Path | None, category: str) -> ResourceUrlType:
     if not root_url:
         root_url = Path(".")
     elif isinstance(root_url, str):

src/ewoksserver/app/backends/json_backend.py

@@ -1,19 +1,17 @@
 import json
 import logging
 from pathlib import Path
-from typing import Dict
 from typing import Iterator
-from typing import Union
 
 ResourceIdentifierType = str
 ResourceUrlType = Path
-ResourceContentType = Dict
+ResourceContentType = dict
 
 
 _logger = logging.getLogger(__name__)
 
 
-def root_url(root_url: Union[str, Path, None], category: str) -> ResourceUrlType:
+def root_url(root_url: str | Path | None, category: str) -> ResourceUrlType:
     if not root_url:
         root_url = Path(".")
     elif isinstance(root_url, str):

src/ewoksserver/app/config.py

@@ -5,10 +5,9 @@
 import os
 import sys
 import warnings
-from typing import Optional
+from typing import Annotated
 
 from fastapi import Depends
-from typing_extensions import Annotated
 
 from .models import AppSettings
 from .models import EwoksSettings
@@ -72,8 +71,8 @@ def _resolve_ewoks_scheduling_settings(celery: dict | None) -> dict:
 
 
 def create_ewoks_settings(
-    config: Optional[str] = None,
-    dir: Optional[str] = None,
+    config: str | None = None,
+    dir: str | None = None,
     without_events: bool = False,
     frontend_tests: bool = False,
     no_discovery_at_launch: bool = False,

src/ewoksserver/app/cors.py

@@ -1,6 +1,3 @@
-from typing import Dict
-from typing import Optional
-
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
@@ -16,7 +13,7 @@ def enable_cors(app: FastAPI) -> None:
     )
 
 
-def get_cors_options(app: FastAPI) -> Optional[Dict]:
+def get_cors_options(app: FastAPI) -> dict | None:
     for mw in app.user_middleware:
         if mw.cls is CORSMiddleware:
             return mw.options

src/ewoksserver/app/models.py

@@ -1,9 +1,6 @@
 import logging
 from enum import Enum
 from pathlib import Path
-from typing import Dict
-from typing import List
-from typing import Optional
 
 from pydantic import BaseModel
 from pydantic import Field
@@ -19,13 +16,13 @@ class EwoksSchedulingType(str, Enum):
 
 class EwoksDiscoverySettings(BaseModel):
     on_start_up: bool = Field(default=True, title="Discover ewoks tasks on startup")
-    timeout: Optional[float] = Field(
+    timeout: float | None = Field(
         default=None, title="Timeout for task discovery (in seconds)"
     )
 
 
 class EwoksExecutionSettings(BaseModel):
-    handlers: List[Dict] = Field(default=list(), title="Ewoks execution handlers")
+    handlers: list[dict] = Field(default=list(), title="Ewoks execution handlers")
 
 
 class EwoksJobSettings(BaseModel):

src/ewoksserver/app/routes/backend.py

@@ -1,11 +1,6 @@
 from dataclasses import dataclass
 from typing import Callable
-from typing import Dict
-from typing import List
 from typing import Mapping
-from typing import Set
-from typing import Tuple
-from typing import Union
 
 from fastapi import APIRouter
 from fastapi import FastAPI
@@ -14,8 +9,8 @@
 from . import BACKEND_PREFIX
 
 AppGenerator = Callable[[], ASGIApp]
-RouterType = Union[APIRouter, AppGenerator]
-VersionTuple = Tuple[int, int, int]
+RouterType = APIRouter | AppGenerator
+VersionTuple = tuple[int, int, int]
 
 
 @dataclass
@@ -27,8 +22,8 @@ class Route:
 
 
 def get_routes(
-    tag: str, routers: Dict[VersionTuple, RouterType], suffix: str = ""
-) -> Dict[Tuple[int], Route]:
+    tag: str, routers: dict[VersionTuple, RouterType], suffix: str = ""
+) -> dict[tuple[int], Route]:
     """Generate routes with versioned paths for all strict and major versions.
     In addition add a route with non-versioned path for the latest version."""
     routes = dict()
@@ -82,7 +77,7 @@ def assert_route_versions(*all_routes: Mapping[VersionTuple, RouterType]) -> Non
     assert len(versions) == 1, "Not all routes have the same versions"
 
 
-def extract_version_tags(all_routes: List[Dict[VersionTuple, Route]]) -> Set[str]:
+def extract_version_tags(all_routes: list[dict[VersionTuple, Route]]) -> set[str]:
     """Extract all version tags"""
     tags = set()
     for routes in all_routes:
@@ -92,14 +87,14 @@ def extract_version_tags(all_routes: List[Dict[VersionTuple, Route]]) -> Set[str
     return tags
 
 
-def extract_latest_version(all_routes: List[Dict[VersionTuple, Route]]) -> VersionTuple:
+def extract_latest_version(all_routes: list[dict[VersionTuple, Route]]) -> VersionTuple:
     """Extract the latest version"""
     return max(sorted(routes)[-1][:3] for routes in all_routes)
 
 
 def add_routes(
     app: FastAPI,
-    all_routes: List[Dict[VersionTuple, Route]],
+    all_routes: list[dict[VersionTuple, Route]],
     no_older_versions: bool = False,
 ) -> None:
     """Add routes to a fastapi app"""

src/ewoksserver/app/routes/execution/events.py

@@ -1,7 +1,6 @@
 import logging
 from contextlib import contextmanager
 from typing import Generator
-from typing import Optional
 
 from ewoksjob.events.readers import EwoksEventReader
 from ewoksjob.events.readers import instantiate_reader
@@ -14,7 +13,7 @@
 @contextmanager
 def reader_context(
     ewoks_settings: EwoksSettingsType,
-) -> Generator[Optional[EwoksEventReader], None, None]:
+) -> Generator[EwoksEventReader | None, None, None]:
     r = _reader(ewoks_settings)
     try:
         yield r
@@ -23,7 +22,7 @@ def reader_context(
             r.close()
 
 
-def _reader(ewoks_settings: EwoksSettingsType) -> Optional[EwoksEventReader]:
+def _reader(ewoks_settings: EwoksSettingsType) -> EwoksEventReader | None:
     handlers = ewoks_settings.ewoks_execution.handlers
     argmap = {"uri": "url"}
     for name in ("Redis", "Sqlite3", None):

src/ewoksserver/app/routes/execution/models.py

@@ -1,24 +1,21 @@
-from typing import Dict
-from typing import List
-from typing import Optional
 from typing import Union
 
 from pydantic import BaseModel
 from pydantic import Field
 
 
 class EwoksExecuteOptions_v1(BaseModel):
-    execute_arguments: Optional[Dict] = Field(
+    execute_arguments: dict | None = Field(
         title="Workflow execution options", default=None
     )
-    worker_options: Optional[Dict] = Field(title="Worker options", default=None)
+    worker_options: dict | None = Field(title="Worker options", default=None)
 
 
 class EwoksExecuteOptions_v2(BaseModel):
-    execute_arguments: Optional[Dict] = Field(
+    execute_arguments: dict | None = Field(
         title="Workflow execution options", default=None
     )
-    submit_arguments: Optional[Dict] = Field(
+    submit_arguments: dict | None = Field(
         title="Workflow submission options", default=None
     )
 
@@ -32,93 +29,91 @@ class EwoksEvent_v1(BaseModel):
     process_id: int = Field(title="Process ID where the job was executed")
     user_name: str = Field(title="User name under which the job was executed")
     job_id: str = Field(title="Workflow execution job identifier")
-    binding: Optional[str] = Field(title="Workflow execution engine", default=None)
+    binding: str | None = Field(title="Workflow execution engine", default=None)
     context: str = Field(title="Event context (job, workflow, node)")
-    workflow_id: Optional[str] = Field(title="Workflow identifier", default=None)
-    node_id: Optional[str] = Field(title="Workflow node identifier", default=None)
-    task_id: Optional[str] = Field(title="Workflow task identifier", default=None)
+    workflow_id: str | None = Field(title="Workflow identifier", default=None)
+    node_id: str | None = Field(title="Workflow node identifier", default=None)
+    task_id: str | None = Field(title="Workflow task identifier", default=None)
     type: str = Field(title="Event type (start, end, progress)")
     time: str = Field(title="Event context send time")
-    error: Optional[bool] = Field(title="Workflow execution failed", default=None)
-    error_message: Optional[str] = Field(
+    error: bool | None = Field(title="Workflow execution failed", default=None)
+    error_message: str | None = Field(
         title="Workflow execution error message", default=None
     )
-    error_traceback: Optional[str] = Field(
+    error_traceback: str | None = Field(
         title="Workflow execution error traceback", default=None
     )
-    progress: Optional[int] = Field(title="Task progress in percentage", default=None)
-    task_uri: Optional[str] = Field(title="Workflow task output URI", default=None)
-    input_uris: Optional[List[Dict]] = Field(
+    progress: int | None = Field(title="Task progress in percentage", default=None)
+    task_uri: str | None = Field(title="Workflow task output URI", default=None)
+    input_uris: list[dict] | None = Field(
         title="Workflow task input URIs", default=None
     )
-    output_uris: Optional[List[Dict]] = Field(
+    output_uris: list[dict] | None = Field(
         title="Workflow task output URIs", default=None
     )
 
 
 class EwoksEventFilter(BaseModel):
-    user_name: Optional[str] = Field(
+    user_name: str | None = Field(
         title="User name under which the job was executed", default=None
     )
-    job_id: Optional[str] = Field(
-        title="Workflow execution job identifier", default=None
-    )
-    context: Optional[str] = Field(
+    job_id: str | None = Field(title="Workflow execution job identifier", default=None)
+    context: str | None = Field(
         title="Event context (job, workflow, node)", default=None
     )
-    workflow_id: Optional[str] = Field(title="Workflow identifier", default=None)
-    node_id: Optional[str] = Field(title="Workflow node identifier", default=None)
-    task_id: Optional[str] = Field(title="Workflow task identifier", default=None)
-    type: Optional[str] = Field(title="Event type (start, end, progress)", default=None)
-    starttime: Optional[str] = Field(title="Only events after this time", default=None)
-    endtime: Optional[str] = Field(title="Only events before this time", default=None)
-    error: Optional[bool] = Field(title="Workflow execution failed", default=None)
+    workflow_id: str | None = Field(title="Workflow identifier", default=None)
+    node_id: str | None = Field(title="Workflow node identifier", default=None)
+    task_id: str | None = Field(title="Workflow task identifier", default=None)
+    type: str | None = Field(title="Event type (start, end, progress)", default=None)
+    starttime: str | None = Field(title="Only events after this time", default=None)
+    endtime: str | None = Field(title="Only events before this time", default=None)
+    error: bool | None = Field(title="Workflow execution failed", default=None)
 
 
 class EwoksEventList_v1(BaseModel):
-    jobs: List[List[EwoksEvent_v1]] = Field(
+    jobs: list[list[EwoksEvent_v1]] = Field(
         title="Workflow execution jobs grouped per job ID"
     )
 
 
 class EwoksWorkerList(BaseModel):
-    workers: Optional[List[str]] = Field(title="Available workers for execution")
+    workers: list[str] | None = Field(title="Available workers for execution")
 
 
 class EwoksQueueList(BaseModel):
-    queues: Optional[List[str]] = Field(title="Available queues for execution")
+    queues: list[str] | None = Field(title="Available queues for execution")
 
 
 class EwoksEvent_v2(BaseModel):
     host_name: str = Field(title="Host where the job was executed")
     process_id: int = Field(title="Process ID where the job was executed")
     user_name: str = Field(title="User name under which the job was executed")
     job_id: str = Field(title="Workflow execution job identifier")
-    engine: Optional[str] = Field(title="Workflow execution engine", default=None)
+    engine: str | None = Field(title="Workflow execution engine", default=None)
     context: str = Field(title="Event context (job, workflow, node)")
-    workflow_id: Optional[str] = Field(title="Workflow identifier", default=None)
-    node_id: Optional[str] = Field(title="Workflow node identifier", default=None)
-    task_id: Optional[str] = Field(title="Workflow task identifier", default=None)
+    workflow_id: str | None = Field(title="Workflow identifier", default=None)
+    node_id: str | None = Field(title="Workflow node identifier", default=None)
+    task_id: str | None = Field(title="Workflow task identifier", default=None)
     type: str = Field(title="Event type (start, end, progress)")
     time: str = Field(title="Event context send time")
-    error: Optional[bool] = Field(title="Workflow execution failed", default=None)
-    error_message: Optional[str] = Field(
+    error: bool | None = Field(title="Workflow execution failed", default=None)
+    error_message: str | None = Field(
         title="Workflow execution error message", default=None
     )
-    error_traceback: Optional[str] = Field(
+    error_traceback: str | None = Field(
         title="Workflow execution error traceback", default=None
     )
-    progress: Optional[int] = Field(title="Task progress in percentage", default=None)
-    task_uri: Optional[str] = Field(title="Workflow task output URI", default=None)
-    input_uris: Optional[List[Dict]] = Field(
+    progress: int | None = Field(title="Task progress in percentage", default=None)
+    task_uri: str | None = Field(title="Workflow task output URI", default=None)
+    input_uris: list[dict] | None = Field(
         title="Workflow task input URIs", default=None
     )
-    output_uris: Optional[List[Dict]] = Field(
+    output_uris: list[dict] | None = Field(
         title="Workflow task output URIs", default=None
     )
 
 
 class EwoksEventList_v2(BaseModel):
-    jobs: List[List[EwoksEvent_v2]] = Field(
+    jobs: list[list[EwoksEvent_v2]] = Field(
         title="Workflow execution jobs grouped per job ID"
     )