diff --git a/beets/autotag/hooks.py b/beets/autotag/hooks.py index 75ce4a663..a458d58eb 100644 --- a/beets/autotag/hooks.py +++ b/beets/autotag/hooks.py @@ -18,17 +18,7 @@ from __future__ import annotations import re from functools import total_ordering -from typing import ( - Any, - Callable, - Iterable, - Iterator, - NamedTuple, - Optional, - TypeVar, - Union, - cast, -) +from typing import Any, Callable, Iterable, Iterator, NamedTuple, TypeVar, cast from jellyfish import levenshtein_distance from unidecode import unidecode @@ -80,46 +70,46 @@ class AlbumInfo(AttrDict): def __init__( self, tracks: list[TrackInfo], - album: Optional[str] = None, - album_id: Optional[str] = None, - artist: Optional[str] = None, - artist_id: Optional[str] = None, - artists: Optional[list[str]] = None, - artists_ids: Optional[list[str]] = None, - asin: Optional[str] = None, - albumtype: Optional[str] = None, - albumtypes: Optional[list[str]] = None, + album: str | None = None, + album_id: str | None = None, + artist: str | None = None, + artist_id: str | None = None, + artists: list[str] | None = None, + artists_ids: list[str] | None = None, + asin: str | None = None, + albumtype: str | None = None, + albumtypes: list[str] | None = None, va: bool = False, - year: Optional[int] = None, - month: Optional[int] = None, - day: Optional[int] = None, - label: Optional[str] = None, - barcode: Optional[str] = None, - mediums: Optional[int] = None, - artist_sort: Optional[str] = None, - artists_sort: Optional[list[str]] = None, - releasegroup_id: Optional[str] = None, - release_group_title: Optional[str] = None, - catalognum: Optional[str] = None, - script: Optional[str] = None, - language: Optional[str] = None, - country: Optional[str] = None, - style: Optional[str] = None, - genre: Optional[str] = None, - albumstatus: Optional[str] = None, - media: Optional[str] = None, - albumdisambig: Optional[str] = None, - releasegroupdisambig: Optional[str] = None, - artist_credit: Optional[str] = None, - artists_credit: Optional[list[str]] = None, - original_year: Optional[int] = None, - original_month: Optional[int] = None, - original_day: Optional[int] = None, - data_source: Optional[str] = None, - data_url: Optional[str] = None, - discogs_albumid: Optional[str] = None, - discogs_labelid: Optional[str] = None, - discogs_artistid: Optional[str] = None, + year: int | None = None, + month: int | None = None, + day: int | None = None, + label: str | None = None, + barcode: str | None = None, + mediums: int | None = None, + artist_sort: str | None = None, + artists_sort: list[str] | None = None, + releasegroup_id: str | None = None, + release_group_title: str | None = None, + catalognum: str | None = None, + script: str | None = None, + language: str | None = None, + country: str | None = None, + style: str | None = None, + genre: str | None = None, + albumstatus: str | None = None, + media: str | None = None, + albumdisambig: str | None = None, + releasegroupdisambig: str | None = None, + artist_credit: str | None = None, + artists_credit: list[str] | None = None, + original_year: int | None = None, + original_month: int | None = None, + original_day: int | None = None, + data_source: str | None = None, + data_url: str | None = None, + discogs_albumid: str | None = None, + discogs_labelid: str | None = None, + discogs_artistid: str | None = None, **kwargs, ): self.album = album @@ -187,38 +177,38 @@ class TrackInfo(AttrDict): # TYPING: are all of these correct? I've assumed optional strings def __init__( self, - title: Optional[str] = None, - track_id: Optional[str] = None, - release_track_id: Optional[str] = None, - artist: Optional[str] = None, - artist_id: Optional[str] = None, - artists: Optional[list[str]] = None, - artists_ids: Optional[list[str]] = None, - length: Optional[float] = None, - index: Optional[int] = None, - medium: Optional[int] = None, - medium_index: Optional[int] = None, - medium_total: Optional[int] = None, - artist_sort: Optional[str] = None, - artists_sort: Optional[list[str]] = None, - disctitle: Optional[str] = None, - artist_credit: Optional[str] = None, - artists_credit: Optional[list[str]] = None, - data_source: Optional[str] = None, - data_url: Optional[str] = None, - media: Optional[str] = None, - lyricist: Optional[str] = None, - composer: Optional[str] = None, - composer_sort: Optional[str] = None, - arranger: Optional[str] = None, - track_alt: Optional[str] = None, - work: Optional[str] = None, - mb_workid: Optional[str] = None, - work_disambig: Optional[str] = None, - bpm: Optional[str] = None, - initial_key: Optional[str] = None, - genre: Optional[str] = None, - album: Optional[str] = None, + title: str | None = None, + track_id: str | None = None, + release_track_id: str | None = None, + artist: str | None = None, + artist_id: str | None = None, + artists: list[str] | None = None, + artists_ids: list[str] | None = None, + length: float | None = None, + index: int | None = None, + medium: int | None = None, + medium_index: int | None = None, + medium_total: int | None = None, + artist_sort: str | None = None, + artists_sort: list[str] | None = None, + disctitle: str | None = None, + artist_credit: str | None = None, + artists_credit: list[str] | None = None, + data_source: str | None = None, + data_url: str | None = None, + media: str | None = None, + lyricist: str | None = None, + composer: str | None = None, + composer_sort: str | None = None, + arranger: str | None = None, + track_alt: str | None = None, + work: str | None = None, + mb_workid: str | None = None, + work_disambig: str | None = None, + bpm: str | None = None, + initial_key: str | None = None, + genre: str | None = None, + album: str | None = None, **kwargs, ): self.title = title @@ -298,7 +288,7 @@ def _string_dist_basic(str1: str, str2: str) -> float: return levenshtein_distance(str1, str2) / float(max(len(str1), len(str2))) -def string_dist(str1: Optional[str], str2: Optional[str]) -> float: +def string_dist(str1: str | None, str2: str | None) -> float: """Gives an "intuitive" edit distance between two strings. This is an edit distance, normalized by the string length, with a number of tweaks that reflect intuition about text. @@ -474,7 +464,7 @@ class Distance: # Adding components. - def _eq(self, value1: Union[re.Pattern[str], Any], value2: Any) -> bool: + def _eq(self, value1: re.Pattern[str] | Any, value2: Any) -> bool: """Returns True if `value1` is equal to `value2`. `value1` may be a compiled regular expression, in which case it will be matched against `value2`. @@ -498,7 +488,7 @@ class Distance: self, key: str, value: Any, - options: Union[list[Any], tuple[Any, ...], Any], + options: list[Any] | tuple[Any, ...] | Any, ): """Adds a distance penalty of 1.0 if `value` doesn't match any of the values in `options`. If an option is a compiled regular @@ -541,7 +531,7 @@ class Distance: self, key: str, value: Any, - options: Union[list[Any], tuple[Any, ...], Any], + options: list[Any] | tuple[Any, ...] | Any, ): """Adds a distance penalty that corresponds to the position at which `value` appears in `options`. A distance penalty of 0.0 @@ -563,8 +553,8 @@ class Distance: def add_ratio( self, key: str, - number1: Union[int, float], - number2: Union[int, float], + number1: int | float, + number2: int | float, ): """Adds a distance penalty for `number1` as a ratio of `number2`. `number1` is bound at 0 and `number2`. @@ -576,7 +566,7 @@ class Distance: dist = 0.0 self.add(key, dist) - def add_string(self, key: str, str1: Optional[str], str2: Optional[str]): + def add_string(self, key: str, str1: str | None, str2: str | None): """Adds a distance penalty based on the edit distance between `str1` and `str2`. """ @@ -603,7 +593,7 @@ class TrackMatch(NamedTuple): # Aggregation of sources. -def album_for_mbid(release_id: str) -> Optional[AlbumInfo]: +def album_for_mbid(release_id: str) -> AlbumInfo | None: """Get an AlbumInfo object for a MusicBrainz release ID. Return None if the ID is not found. """ @@ -617,7 +607,7 @@ def album_for_mbid(release_id: str) -> Optional[AlbumInfo]: return None -def track_for_mbid(recording_id: str) -> Optional[TrackInfo]: +def track_for_mbid(recording_id: str) -> TrackInfo | None: """Get a TrackInfo object for a MusicBrainz recording ID. Return None if the ID is not found. """ diff --git a/beets/autotag/match.py b/beets/autotag/match.py index 75f8e91e8..06893f5ca 100644 --- a/beets/autotag/match.py +++ b/beets/autotag/match.py @@ -21,16 +21,7 @@ from __future__ import annotations import datetime import re from enum import IntEnum -from typing import ( - Any, - Iterable, - NamedTuple, - Optional, - Sequence, - TypeVar, - Union, - cast, -) +from typing import Any, Iterable, NamedTuple, Sequence, TypeVar, Union, cast from munkres import Munkres @@ -474,8 +465,8 @@ def _add_candidate( def tag_album( items, - search_artist: Optional[str] = None, - search_album: Optional[str] = None, + search_artist: str | None = None, + search_album: str | None = None, search_ids: list[str] = [], ) -> tuple[str, str, Proposal]: """Return a tuple of the current artist name, the current album @@ -566,9 +557,9 @@ def tag_album( def tag_item( item, - search_artist: Optional[str] = None, - search_title: Optional[str] = None, - search_ids: Optional[list[str]] = None, + search_artist: str | None = None, + search_title: str | None = None, + search_ids: list[str] | None = None, ) -> Proposal: """Find metadata for a single track. Return a `Proposal` consisting of `TrackMatch` objects. @@ -581,7 +572,7 @@ def tag_item( # Holds candidates found so far: keys are MBIDs; values are # (distance, TrackInfo) pairs. candidates = {} - rec: Optional[Recommendation] = None + rec: Recommendation | None = None # First, try matching by MusicBrainz ID. trackids = search_ids or [t for t in [item.mb_trackid] if t] diff --git a/beets/autotag/mb.py b/beets/autotag/mb.py index 0ecbcce0a..084610772 100644 --- a/beets/autotag/mb.py +++ b/beets/autotag/mb.py @@ -20,7 +20,7 @@ import re import traceback from collections import Counter from itertools import product -from typing import Any, Iterator, Optional, Sequence, cast +from typing import Any, Iterator, Sequence, cast from urllib.parse import urljoin import musicbrainzngs @@ -277,10 +277,10 @@ def _get_related_artist_names(relations, relation_type): def track_info( recording: dict, - index: Optional[int] = None, - medium: Optional[int] = None, - medium_index: Optional[int] = None, - medium_total: Optional[int] = None, + index: int | None = None, + medium: int | None = None, + medium_index: int | None = None, + medium_total: int | None = None, ) -> beets.autotag.hooks.TrackInfo: """Translates a MusicBrainz recording result dictionary into a beets ``TrackInfo`` object. Three parameters are optional and are used @@ -661,8 +661,8 @@ def album_info(release: dict) -> beets.autotag.hooks.AlbumInfo: def match_album( artist: str, album: str, - tracks: Optional[int] = None, - extra_tags: Optional[dict[str, Any]] = None, + tracks: int | None = None, + extra_tags: dict[str, Any] | None = None, ) -> Iterator[beets.autotag.hooks.AlbumInfo]: """Searches for a single album ("release" in MusicBrainz parlance) and returns an iterator over AlbumInfo objects. May raise a @@ -739,7 +739,7 @@ def match_track( yield track_info(recording) -def _parse_id(s: str) -> Optional[str]: +def _parse_id(s: str) -> str | None: """Search for a MusicBrainz ID in the given string and return it. If no ID can be found, return None. """ @@ -757,7 +757,7 @@ def _is_translation(r): def _find_actual_release_from_pseudo_release( pseudo_rel: dict, -) -> Optional[dict]: +) -> dict | None: try: relations = pseudo_rel["release"]["release-relation-list"] except KeyError: @@ -776,7 +776,7 @@ def _find_actual_release_from_pseudo_release( def _merge_pseudo_and_actual_album( pseudo: beets.autotag.hooks.AlbumInfo, actual: beets.autotag.hooks.AlbumInfo -) -> Optional[beets.autotag.hooks.AlbumInfo]: +) -> beets.autotag.hooks.AlbumInfo | None: """ Merges a pseudo release with its actual release. @@ -814,7 +814,7 @@ def _merge_pseudo_and_actual_album( return merged -def album_for_id(releaseid: str) -> Optional[beets.autotag.hooks.AlbumInfo]: +def album_for_id(releaseid: str) -> beets.autotag.hooks.AlbumInfo | None: """Fetches an album by its MusicBrainz ID and returns an AlbumInfo object or None if the album is not found. May raise a MusicBrainzAPIError. @@ -852,7 +852,7 @@ def album_for_id(releaseid: str) -> Optional[beets.autotag.hooks.AlbumInfo]: return release -def track_for_id(releaseid: str) -> Optional[beets.autotag.hooks.TrackInfo]: +def track_for_id(releaseid: str) -> beets.autotag.hooks.TrackInfo | None: """Fetches a track by its MusicBrainz ID. Returns a TrackInfo object or None if no track is found. May raise a MusicBrainzAPIError. """ diff --git a/beets/dbcore/db.py b/beets/dbcore/db.py index d7374e89c..407983d24 100755 --- a/beets/dbcore/db.py +++ b/beets/dbcore/db.py @@ -35,10 +35,8 @@ from typing import ( Iterable, Iterator, Mapping, - Optional, Sequence, TypeVar, - Union, cast, ) @@ -115,7 +113,7 @@ class FormattedMapping(Mapping[str, str]): def get( # type: ignore self, key: str, - default: Optional[str] = None, + default: str | None = None, ) -> str: """Similar to Mapping.get(key, default), but always formats to str.""" if default is None: @@ -215,7 +213,7 @@ class LazyConvertDict: for key in self: yield key, self[key] - def get(self, key: str, default: Optional[Any] = None): + def get(self, key: str, default: Any | None = None): """Get the value for a given key or `default` if it does not exist. """ @@ -358,7 +356,7 @@ class Model(ABC): # Basic operation. - def __init__(self, db: Optional[Database] = None, **values): + def __init__(self, db: Database | None = None, **values): """Create a new object with an optional Database association and initial field values. """ @@ -374,7 +372,7 @@ class Model(ABC): @classmethod def _awaken( cls: type[AnyModel], - db: Optional[Database] = None, + db: Database | None = None, fixed_values: dict[str, Any] = {}, flex_values: dict[str, Any] = {}, ) -> AnyModel: @@ -574,7 +572,7 @@ class Model(ABC): # Database interaction (CRUD methods). - def store(self, fields: Optional[Iterable[str]] = None): + def store(self, fields: Iterable[str] | None = None): """Save the object's metadata into the library database. :param fields: the fields to be stored. If not specified, all fields will be. @@ -648,7 +646,7 @@ class Model(ABC): f"DELETE FROM {self._flex_table} WHERE entity_id=?", (self.id,) ) - def add(self, db: Optional[Database] = None): + def add(self, db: Database | None = None): """Add the object to the library database. This object must be associated with a database; you can provide one via the `db` parameter or use the currently associated database. @@ -687,7 +685,7 @@ class Model(ABC): def evaluate_template( self, - template: Union[str, functemplate.Template], + template: str | functemplate.Template, for_path: bool = False, ) -> str: """Evaluate a template (a string or a `Template` object) using @@ -763,7 +761,7 @@ class Results(Generic[AnyModel]): rows: list[Mapping], db: Database, flex_rows, - query: Optional[Query] = None, + query: Query | None = None, sort=None, ): """Create a result set that will construct objects of type @@ -907,7 +905,7 @@ class Results(Generic[AnyModel]): except StopIteration: raise IndexError(f"result index {n} out of range") - def get(self) -> Optional[AnyModel]: + def get(self) -> AnyModel | None: """Return the first matching object, or None if no objects match. """ @@ -1105,7 +1103,7 @@ class Database: value = value.decode() return re.search(pattern, str(value)) is not None - def bytelower(bytestring: Optional[AnyStr]) -> Optional[AnyStr]: + def bytelower(bytestring: AnyStr | None) -> AnyStr | None: """A custom ``bytelower`` sqlite function so we can compare bytestrings in a semi case insensitive fashion. @@ -1227,8 +1225,8 @@ class Database: def _fetch( self, model_cls: type[AnyModel], - query: Optional[Query] = None, - sort: Optional[Sort] = None, + query: Query | None = None, + sort: Sort | None = None, ) -> Results[AnyModel]: """Fetch the objects of type `model_cls` matching the given query. The query may be given as a string, string sequence, a @@ -1286,7 +1284,7 @@ class Database: self, model_cls: type[AnyModel], id, - ) -> Optional[AnyModel]: + ) -> AnyModel | None: """Get a Model object by its id or None if the id does not exist. """ diff --git a/beets/dbcore/query.py b/beets/dbcore/query.py index 6a84d8396..e1e4a98e0 100644 --- a/beets/dbcore/query.py +++ b/beets/dbcore/query.py @@ -29,7 +29,6 @@ from typing import ( Generic, Iterator, MutableSequence, - Optional, Pattern, Sequence, TypeVar, @@ -83,7 +82,7 @@ class Query(ABC): """Return a set with field names that this query operates on.""" return set() - def clause(self) -> tuple[Optional[str], Sequence[Any]]: + def clause(self) -> tuple[str | None, Sequence[Any]]: """Generate an SQLite expression implementing the query. Return (clause, subvals) where clause is a valid sqlite @@ -149,7 +148,7 @@ class FieldQuery(Query, Generic[P]): def col_clause(self) -> tuple[str, Sequence[SQLiteType]]: return self.field, () - def clause(self) -> tuple[Optional[str], Sequence[SQLiteType]]: + def clause(self) -> tuple[str | None, Sequence[SQLiteType]]: if self.fast: return self.col_clause() else: @@ -329,7 +328,7 @@ class BytesQuery(FieldQuery[bytes]): `MatchQuery` when matching on BLOB values. """ - def __init__(self, field_name: str, pattern: Union[bytes, str, memoryview]): + def __init__(self, field_name: str, pattern: bytes | str | memoryview): # Use a buffer/memoryview representation of the pattern for SQLite # matching. This instructs SQLite to treat the blob as binary # rather than encoded Unicode. @@ -364,7 +363,7 @@ class NumericQuery(FieldQuery[str]): a float. """ - def _convert(self, s: str) -> Union[float, int, None]: + def _convert(self, s: str) -> float | int | None: """Convert a string to a numeric type (float or int). Return None if `s` is empty. @@ -481,7 +480,7 @@ class CollectionQuery(Query): def clause_with_joiner( self, joiner: str, - ) -> tuple[Optional[str], Sequence[SQLiteType]]: + ) -> tuple[str | None, Sequence[SQLiteType]]: """Return a clause created by joining together the clauses of all subqueries with the string joiner (padded by spaces). """ @@ -532,7 +531,7 @@ class AnyFieldQuery(CollectionQuery): # TYPING ERROR super().__init__(subqueries) - def clause(self) -> tuple[Optional[str], Sequence[SQLiteType]]: + def clause(self) -> tuple[str | None, Sequence[SQLiteType]]: return self.clause_with_joiner("or") def match(self, obj: Model) -> bool: @@ -571,7 +570,7 @@ class MutableCollectionQuery(CollectionQuery): class AndQuery(MutableCollectionQuery): """A conjunction of a list of other queries.""" - def clause(self) -> tuple[Optional[str], Sequence[SQLiteType]]: + def clause(self) -> tuple[str | None, Sequence[SQLiteType]]: return self.clause_with_joiner("and") def match(self, obj: Model) -> bool: @@ -581,7 +580,7 @@ class AndQuery(MutableCollectionQuery): class OrQuery(MutableCollectionQuery): """A conjunction of a list of other queries.""" - def clause(self) -> tuple[Optional[str], Sequence[SQLiteType]]: + def clause(self) -> tuple[str | None, Sequence[SQLiteType]]: return self.clause_with_joiner("or") def match(self, obj: Model) -> bool: @@ -601,7 +600,7 @@ class NotQuery(Query): def __init__(self, subquery): self.subquery = subquery - def clause(self) -> tuple[Optional[str], Sequence[SQLiteType]]: + def clause(self) -> tuple[str | None, Sequence[SQLiteType]]: clause, subvals = self.subquery.clause() if clause: return f"not ({clause})", subvals @@ -646,7 +645,7 @@ class FalseQuery(Query): # Time/date queries. -def _parse_periods(pattern: str) -> tuple[Optional[Period], Optional[Period]]: +def _parse_periods(pattern: str) -> tuple[Period | None, Period | None]: """Parse a string containing two dates separated by two dots (..). Return a pair of `Period` objects. """ @@ -692,7 +691,7 @@ class Period: self.precision = precision @classmethod - def parse(cls: type[Period], string: str) -> Optional[Period]: + def parse(cls: type[Period], string: str) -> Period | None: """Parse a date and return a `Period` object or `None` if the string is empty, or raise an InvalidQueryArgumentValueError if the string cannot be parsed to a date. @@ -711,7 +710,7 @@ class Period: def find_date_and_format( string: str, - ) -> Union[tuple[None, None], tuple[datetime, int]]: + ) -> tuple[None, None] | tuple[datetime, int]: for ord, format in enumerate(cls.date_formats): for format_option in format: try: @@ -725,7 +724,7 @@ class Period: if not string: return None - date: Optional[datetime] + date: datetime | None # Check for a relative date. match_dq = re.match(cls.relative_re, string) @@ -785,7 +784,7 @@ class DateInterval: A right endpoint of None means towards infinity. """ - def __init__(self, start: Optional[datetime], end: Optional[datetime]): + def __init__(self, start: datetime | None, end: datetime | None): if start is not None and end is not None and not start < end: raise ValueError( "start date {} is not before end date {}".format(start, end) @@ -796,8 +795,8 @@ class DateInterval: @classmethod def from_periods( cls, - start: Optional[Period], - end: Optional[Period], + start: Period | None, + end: Period | None, ) -> DateInterval: """Create an interval with two Periods as the endpoints.""" end_date = end.open_right_endpoint() if end is not None else None @@ -871,7 +870,7 @@ class DurationQuery(NumericQuery): or M:SS time interval. """ - def _convert(self, s: str) -> Optional[float]: + def _convert(self, s: str) -> float | None: """Convert a M:SS or numeric string to a float. Return None if `s` is empty. @@ -898,7 +897,7 @@ class Sort: the database. """ - def order_clause(self) -> Optional[str]: + def order_clause(self) -> str | None: """Generates a SQL fragment to be used in a ORDER BY clause, or None if no fragment is used (i.e., this is a slow sort). """ @@ -927,7 +926,7 @@ class Sort: class MultipleSort(Sort): """Sort that encapsulates multiple sub-sorts.""" - def __init__(self, sorts: Optional[list[Sort]] = None): + def __init__(self, sorts: list[Sort] | None = None): self.sorts = sorts or [] def add_sort(self, sort: Sort): diff --git a/beets/dbcore/queryparse.py b/beets/dbcore/queryparse.py index 3efeafd6e..d35fec0e3 100644 --- a/beets/dbcore/queryparse.py +++ b/beets/dbcore/queryparse.py @@ -14,9 +14,11 @@ """Parsing of strings into DBCore queries.""" +from __future__ import annotations + import itertools import re -from typing import Collection, Optional, Sequence +from typing import Collection, Sequence from . import Model, query from .query import Sort @@ -38,7 +40,7 @@ def parse_query_part( query_classes: dict[str, type[query.FieldQuery]] = {}, prefixes: dict = {}, default_class: type[query.SubstringQuery] = query.SubstringQuery, -) -> tuple[Optional[str], str, type[query.FieldQuery], bool]: +) -> tuple[str | None, str, type[query.FieldQuery], bool]: """Parse a single *query part*, which is a chunk of a complete query string representing a single criterion. diff --git a/beets/dbcore/types.py b/beets/dbcore/types.py index b7af60a23..7c546eb92 100644 --- a/beets/dbcore/types.py +++ b/beets/dbcore/types.py @@ -14,9 +14,11 @@ """Representation of type information for DBCore model fields.""" +from __future__ import annotations + import typing from abc import ABC -from typing import Any, Generic, TypeVar, Union, cast +from typing import Any, Generic, TypeVar, cast from beets.util import str2bool @@ -69,7 +71,7 @@ class Type(ABC, Generic[T, N]): # have a field null_type similar to `model_type` and use that here. return cast(N, self.model_type()) - def format(self, value: Union[N, T]) -> str: + def format(self, value: N | T) -> str: """Given a value of this type, produce a Unicode string representing the value. This is used in template evaluation. """ @@ -83,7 +85,7 @@ class Type(ABC, Generic[T, N]): else: return str(value) - def parse(self, string: str) -> Union[T, N]: + def parse(self, string: str) -> T | N: """Parse a (possibly human-written) string and return the indicated value of this type. """ @@ -92,7 +94,7 @@ class Type(ABC, Generic[T, N]): except ValueError: return self.null - def normalize(self, value: Any) -> Union[T, N]: + def normalize(self, value: Any) -> T | N: """Given a value that will be assigned into a field of this type, normalize the value to have the appropriate type. This base implementation only reinterprets `None`. @@ -107,8 +109,8 @@ class Type(ABC, Generic[T, N]): def from_sql( self, - sql_value: Union[None, int, float, str, bytes], - ) -> Union[T, N]: + sql_value: None | int | float | str | bytes, + ) -> T | N: """Receives the value stored in the SQL backend and return the value to be stored in the model. @@ -129,7 +131,7 @@ class Type(ABC, Generic[T, N]): else: return self.normalize(sql_value) - def to_sql(self, model_value: Any) -> Union[None, int, float, str, bytes]: + def to_sql(self, model_value: Any) -> None | int | float | str | bytes: """Convert a value as stored in the model object to a value used by the database adapter. """ @@ -154,7 +156,7 @@ class BaseInteger(Type[int, N]): query = NumericQuery model_type = int - def normalize(self, value: Any) -> Union[int, N]: + def normalize(self, value: Any) -> int | N: try: return self.model_type(round(float(value))) except ValueError: @@ -183,7 +185,7 @@ class BasePaddedInt(BaseInteger[N]): def __init__(self, digits: int): self.digits = digits - def format(self, value: Union[int, N]) -> str: + def format(self, value: int | N) -> str: return "{0:0{1}d}".format(value or 0, self.digits) @@ -238,7 +240,7 @@ class BaseFloat(Type[float, N]): def __init__(self, digits: int = 1): self.digits = digits - def format(self, value: Union[float, N]) -> str: + def format(self, value: float | N) -> str: return "{0:.{1}f}".format(value or 0, self.digits) @@ -264,7 +266,7 @@ class BaseString(Type[T, N]): sql = "TEXT" query = SubstringQuery - def normalize(self, value: Any) -> Union[T, N]: + def normalize(self, value: Any) -> T | N: if value is None: return self.null else: diff --git a/beetsplug/replaygain.py b/beetsplug/replaygain.py index da214dead..137ff0e49 100644 --- a/beetsplug/replaygain.py +++ b/beetsplug/replaygain.py @@ -13,6 +13,8 @@ # included in all copies or substantial portions of the Software. +from __future__ import annotations + import collections import enum import math @@ -28,7 +30,7 @@ from dataclasses import dataclass from logging import Logger from multiprocessing.pool import ThreadPool from threading import Event, Thread -from typing import Any, Callable, Optional, Sequence, TypeVar, Union, cast +from typing import Any, Callable, Sequence, TypeVar, cast from confuse import ConfigView @@ -121,9 +123,9 @@ class RgTask: def __init__( self, items: Sequence[Item], - album: Optional[Album], + album: Album | None, target_level: float, - peak_method: Optional[PeakMethod], + peak_method: PeakMethod | None, backend_name: str, log: Logger, ): @@ -133,8 +135,8 @@ class RgTask: self.peak_method = peak_method self.backend_name = backend_name self._log = log - self.album_gain: Optional[Gain] = None - self.track_gains: Optional[list[Gain]] = None + self.album_gain: Gain | None = None + self.track_gains: list[Gain] | None = None def _store_track_gain(self, item: Item, track_gain: Gain): """Store track gain for a single item in the database.""" @@ -223,7 +225,7 @@ class R128Task(RgTask): def __init__( self, items: Sequence[Item], - album: Optional[Album], + album: Album | None, target_level: float, backend_name: str, log: Logger, @@ -396,8 +398,8 @@ class FfmpegBackend(Backend): return task def _construct_cmd( - self, item: Item, peak_method: Optional[PeakMethod] - ) -> list[Union[str, bytes]]: + self, item: Item, peak_method: PeakMethod | None + ) -> list[str | bytes]: """Construct the shell command to analyse items.""" return [ self._ffmpeg_path, @@ -420,7 +422,7 @@ class FfmpegBackend(Backend): self, item: Item, target_level: float, - peak_method: Optional[PeakMethod], + peak_method: PeakMethod | None, count_blocks: bool = True, ) -> tuple[Gain, int]: """Analyse item. Return a pair of a Gain object and the number @@ -654,7 +656,7 @@ class CommandBackend(Backend): # tag-writing; this turns the mp3gain/aacgain tool into a gain # calculator rather than a tag manipulator because we take care # of changing tags ourselves. - cmd: list[Union[bytes, str]] = [self.command, "-o", "-s", "s"] + cmd: list[bytes | str] = [self.command, "-o", "-s", "s"] if self.noclip: # Adjust to avoid clipping. cmd = cmd + ["-k"] @@ -1179,7 +1181,7 @@ class ExceptionWatcher(Thread): # whether `_stopevent` is set pass - def join(self, timeout: Optional[float] = None): + def join(self, timeout: float | None = None): self._stopevent.set() Thread.join(self, timeout) @@ -1319,7 +1321,7 @@ class ReplayGainPlugin(BeetsPlugin): self, items: Sequence[Item], use_r128: bool, - album: Optional[Album] = None, + album: Album | None = None, ) -> RgTask: if use_r128: return R128Task(