use native union type syntax

unwind/db.py

@@ -4,7 +4,7 @@ import logging
 import re
 import threading
 from pathlib import Path
-from typing import Any, Iterable, Literal, Optional, Type, TypeVar, Union
+from typing import Any, Iterable, Literal, Type, TypeVar
 
 import sqlalchemy
 from databases import Database
@@ -26,7 +26,7 @@ from .types import ULID
 log = logging.getLogger(__name__)
 T = TypeVar("T")
 
-_shared_connection: Optional[Database] = None
+_shared_connection: Database | None = None
 
 
 async def open_connection_pool() -> None:
@@ -131,7 +131,7 @@ async def apply_db_patches(db: Database):
         await db.execute("vacuum")
 
 
-async def get_import_progress() -> Optional[Progress]:
+async def get_import_progress() -> Progress | None:
     """Return the latest import progress."""
     return await get(Progress, type="import-imdb-movies", order_by="started DESC")
 
@@ -244,7 +244,7 @@ ModelType = TypeVar("ModelType")
 
 async def get(
     model: Type[ModelType], *, order_by: str = None, **kwds
-) -> Optional[ModelType]:
+) -> ModelType | None:
     """Load a model instance from the database.
 
     Passing `kwds` allows to filter the instance to load. You have to encode the
@@ -415,7 +415,7 @@ async def find_ratings(
     limit_rows: int = 10,
     user_ids: Iterable[str] = [],
 ):
-    values: dict[str, Union[int, str]] = {
+    values: dict[str, int | str] = {
         "limit_rows": limit_rows,
     }
 
@@ -598,7 +598,7 @@ async def find_movies(
     include_unrated: bool = False,
     user_ids: list[ULID] = [],
 ) -> Iterable[tuple[Movie, list[Rating]]]:
-    values: dict[str, Union[int, str]] = {
+    values: dict[str, int | str] = {
         "limit_rows": limit_rows,
         "skip_rows": skip_rows,
     }

unwind/imdb.py

@@ -2,7 +2,6 @@ import logging
 import re
 from collections import namedtuple
 from datetime import datetime
-from typing import Optional, Tuple
 from urllib.parse import urljoin
 
 from . import db
@@ -153,7 +152,7 @@ def movie_and_rating_from_item(item) -> tuple[Movie, Rating]:
 ForgedRequest = namedtuple("ForgedRequest", "url headers")
 
 
-async def parse_page(url) -> Tuple[list[Rating], Optional[str]]:
+async def parse_page(url) -> tuple[list[Rating], str | None]:
     ratings = []
 
     soup = soup_from_url(url)

unwind/imdb_import.py

@@ -4,7 +4,7 @@ import logging
 from dataclasses import dataclass, fields
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Generator, Literal, Optional, Type, TypeVar, overload
+from typing import Generator, Literal, Type, TypeVar, overload
 
 from . import config, db, request
 from .db import add_or_update_many_movies
@@ -27,10 +27,10 @@ class BasicRow:
     primaryTitle: str
     originalTitle: str
     isAdult: bool
-    startYear: Optional[int]
-    endYear: Optional[int]
-    runtimeMinutes: Optional[int]
-    genres: Optional[set[str]]
+    startYear: int | None
+    endYear: int | None
+    runtimeMinutes: int | None
+    genres: set[str] | None
 
     @classmethod
     def from_row(cls, row):

unwind/models.py

@@ -9,7 +9,6 @@ from typing import (
     ClassVar,
     Container,
     Literal,
-    Optional,
     Type,
     TypeVar,
     Union,
@@ -25,7 +24,7 @@ JSONObject = dict[str, JSON]
 T = TypeVar("T")
 
 
-def annotations(tp: Type) -> Optional[tuple]:
+def annotations(tp: Type) -> tuple | None:
     return tp.__metadata__ if hasattr(tp, "__metadata__") else None
 
 
@@ -61,7 +60,7 @@ def is_optional(tp: Type) -> bool:
     return len(args) == 2 and type(None) in args
 
 
-def optional_type(tp: Type) -> Optional[Type]:
+def optional_type(tp: Type) -> Type | None:
     """Return the wrapped type from an optional type.
 
     For example this will return `int` for `Optional[int]`.
@@ -206,7 +205,7 @@ class Progress:
     type: str = None
     state: str = None
     started: datetime = field(default_factory=utcnow)
-    stopped: Optional[str] = None
+    stopped: str | None = None
 
     @property
     def _state(self) -> dict:
@@ -243,15 +242,15 @@ class Movie:
 
     id: ULID = field(default_factory=ULID)
     title: str = None  # canonical title (usually English)
-    original_title: Optional[
-        str
-    ] = None  # original title (usually transscribed to latin script)
+    original_title: str | None = (
+        None  # original title (usually transscribed to latin script)
+    )
     release_year: int = None  # canonical release date
     media_type: str = None
     imdb_id: str = None
-    imdb_score: Optional[int] = None  # range: [0,100]
-    imdb_votes: Optional[int] = None
-    runtime: Optional[int] = None  # minutes
+    imdb_score: int | None = None  # range: [0,100]
+    imdb_votes: int | None = None
+    runtime: int | None = None  # minutes
     genres: set[str] = None
     created: datetime = field(default_factory=utcnow)
     updated: datetime = field(default_factory=utcnow)
@@ -292,7 +291,7 @@ dataclass containing the ID of the linked data.
 The contents of the Relation are ignored or discarded when using
 `asplain`, `fromplain`, and `validate`.
 """
-Relation = Annotated[Optional[T], _RelationSentinel]
+Relation = Annotated[T | None, _RelationSentinel]
 
 
 @dataclass
@@ -309,8 +308,8 @@ class Rating:
 
     score: int = None  # range: [0,100]
     rating_date: datetime = None
-    favorite: Optional[bool] = None
-    finished: Optional[bool] = None
+    favorite: bool | None = None
+    finished: bool | None = None
 
     def __eq__(self, other):
         """Return wether two Ratings are equal.
@@ -342,11 +341,11 @@ class User:
     secret: str = None
     groups: list[dict[str, str]] = field(default_factory=list)
 
-    def has_access(self, group_id: Union[ULID, str], access: Access = "r"):
+    def has_access(self, group_id: ULID | str, access: Access = "r"):
         group_id = group_id if isinstance(group_id, str) else str(group_id)
         return any(g["id"] == group_id and access == g["access"] for g in self.groups)
 
-    def set_access(self, group_id: Union[ULID, str], access: Access):
+    def set_access(self, group_id: ULID | str, access: Access):
         group_id = group_id if isinstance(group_id, str) else str(group_id)
         for g in self.groups:
             if g["id"] == group_id:

unwind/request.py

@@ -11,7 +11,7 @@ from hashlib import md5
 from pathlib import Path
 from random import random
 from time import sleep, time
-from typing import Callable, Optional, Union
+from typing import Callable
 
 import bs4
 import requests
@@ -142,7 +142,7 @@ class RedirectError(RuntimeError):
         super().__init__(f"Redirected: {from_url} -> {to_url}")
 
 
-def cache_path(req) -> Optional[Path]:
+def cache_path(req) -> Path | None:
     if not config.cachedir:
         return
     sig = repr(req.url)  # + repr(sorted(req.headers.items()))
@@ -215,7 +215,7 @@ def last_modified_from_file(path: Path):
 
 def download(
     url: str,
-    file_path: Union[Path, str] = None,
+    file_path: Path | str = None,
     *,
     replace_existing: bool = None,
     only_if_newer: bool = False,

unwind/types.py

@@ -1,5 +1,5 @@
 import re
-from typing import Union, cast
+from typing import cast
 
 import ulid
 from ulid.hints import Buffer
@@ -16,7 +16,7 @@ class ULID(ulid.ULID):
 
     _pattern = re.compile(r"^[0-9A-HJKMNP-TV-Z]{26}$")
 
-    def __init__(self, buffer: Union[Buffer, ulid.ULID, str, None] = None):
+    def __init__(self, buffer: Buffer | ulid.ULID | str | None = None):
         if isinstance(buffer, str):
             if not self._pattern.search(buffer):
                 raise ValueError("Invalid ULID.")

unwind/web.py

@@ -2,7 +2,7 @@ import asyncio
 import logging
 import secrets
 from json.decoder import JSONDecodeError
-from typing import Literal, Optional, overload
+from typing import Literal, overload
 
 from starlette.applications import Starlette
 from starlette.authentication import (
@@ -97,7 +97,7 @@ def yearcomp(s: str):
     return comp, int(s)
 
 
-def as_int(x, *, max: int = None, min: Optional[int] = 1, default: int = None):
+def as_int(x, *, max: int = None, min: int | None = 1, default: int = None):
     try:
         if not isinstance(x, int):
             x = int(x)
@@ -158,7 +158,7 @@ def is_admin(request):
     return "admin" in request.auth.scopes
 
 
-async def auth_user(request) -> Optional[User]:
+async def auth_user(request) -> User | None:
     if not isinstance(request.user, AuthedUser):
         return
 

unwind/web_models.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from typing import Container, Iterable, Optional
+from typing import Container, Iterable
 
 from . import imdb, models
 
@@ -10,14 +10,14 @@ Score100 = int  # [0, 100]
 @dataclass
 class Rating:
     canonical_title: str
-    imdb_score: Optional[Score100]
-    imdb_votes: Optional[int]
+    imdb_score: Score100 | None
+    imdb_votes: int | None
     media_type: str
     movie_imdb_id: str
-    original_title: Optional[str]
+    original_title: str | None
     release_year: int
-    user_id: Optional[str]
-    user_score: Optional[Score100]
+    user_id: str | None
+    user_score: Score100 | None
 
     @classmethod
     def from_movie(cls, movie: models.Movie, *, rating: models.Rating = None):
@@ -37,11 +37,11 @@ class Rating:
 @dataclass
 class RatingAggregate:
     canonical_title: str
-    imdb_score: Optional[Score100]
-    imdb_votes: Optional[int]
+    imdb_score: Score100 | None
+    imdb_votes: int | None
     link: URL
     media_type: str
-    original_title: Optional[str]
+    original_title: str | None
     user_scores: list[Score100]
     year: int