Modernize type hints

exodus-gw uses python 3.11. From python 3.10 onwards, type hints can be
simplified in various ways, but this was not taken advantage of here.
Update all the code to consistently use a more modern form:

- container classes like typing.List are no longer needed, the builtin
  containers like list, dict can be used directly.

- 'Union[x, y]' can be written 'x|y' without any imports

- 'Optional[x]' can be written 'x|None' without any imports

- some features of 'typing' have been moved to 'collections.abc'

exodus_gw/auth.py

@@ -1,6 +1,5 @@
 import base64
 import logging
-from typing import List, Optional, Set
 
 from fastapi import Depends, HTTPException, Request
 from pydantic import BaseModel
@@ -14,17 +13,17 @@
 class ClientContext(BaseModel):
     """Call context data relating to service accounts / machine users."""
 
-    roles: List[str] = []
+    roles: list[str] = []
     authenticated: bool = False
-    serviceAccountId: Optional[str] = None
+    serviceAccountId: str | None = None
 
 
 class UserContext(BaseModel):
     """Call context data relating to human users."""
 
-    roles: List[str] = []
+    roles: list[str] = []
     authenticated: bool = False
-    internalUsername: Optional[str] = None
+    internalUsername: str | None = None
 
 
 class CallContext(BaseModel):
@@ -79,7 +78,7 @@ async def caller_name(context: CallContext = Depends(call_context)) -> str:
 
 async def caller_roles(
     context: CallContext = Depends(call_context),
-) -> Set[str]:
+) -> set[str]:
     """Returns all roles held by the caller of the current request.
 
     This will be an empty set for unauthenticated requests.
@@ -103,8 +102,8 @@ def needs_role(rolename):
 
     async def check_roles(
         request: Request,
-        env: Optional[str] = None,
-        roles: Set[str] = Depends(caller_roles),
+        env: str | None = None,
+        roles: set[str] = Depends(caller_roles),
         caller_name: str = Depends(caller_name),
     ):
         role = env + "-" + rolename if env else rolename
@@ -134,7 +133,7 @@ async def check_roles(
 
 async def log_login(
     request: Request,
-    roles: Set[str] = Depends(caller_roles),
+    roles: set[str] = Depends(caller_roles),
     caller_name: str = Depends(caller_name),
 ):
     if caller_name != "<anonymous user>":

exodus_gw/aws/dynamodb.py

@@ -3,7 +3,7 @@
 from datetime import datetime
 from itertools import islice
 from threading import Lock
-from typing import Any, Dict, List, Optional
+from typing import Any
 
 import backoff
 from botocore.exceptions import EndpointConnectionError
@@ -22,8 +22,8 @@ def __init__(
         env: str,
         settings: Settings,
         from_date: str,
-        env_obj: Optional[Environment] = None,
-        deadline: Optional[datetime] = None,
+        env_obj: Environment | None = None,
+        deadline: datetime | None = None,
     ):
         self.env = env
         self.settings = settings
@@ -45,12 +45,12 @@ def definitions(self):
                     self._definitions = self.query_definitions()
         return self._definitions
 
-    def query_definitions(self) -> Dict[str, Any]:
+    def query_definitions(self) -> dict[str, Any]:
         """Query the definitions in the config_table. If definitions are found, return them. Otherwise,
         return an empty dictionary."""
 
         # Return an empty dict if a query result is not found
-        out: Dict[str, Any] = {}
+        out: dict[str, Any] = {}
 
         LOG.info(
             "Loading exodus-config as at %s.",
@@ -75,12 +75,12 @@ def query_definitions(self) -> Dict[str, Any]:
 
     def create_request(
         self,
-        items: List[models.Item],
+        items: list[models.Item],
         delete: bool = False,
     ):
         """Create the dictionary structure expected by batch_write_item."""
         table_name = self.env_obj.table
-        request: Dict[str, List[Any]] = {table_name: []}
+        request: dict[str, list[Any]] = {table_name: []}
 
         uri_aliases = []
         for k, v in self.definitions.items():
@@ -140,7 +140,7 @@ def create_config_request(self, config):
         }
         return request
 
-    def batch_write(self, request: Dict[str, Any]):
+    def batch_write(self, request: dict[str, Any]):
         """Wrapper for batch_write_item with retries and item count validation.
 
         Item limit of 25 is, at this time, imposed by AWS's boto3 library.
@@ -196,15 +196,15 @@ def _batch_write(req):
 
         return _batch_write(request)
 
-    def get_batches(self, items: List[models.Item]):
+    def get_batches(self, items: list[models.Item]):
         """Divide the publish items into batches of size 'write_batch_size'."""
         it = iter(items)
         batches = list(
             iter(lambda: tuple(islice(it, self.settings.write_batch_size)), ())
         )
         return batches
 
-    def write_batch(self, items: List[models.Item], delete: bool = False):
+    def write_batch(self, items: list[models.Item], delete: bool = False):
         """Submit a batch of given items for writing via batch_write."""
 
         request = self.create_request(list(items), delete)

exodus_gw/aws/log.py

@@ -1,7 +1,7 @@
 """AWS logging utilities."""
 
 import logging
-from typing import Any, Optional, Union
+from typing import Any
 
 import aioboto3
 import boto3.session
@@ -27,9 +27,9 @@ def request_logger(request: AWSPreparedRequest, **_kwargs):
 
 
 def response_logger(
-    response: Optional[tuple[AWSResponse, Any]],
+    response: tuple[AWSResponse, Any] | None,
     request_dict: dict[str, Any],
-    caught_exception: Optional[Exception],
+    caught_exception: Exception | None,
     **_kwargs
 ):
     # Callback for logging responses from AWS.
@@ -55,7 +55,7 @@ def response_logger(
     )
 
 
-def add_loggers(session: Union[boto3.session.Session, aioboto3.Session]):
+def add_loggers(session: boto3.session.Session | aioboto3.Session):
     """Add some custom loggers onto a boto session."""
 
     # Log just before we send requests.

exodus_gw/aws/util.py

@@ -1,7 +1,7 @@
 import io
 import logging
 import re
-from typing import AnyStr, Dict
+from typing import AnyStr
 from xml.etree.ElementTree import Element, ElementTree, SubElement
 
 from defusedxml.ElementTree import fromstring
@@ -27,7 +27,7 @@ def extract_request_metadata(request: Request, settings: Settings):
     return metadata
 
 
-def validate_metadata(metadata: Dict[str, str], settings: Settings):
+def validate_metadata(metadata: dict[str, str], settings: Settings):
     valid_meta_fields = settings.upload_meta_fields
     for k, v in metadata.items():
         if k not in valid_meta_fields.keys():

exodus_gw/dramatiq/middleware/scheduler.py

@@ -3,7 +3,6 @@
 import uuid
 from collections.abc import Callable
 from datetime import datetime
-from typing import Optional
 
 import pycron
 from dramatiq import Middleware
@@ -78,7 +77,7 @@ def __wrap_with_schedule(self, broker, actor):
         actor.options["unscheduled_fn"] = unscheduled_fn
 
         @functools.wraps(unscheduled_fn)
-        def new_fn(last_run: Optional[float] = None):
+        def new_fn(last_run: float | None = None):
             cron_rule = getattr(settings, settings_key)
             now = datetime.utcnow()
 

exodus_gw/migrations/test.py

@@ -2,10 +2,10 @@
 
 import functools
 import os
-import typing
+from collections.abc import Callable
 
 
-def tested_by(data_fn: typing.Callable[[], None]):
+def tested_by(data_fn: Callable[[], None]):
     """A decorator declaring a function which provides test data for
     an upgrade or downgrade operation.
 
@@ -29,7 +29,7 @@ def tested_by(data_fn: typing.Callable[[], None]):
     that migration can complete without crashing.
     """
 
-    def decorator(fn: typing.Callable[[], None]):
+    def decorator(fn: Callable[[], None]):
         @functools.wraps(fn)
         def fn_with_data():
             # Call the data function before the real migration function,

exodus_gw/models/dramatiq.py

@@ -1,5 +1,5 @@
 import datetime
-from typing import Any, Dict, Optional
+from typing import Any
 
 from sqlalchemy import DateTime, String
 from sqlalchemy.dialects.postgresql import JSONB
@@ -25,7 +25,7 @@ class DramatiqMessage(Base):
     # Null means message is not yet assigned.
     # Not a foreign key since consumers can disappear while leaving
     # their messages behind.
-    consumer_id: Mapped[Optional[str]] = mapped_column(String)
+    consumer_id: Mapped[str | None] = mapped_column(String)
 
     consumer = relationship(
         "DramatiqConsumer",
@@ -40,7 +40,7 @@ class DramatiqMessage(Base):
     actor: Mapped[str] = mapped_column(String)
 
     # Full message body.
-    body: Mapped[Dict[str, Any]] = mapped_column(JSONB)
+    body: Mapped[dict[str, Any]] = mapped_column(JSONB)
 
 
 class DramatiqConsumer(Base):

exodus_gw/models/publish.py

@@ -1,7 +1,7 @@
 import uuid
 from collections.abc import Sequence
 from datetime import datetime
-from typing import Any, Optional, Union
+from typing import Any, Union
 
 from fastapi import HTTPException
 from sqlalchemy import (
@@ -32,13 +32,13 @@ class Publish(Base):
     )
     env: Mapped[str] = mapped_column(String)
     state: Mapped[str] = mapped_column(String)
-    updated: Mapped[Optional[datetime]] = mapped_column(DateTime())
+    updated: Mapped[datetime | None] = mapped_column(DateTime())
     items = relationship(
         "Item", back_populates="publish", cascade="all, delete-orphan"
     )
 
     def resolve_links(
-        self, ln_items: Optional[Sequence[Union[ItemBase, "Item"]]] = None
+        self, ln_items: Sequence[Union[ItemBase, "Item"]] | None = None
     ):
         """Resolve links on publish items.
 
@@ -96,7 +96,7 @@ def resolve_links(
             Item.publish_id == self.id, Item.web_uri.in_(ln_item_paths)
         )
 
-        matches: dict[str, dict[str, Optional[str]]] = {
+        matches: dict[str, dict[str, str | None]] = {
             row.match.web_uri: {
                 "object_key": row.match.object_key,
                 "content_type": row.match.content_type,
@@ -154,9 +154,9 @@ class Item(Base):
         default=lambda: str(uuid.uuid4()),
     )
     web_uri: Mapped[str] = mapped_column(String)
-    object_key: Mapped[Optional[str]] = mapped_column(String)
-    content_type: Mapped[Optional[str]] = mapped_column(String)
-    link_to: Mapped[Optional[str]] = mapped_column(String)
+    object_key: Mapped[str | None] = mapped_column(String)
+    content_type: Mapped[str | None] = mapped_column(String)
+    link_to: Mapped[str | None] = mapped_column(String)
 
     dirty: Mapped[bool] = mapped_column(Boolean, default=True)
     """True if item still needs to be written to DynamoDB."""
@@ -179,5 +179,5 @@ class Item(Base):
 
 @event.listens_for(Publish, "before_update")
 @event.listens_for(Item, "before_update")
-def set_updated(_mapper, _connection, entity: Union[Publish, Item]):
+def set_updated(_mapper, _connection, entity: Publish | Item):
     entity.updated = datetime.utcnow()

exodus_gw/models/service.py

@@ -1,6 +1,5 @@
 from datetime import datetime
 from enum import Enum
-from typing import Optional
 
 from sqlalchemy import DateTime, ForeignKey, String, event
 from sqlalchemy.orm import Mapped, mapped_column
@@ -24,8 +23,8 @@ class Task(Base):
     id: Mapped[str] = mapped_column(Uuid(as_uuid=False), primary_key=True)
     type: Mapped[str]
     state: Mapped[str] = mapped_column(String)
-    updated: Mapped[Optional[datetime]] = mapped_column(DateTime())
-    deadline: Mapped[Optional[datetime]] = mapped_column(DateTime())
+    updated: Mapped[datetime | None] = mapped_column(DateTime())
+    deadline: Mapped[datetime | None] = mapped_column(DateTime())
 
 
 class CommitTask(Task):

exodus_gw/routers/deploy.py

@@ -2,7 +2,7 @@
 
 import logging
 from datetime import datetime, timezone
-from typing import Any, Dict
+from typing import Any
 
 import jsonschema
 from fastapi import APIRouter, Body, HTTPException
@@ -113,7 +113,7 @@ def alias_schema(description):
     },
 )
 def deploy_config(
-    config: Dict[str, Any] = Body(
+    config: dict[str, Any] = Body(
         ...,
         examples=[
             {

exodus_gw/routers/publish.py

@@ -113,7 +113,6 @@
 import logging
 import os
 from datetime import datetime, timedelta
-from typing import Dict, List, Optional, Union
 from uuid import uuid4
 
 from fastapi import APIRouter, Body, HTTPException, Query
@@ -180,7 +179,7 @@ def publish(
     dependencies=[auth.needs_role("publisher")],
 )
 def update_publish_items(
-    items: List[schemas.ItemBase] = Body(
+    items: list[schemas.ItemBase] = Body(
         ...,
         examples=[
             [
@@ -209,7 +208,7 @@ def update_publish_items(
     env: Environment = deps.env,
     db: Session = deps.db,
     settings: Settings = deps.settings,
-) -> Dict[None, None]:
+) -> dict[None, None]:
     """Add publish items to an existing publish object.
 
     **Required roles**: `{env}-publisher`
@@ -256,7 +255,7 @@ def update_publish_items(
     # (2) any item we're about to save whose 'link_to' matches one of the 'web_uri'
     #     of a non-link item already in the DB.
     #
-    resolvable: list[Union[schemas.ItemBase, models.Item]] = []
+    resolvable: list[schemas.ItemBase | models.Item] = []
     resolvable.extend(items)
     resolvable.extend(
         db.query(models.Item)
@@ -383,10 +382,10 @@ def commit_publish(
     env: Environment = deps.env,
     db: Session = deps.db,
     settings: Settings = deps.settings,
-    deadline: Union[str, None] = Query(
+    deadline: str | None = Query(
         default=None, examples=["2022-07-25T15:47:47Z"]
     ),
-    commit_mode: Optional[models.CommitModes] = Query(
+    commit_mode: models.CommitModes | None = Query(
         default=None,
         title="commit mode",
         description="See: [Two-phase commit](#section/Two-phase-commit)",

exodus_gw/routers/service.py

@@ -23,7 +23,7 @@
     "/",
     include_in_schema=False,
 )
-async def redirect(accept: Optional[str] = Header(default=None)):
+async def redirect(accept: str | None = Header(default=None)):
     """Redirect from service root to API docs. For browsers only."""
 
     # We only send the redirect if it seems like the client is a browser.