Replace Union types by PEP604 pipe character

This commit is contained in:
Šarūnas Nejus 2024-11-17 06:25:44 +00:00
parent fbfdfd5444
commit 7ef1b61070
No known key found for this signature in database
GPG key ID: DD28F6704DBE3435
8 changed files with 163 additions and 179 deletions

View file

@ -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.
"""

View file

@ -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]

View file

@ -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.
"""

View file

@ -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.
"""

View file

@ -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):

View file

@ -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.

View file

@ -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:

View file

@ -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(