From 3320d53eda370ab3c981d3a15350806edc1ccae2 Mon Sep 17 00:00:00 2001 From: ducklet Date: Thu, 2 Feb 2023 23:46:02 +0100 Subject: [PATCH] use native union type syntax --- unwind/db.py | 12 ++++++------ unwind/imdb.py | 3 +-- unwind/imdb_import.py | 10 +++++----- unwind/models.py | 29 ++++++++++++++--------------- unwind/request.py | 6 +++--- unwind/types.py | 4 ++-- unwind/web.py | 6 +++--- unwind/web_models.py | 18 +++++++++--------- 8 files changed, 43 insertions(+), 45 deletions(-) diff --git a/unwind/db.py b/unwind/db.py index 13217e9..bff8c20 100644 --- a/unwind/db.py +++ b/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, } diff --git a/unwind/imdb.py b/unwind/imdb.py index e541277..9288e6f 100644 --- a/unwind/imdb.py +++ b/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) diff --git a/unwind/imdb_import.py b/unwind/imdb_import.py index 45360aa..61a892c 100644 --- a/unwind/imdb_import.py +++ b/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): diff --git a/unwind/models.py b/unwind/models.py index 37cd48d..674337d 100644 --- a/unwind/models.py +++ b/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: diff --git a/unwind/request.py b/unwind/request.py index 81f9f29..0b6e07c 100644 --- a/unwind/request.py +++ b/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, diff --git a/unwind/types.py b/unwind/types.py index a54e0ec..94c0e00 100644 --- a/unwind/types.py +++ b/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.") diff --git a/unwind/web.py b/unwind/web.py index e194c10..b8705a1 100644 --- a/unwind/web.py +++ b/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 diff --git a/unwind/web_models.py b/unwind/web_models.py index 06bcb8c..e514c5f 100644 --- a/unwind/web_models.py +++ b/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