Replace format calls with f-strings

This commit is contained in:
Šarūnas Nejus 2025-08-30 18:41:40 +01:00
parent 6c21482b7a
commit 4a361bd501
No known key found for this signature in database
GPG key ID: DD28F6704DBE3435
50 changed files with 335 additions and 531 deletions

View file

@ -238,11 +238,7 @@ There are a few coding conventions we use in beets:
.. code-block:: python
with g.lib.transaction() as tx:
rows = tx.query(
"SELECT DISTINCT '{0}' FROM '{1}' ORDER BY '{2}'".format(
field, model._table, sort_field
)
)
rows = tx.query("SELECT DISTINCT {field} FROM {model._table} ORDER BY {sort_field}")
To fetch Item objects from the database, use lib.items(…) and supply a query
as an argument. Resist the urge to write raw SQL for your query. If you must

View file

@ -35,7 +35,7 @@ class IncludeLazyConfig(confuse.LazyConfig):
except confuse.NotFoundError:
pass
except confuse.ConfigReadError as err:
stderr.write("configuration `import` failed: {}".format(err.reason))
stderr.write(f"configuration `import` failed: {err.reason}")
config = IncludeLazyConfig("beets", __name__)

View file

@ -79,9 +79,9 @@ def string_dist(str1: str | None, str2: str | None) -> float:
# "something, the".
for word in SD_END_WORDS:
if str1.endswith(", %s" % word):
str1 = "{} {}".format(word, str1[: -len(word) - 2])
str1 = f"{word} {str1[: -len(word) - 2]}"
if str2.endswith(", %s" % word):
str2 = "{} {}".format(word, str2[: -len(word) - 2])
str2 = f"{word} {str2[: -len(word) - 2]}"
# Perform a couple of basic normalizing substitutions.
for pat, repl in SD_REPLACE:
@ -230,7 +230,7 @@ class Distance:
"""Adds all the distance penalties from `dist`."""
if not isinstance(dist, Distance):
raise ValueError(
"`dist` must be a Distance object, not {}".format(type(dist))
f"`dist` must be a Distance object, not {type(dist)}"
)
for key, penalties in dist._penalties.items():
self._penalties.setdefault(key, []).extend(penalties)

View file

@ -390,9 +390,9 @@ class Model(ABC, Generic[D]):
return obj
def __repr__(self) -> str:
return "{}({})".format(
type(self).__name__,
", ".join(f"{k}={v!r}" for k, v in dict(self).items()),
return (
f"{type(self).__name__}"
f"({', '.join(f'{k}={v!r}' for k, v in dict(self).items())})"
)
def clear_dirty(self):
@ -409,9 +409,9 @@ class Model(ABC, Generic[D]):
exception is raised otherwise.
"""
if not self._db:
raise ValueError("{} has no database".format(type(self).__name__))
raise ValueError(f"{type(self).__name__} has no database")
if need_id and not self.id:
raise ValueError("{} has no id".format(type(self).__name__))
raise ValueError(f"{type(self).__name__} has no id")
return self._db
@ -595,9 +595,7 @@ class Model(ABC, Generic[D]):
with db.transaction() as tx:
# Main table update.
if assignments:
query = "UPDATE {} SET {} WHERE id=?".format(
self._table, ",".join(assignments)
)
query = f"UPDATE {self._table} SET {','.join(assignments)} WHERE id=?"
subvars.append(self.id)
tx.mutate(query, subvars)
@ -607,9 +605,9 @@ class Model(ABC, Generic[D]):
self._dirty.remove(key)
value = self._type(key).to_sql(value)
tx.mutate(
"INSERT INTO {} "
f"INSERT INTO {self._flex_table} "
"(entity_id, key, value) "
"VALUES (?, ?, ?);".format(self._flex_table),
"VALUES (?, ?, ?);",
(self.id, key, value),
)
@ -1173,9 +1171,7 @@ class Database:
columns = []
for name, typ in fields.items():
columns.append(f"{name} {typ.sql}")
setup_sql = "CREATE TABLE {} ({});\n".format(
table, ", ".join(columns)
)
setup_sql = f"CREATE TABLE {table} ({', '.join(columns)});\n"
else:
# Table exists does not match the field set.
@ -1183,8 +1179,8 @@ class Database:
for name, typ in fields.items():
if name in current_fields:
continue
setup_sql += "ALTER TABLE {} ADD COLUMN {} {};\n".format(
table, name, typ.sql
setup_sql += (
f"ALTER TABLE {table} ADD COLUMN {name} {typ.sql};\n"
)
with self.transaction() as tx:
@ -1195,18 +1191,16 @@ class Database:
for the given entity (if they don't exist).
"""
with self.transaction() as tx:
tx.script(
"""
CREATE TABLE IF NOT EXISTS {0} (
tx.script(f"""
CREATE TABLE IF NOT EXISTS {flex_table} (
id INTEGER PRIMARY KEY,
entity_id INTEGER,
key TEXT,
value TEXT,
UNIQUE(entity_id, key) ON CONFLICT REPLACE);
CREATE INDEX IF NOT EXISTS {0}_by_entity
ON {0} (entity_id);
""".format(flex_table)
)
CREATE INDEX IF NOT EXISTS {flex_table}_by_entity
ON {flex_table} (entity_id);
""")
# Querying.

View file

@ -475,7 +475,7 @@ class NumericQuery(FieldQuery[str]):
else:
if self.rangemin is not None and self.rangemax is not None:
return (
"{0} >= ? AND {0} <= ?".format(self.field),
f"{self.field} >= ? AND {self.field} <= ?",
(self.rangemin, self.rangemax),
)
elif self.rangemin is not None:
@ -800,9 +800,7 @@ class DateInterval:
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)
)
raise ValueError(f"start date {start} is not before end date {end}")
self.start = start
self.end = end
@ -1074,9 +1072,9 @@ class FixedFieldSort(FieldSort):
if self.case_insensitive:
field = (
"(CASE "
"WHEN TYPEOF({0})='text' THEN LOWER({0}) "
"WHEN TYPEOF({0})='blob' THEN LOWER({0}) "
"ELSE {0} END)".format(self.field)
f"WHEN TYPEOF({self.field})='text' THEN LOWER({self.field}) "
f"WHEN TYPEOF({self.field})='blob' THEN LOWER({self.field}) "
f"ELSE {self.field} END)"
)
else:
field = self.field

View file

@ -194,7 +194,7 @@ class BasePaddedInt(BaseInteger[N]):
self.digits = digits
def format(self, value: int | N) -> str:
return "{0:0{1}d}".format(value or 0, self.digits)
return f"{value or 0:0{self.digits}d}"
class PaddedInt(BasePaddedInt[int]):
@ -219,7 +219,7 @@ class ScaledInt(Integer):
self.suffix = suffix
def format(self, value: int) -> str:
return "{}{}".format((value or 0) // self.unit, self.suffix)
return f"{(value or 0) // self.unit}{self.suffix}"
class Id(NullInteger):
@ -249,7 +249,7 @@ class BaseFloat(Type[float, N]):
self.digits = digits
def format(self, value: float | N) -> str:
return "{0:.{1}f}".format(value or 0, self.digits)
return f"{value or 0:.{self.digits}f}"
class Float(BaseFloat[float]):

View file

@ -341,9 +341,7 @@ def _resolve_duplicates(session: ImportSession, task: ImportTask):
if task.choice_flag in (Action.ASIS, Action.APPLY, Action.RETAG):
found_duplicates = task.find_duplicates(session.lib)
if found_duplicates:
log.debug(
"found duplicates: {}".format([o.id for o in found_duplicates])
)
log.debug(f"found duplicates: {[o.id for o in found_duplicates]}")
# Get the default action to follow from config.
duplicate_action = config["import"]["duplicate_action"].as_choice(

View file

@ -844,12 +844,9 @@ class Item(LibModel):
# This must not use `with_album=True`, because that might access
# the database. When debugging, that is not guaranteed to succeed, and
# can even deadlock due to the database lock.
return "{}({})".format(
type(self).__name__,
", ".join(
"{}={!r}".format(k, self[k])
for k in self.keys(with_album=False)
),
return (
f"{type(self).__name__}"
f"({', '.join(f'{k}={self[k]!r}' for k in self.keys(with_album=False))})"
)
def keys(self, computed=False, with_album=True):

View file

@ -424,9 +424,9 @@ def types(model_cls: type[AnyModel]) -> dict[str, Type]:
for field in plugin_types:
if field in types and plugin_types[field] != types[field]:
raise PluginConflictError(
"Plugin {} defines flexible field {} "
f"Plugin {plugin.name} defines flexible field {field} "
"which has already been defined with "
"another type.".format(plugin.name, field)
"another type."
)
types.update(plugin_types)
return types
@ -560,8 +560,8 @@ def feat_tokens(for_artist: bool = True) -> str:
feat_words = ["ft", "featuring", "feat", "feat.", "ft."]
if for_artist:
feat_words += ["with", "vs", "and", "con", "&"]
return r"(?<=[\s(\[])(?:{})(?=\s)".format(
"|".join(re.escape(x) for x in feat_words)
return (
rf"(?<=[\s(\[])(?:{'|'.join(re.escape(x) for x in feat_words)})(?=\s)"
)

View file

@ -726,7 +726,7 @@ def get_replacements():
replacements.append((re.compile(pattern), repl))
except re.error:
raise UserError(
"malformed regular expression in replace: {}".format(pattern)
f"malformed regular expression in replace: {pattern}"
)
return replacements
@ -1163,7 +1163,7 @@ def show_model_changes(new, old=None, fields=None, always=False):
continue
changes.append(
" {}: {}".format(field, colorize("text_highlight", new_fmt[field]))
f" {field}: {colorize('text_highlight', new_fmt[field])}"
)
# Print changes.
@ -1204,22 +1204,16 @@ def show_path_changes(path_changes):
# Print every change over two lines
for source, dest in zip(sources, destinations):
color_source, color_dest = colordiff(source, dest)
print_("{0} \n -> {1}".format(color_source, color_dest))
print_(f"{color_source} \n -> {color_dest}")
else:
# Print every change on a single line, and add a header
title_pad = max_width - len("Source ") + len(" -> ")
print_("Source {0} Destination".format(" " * title_pad))
print_(f"Source {' ' * title_pad} Destination")
for source, dest in zip(sources, destinations):
pad = max_width - len(source)
color_source, color_dest = colordiff(source, dest)
print_(
"{0} {1} -> {2}".format(
color_source,
" " * pad,
color_dest,
)
)
print_(f"{color_source} {' ' * pad} -> {color_dest}")
# Helper functions for option parsing.
@ -1245,9 +1239,7 @@ def _store_dict(option, opt_str, value, parser):
raise ValueError
except ValueError:
raise UserError(
"supplied argument `{}' is not of the form `key=value'".format(
value
)
f"supplied argument `{value}' is not of the form `key=value'"
)
option_values[key] = value
@ -1426,8 +1418,8 @@ class Subcommand:
@root_parser.setter
def root_parser(self, root_parser):
self._root_parser = root_parser
self.parser.prog = "{} {}".format(
as_string(root_parser.get_prog_name()), self.name
self.parser.prog = (
f"{as_string(root_parser.get_prog_name())} {self.name}"
)
@ -1637,10 +1629,8 @@ def _ensure_db_directory_exists(path):
newpath = os.path.dirname(path)
if not os.path.isdir(newpath):
if input_yn(
"The database directory {} does not \
exist. Create it (Y/n)?".format(
util.displayable_path(newpath)
)
f"The database directory {util.displayable_path(newpath)} does not \
exist. Create it (Y/n)?"
):
os.makedirs(newpath)
@ -1660,9 +1650,8 @@ def _open_library(config: confuse.LazyConfig) -> library.Library:
except (sqlite3.OperationalError, sqlite3.DatabaseError) as db_error:
log.debug("{}", traceback.format_exc())
raise UserError(
"database file {} cannot not be opened: {}".format(
util.displayable_path(dbpath), db_error
)
f"database file {util.displayable_path(dbpath)} cannot not be"
f" opened: {db_error}"
)
log.debug(
"library database: {0}\nlibrary directory: {1}",

View file

@ -112,15 +112,11 @@ def _parse_logfiles(logfiles):
yield from _paths_from_logfile(syspath(normpath(logfile)))
except ValueError as err:
raise ui.UserError(
"malformed logfile {}: {}".format(
util.displayable_path(logfile), str(err)
)
f"malformed logfile {util.displayable_path(logfile)}: {err}"
) from err
except OSError as err:
raise ui.UserError(
"unreadable logfile {}: {}".format(
util.displayable_path(logfile), str(err)
)
f"unreadable logfile {util.displayable_path(logfile)}: {err}"
) from err
@ -213,10 +209,10 @@ def get_singleton_disambig_fields(info: hooks.TrackInfo) -> Sequence[str]:
out = []
chosen_fields = config["match"]["singleton_disambig_fields"].as_str_seq()
calculated_values = {
"index": "Index {}".format(str(info.index)),
"track_alt": "Track {}".format(info.track_alt),
"index": f"Index {info.index}",
"track_alt": f"Track {info.track_alt}",
"album": (
"[{}]".format(info.album)
f"[{info.album}]"
if (
config["import"]["singleton_album_disambig"].get()
and info.get("album")
@ -242,7 +238,7 @@ def get_album_disambig_fields(info: hooks.AlbumInfo) -> Sequence[str]:
chosen_fields = config["match"]["album_disambig_fields"].as_str_seq()
calculated_values = {
"media": (
"{}x{}".format(info.mediums, info.media)
f"{info.mediums}x{info.media}"
if (info.mediums and info.mediums > 1)
else info.media
),
@ -277,7 +273,7 @@ def dist_string(dist):
"""Formats a distance (a float) as a colorized similarity percentage
string.
"""
string = "{:.1f}%".format(((1 - dist) * 100))
string = f"{(1 - dist) * 100:.1f}%"
return dist_colorize(string, dist)
@ -295,7 +291,7 @@ def penalty_string(distance, limit=None):
if limit and len(penalties) > limit:
penalties = penalties[:limit] + ["..."]
# Prefix penalty string with U+2260: Not Equal To
penalty_string = "\u2260 {}".format(", ".join(penalties))
penalty_string = f"\u2260 {', '.join(penalties)}"
return ui.colorize("changed", penalty_string)
@ -697,11 +693,9 @@ class AlbumChange(ChangeRepresentation):
# Missing and unmatched tracks.
if self.match.extra_tracks:
print_(
"Missing tracks ({0}/{1} - {2:.1%}):".format(
len(self.match.extra_tracks),
len(self.match.info.tracks),
len(self.match.extra_tracks) / len(self.match.info.tracks),
)
"Missing tracks"
f" ({len(self.match.extra_tracks)}/{len(self.match.info.tracks)} -"
f" {len(self.match.extra_tracks) / len(self.match.info.tracks):.1%}):"
)
for track_info in self.match.extra_tracks:
line = f" ! {track_info.title} (#{self.format_index(track_info)})"
@ -711,9 +705,9 @@ class AlbumChange(ChangeRepresentation):
if self.match.extra_items:
print_(f"Unmatched tracks ({len(self.match.extra_items)}):")
for item in self.match.extra_items:
line = " ! {} (#{})".format(item.title, self.format_index(item))
line = f" ! {item.title} (#{self.format_index(item)})"
if item.length:
line += " ({})".format(human_seconds_short(item.length))
line += f" ({human_seconds_short(item.length)})"
print_(ui.colorize("text_warning", line))
@ -769,7 +763,7 @@ def summarize_items(items, singleton):
"""
summary_parts = []
if not singleton:
summary_parts.append("{} items".format(len(items)))
summary_parts.append(f"{len(items)} items")
format_counts = {}
for item in items:
@ -789,10 +783,11 @@ def summarize_items(items, singleton):
average_bitrate = sum([item.bitrate for item in items]) / len(items)
total_duration = sum([item.length for item in items])
total_filesize = sum([item.filesize for item in items])
summary_parts.append("{}kbps".format(int(average_bitrate / 1000)))
summary_parts.append(f"{int(average_bitrate / 1000)}kbps")
if items[0].format == "FLAC":
sample_bits = "{}kHz/{} bit".format(
round(int(items[0].samplerate) / 1000, 1), items[0].bitdepth
sample_bits = (
f"{round(int(items[0].samplerate) / 1000, 1)}kHz"
f"/{items[0].bitdepth} bit"
)
summary_parts.append(sample_bits)
summary_parts.append(human_seconds_short(total_duration))
@ -885,7 +880,7 @@ def choose_candidate(
if singleton:
print_("No matching recordings found.")
else:
print_("No matching release found for {} tracks.".format(itemcount))
print_(f"No matching release found for {itemcount} tracks.")
print_(
"For help, see: "
"https://beets.readthedocs.org/en/latest/faq.html#nomatch"
@ -910,23 +905,21 @@ def choose_candidate(
# Display list of candidates.
print_("")
print_(
'Finding tags for {} "{} - {}".'.format(
"track" if singleton else "album",
item.artist if singleton else cur_artist,
item.title if singleton else cur_album,
)
f"Finding tags for {'track' if singleton else 'album'}"
f'"{item.artist if singleton else cur_artist} -'
f' {item.title if singleton else cur_album}".'
)
print_(ui.indent(2) + "Candidates:")
for i, match in enumerate(candidates):
# Index, metadata, and distance.
index0 = "{0}.".format(i + 1)
index0 = f"{i + 1}."
index = dist_colorize(index0, match.distance)
dist = "({:.1f}%)".format((1 - match.distance) * 100)
dist = f"({(1 - match.distance) * 100:.1f}%)"
distance = dist_colorize(dist, match.distance)
metadata = "{0} - {1}".format(
match.info.artist,
match.info.title if singleton else match.info.album,
metadata = (
f"{match.info.artist} -"
f" {match.info.title if singleton else match.info.album}"
)
if i == 0:
metadata = dist_colorize(metadata, match.distance)
@ -1015,7 +1008,7 @@ def manual_id(session, task):
Input an ID, either for an album ("release") or a track ("recording").
"""
prompt = "Enter {} ID:".format("release" if task.is_album else "recording")
prompt = f"Enter {'release' if task.is_album else 'recording'} ID:"
search_id = input_(prompt).strip()
if task.is_album:
@ -1043,7 +1036,7 @@ class TerminalImportSession(importer.ImportSession):
path_str0 = displayable_path(task.paths, "\n")
path_str = ui.colorize("import_path", path_str0)
items_str0 = "({} items)".format(len(task.items))
items_str0 = f"({len(task.items)} items)"
items_str = ui.colorize("import_path_items", items_str0)
print_(" ".join([path_str, items_str]))
@ -1217,8 +1210,8 @@ class TerminalImportSession(importer.ImportSession):
def should_resume(self, path):
return ui.input_yn(
"Import of the directory:\n{}\n"
"was interrupted. Resume (Y/n)?".format(displayable_path(path))
f"Import of the directory:\n{displayable_path(path)}\n"
"was interrupted. Resume (Y/n)?"
)
def _get_choices(self, task):
@ -1317,7 +1310,8 @@ def import_files(lib, paths: list[bytes], query):
loghandler = logging.FileHandler(logpath, encoding="utf-8")
except OSError:
raise ui.UserError(
f"Could not open log file for writing: {displayable_path(logpath)}"
"Could not open log file for writing:"
f" {displayable_path(logpath)}"
)
else:
loghandler = None
@ -1362,9 +1356,7 @@ def import_func(lib, opts, args: list[str]):
for path in byte_paths:
if not os.path.exists(syspath(normpath(path))):
raise ui.UserError(
"no such file or directory: {}".format(
displayable_path(path)
)
f"no such file or directory: {displayable_path(path)}"
)
# Check the directories from the logfiles, but don't throw an error in
@ -1374,9 +1366,7 @@ def import_func(lib, opts, args: list[str]):
for path in paths_from_logfiles:
if not os.path.exists(syspath(normpath(path))):
log.warning(
"No such file or directory: {}".format(
displayable_path(path)
)
f"No such file or directory: {displayable_path(path)}"
)
continue
@ -1808,7 +1798,7 @@ def remove_items(lib, query, album, delete, force):
if not force:
# Prepare confirmation with user.
album_str = (
" in {} album{}".format(len(albums), "s" if len(albums) > 1 else "")
f" in {len(albums)} album{'s' if len(albums) > 1 else ''}"
if album
else ""
)
@ -1816,14 +1806,17 @@ def remove_items(lib, query, album, delete, force):
if delete:
fmt = "$path - $title"
prompt = "Really DELETE"
prompt_all = "Really DELETE {} file{}{}".format(
len(items), "s" if len(items) > 1 else "", album_str
prompt_all = (
"Really DELETE"
f" {len(items)} file{'s' if len(items) > 1 else ''}{album_str}"
)
else:
fmt = ""
prompt = "Really remove from the library?"
prompt_all = "Really remove {} item{}{} from the library?".format(
len(items), "s" if len(items) > 1 else "", album_str
prompt_all = (
"Really remove"
f" {len(items)} item{'s' if len(items) > 1 else ''}{album_str}"
" from the library?"
)
# Helpers for printing affected items
@ -1906,23 +1899,13 @@ def show_stats(lib, query, exact):
if exact:
size_str += f" ({total_size} bytes)"
print_(
"""Tracks: {}
Total time: {}{}
{}: {}
Artists: {}
Albums: {}
Album artists: {}""".format(
total_items,
human_seconds(total_time),
f" ({total_time:.2f} seconds)" if exact else "",
"Total size" if exact else "Approximate total size",
size_str,
len(artists),
len(albums),
len(album_artists),
),
)
print_(f"""Tracks: {total_items}
Total time: {human_seconds(total_time)}
{f" ({total_time:.2f} seconds)" if exact else ""}
{"Total size" if exact else "Approximate total size"}: {size_str}
Artists: {len(artists)}
Albums: {len(albums)}
Album artists: {len(album_artists)}""")
def stats_func(lib, opts, args):
@ -1977,7 +1960,7 @@ def modify_items(lib, mods, dels, query, write, move, album, confirm, inherit):
# Apply changes *temporarily*, preview them, and collect modified
# objects.
print_("Modifying {} {}s.".format(len(objs), "album" if album else "item"))
print_(f"Modifying {len(objs)} {'album' if album else 'item'}s.")
changed = []
templates = {
key: functemplate.template(value) for key, value in mods.items()
@ -2213,9 +2196,7 @@ def move_func(lib, opts, args):
if dest is not None:
dest = normpath(dest)
if not os.path.isdir(syspath(dest)):
raise ui.UserError(
"no such directory: {}".format(displayable_path(dest))
)
raise ui.UserError(f"no such directory: {displayable_path(dest)}")
move_items(
lib,
@ -2486,7 +2467,7 @@ def completion_script(commands):
# Command aliases
yield " local aliases='%s'\n" % " ".join(aliases.keys())
for alias, cmd in aliases.items():
yield " local alias__{}={}\n".format(alias.replace("-", "_"), cmd)
yield f" local alias__{alias.replace('-', '_')}={cmd}\n"
yield "\n"
# Fields
@ -2502,8 +2483,9 @@ def completion_script(commands):
for option_type, option_list in opts.items():
if option_list:
option_list = " ".join(option_list)
yield " local {}__{}='{}'\n".format(
option_type, cmd.replace("-", "_"), option_list
yield (
" local"
f" {option_type}__{cmd.replace('-', '_')}='{option_list}'\n"
)
yield " _beet_dispatch\n"

View file

@ -112,7 +112,7 @@ class HumanReadableError(Exception):
elif hasattr(self.reason, "strerror"): # i.e., EnvironmentError
return self.reason.strerror
else:
return '"{}"'.format(str(self.reason))
return f'"{self.reason}"'
def get_message(self):
"""Create the human-readable description of the error, sans
@ -142,18 +142,16 @@ class FilesystemError(HumanReadableError):
def get_message(self):
# Use a nicer English phrasing for some specific verbs.
if self.verb in ("move", "copy", "rename"):
clause = "while {} {} to {}".format(
self._gerund(),
displayable_path(self.paths[0]),
displayable_path(self.paths[1]),
clause = (
f"while {self._gerund()} {displayable_path(self.paths[0])} to"
f" {displayable_path(self.paths[1])}"
)
elif self.verb in ("delete", "write", "create", "read"):
clause = "while {} {}".format(
self._gerund(), displayable_path(self.paths[0])
)
clause = f"while {self._gerund()} {displayable_path(self.paths[0])}"
else:
clause = "during {} of paths {}".format(
self.verb, ", ".join(displayable_path(p) for p in self.paths)
clause = (
f"during {self.verb} of paths"
f" {', '.join(displayable_path(p) for p in self.paths)}"
)
return f"{self._reasonstr()} {clause}"
@ -226,9 +224,8 @@ def sorted_walk(
except OSError as exc:
if logger:
logger.warning(
"could not list directory {}: {}".format(
displayable_path(bytes_path), exc.strerror
)
f"could not list directory {displayable_path(bytes_path)}:"
f" {exc.strerror}"
)
return
dirs = []

View file

@ -54,7 +54,7 @@ def resize_url(url: str, maxwidth: int, quality: int = 0) -> str:
if quality > 0:
params["q"] = quality
return "{}?{}".format(PROXY_URL, urlencode(params))
return f"{PROXY_URL}?{urlencode(params)}"
class LocalBackendNotAvailableError(Exception):

View file

@ -165,9 +165,7 @@ class Call:
self.original = original
def __repr__(self):
return "Call({}, {}, {})".format(
repr(self.ident), repr(self.args), repr(self.original)
)
return f"Call({self.ident!r}, {self.args!r}, {self.original!r})"
def evaluate(self, env):
"""Evaluate the function call in the environment, returning a

View file

@ -42,9 +42,7 @@ def call(args):
try:
return util.command_output(args).stdout
except subprocess.CalledProcessError as e:
raise ABSubmitError(
"{} exited with status {}".format(args[0], e.returncode)
)
raise ABSubmitError(f"{args[0]} exited with status {e.returncode}")
class AcousticBrainzSubmitPlugin(plugins.BeetsPlugin):
@ -63,9 +61,7 @@ class AcousticBrainzSubmitPlugin(plugins.BeetsPlugin):
# Explicit path to extractor
if not os.path.isfile(self.extractor):
raise ui.UserError(
"Extractor command does not exist: {0}.".format(
self.extractor
)
f"Extractor command does not exist: {self.extractor}."
)
else:
# Implicit path to extractor, search for it in path

View file

@ -243,7 +243,7 @@ class AURADocument:
else:
# Increment page token by 1
next_url = request.url.replace(
f"page={page}", "page={}".format(page + 1)
f"page={page}", f"page={page + 1}"
)
# Get only the items in the page range
data = [
@ -427,9 +427,7 @@ class TrackDocument(AURADocument):
return self.error(
"404 Not Found",
"No track with the requested id.",
"There is no track with an id of {} in the library.".format(
track_id
),
f"There is no track with an id of {track_id} in the library.",
)
return self.single_resource_document(
self.get_resource_object(self.lib, track)
@ -513,9 +511,7 @@ class AlbumDocument(AURADocument):
return self.error(
"404 Not Found",
"No album with the requested id.",
"There is no album with an id of {} in the library.".format(
album_id
),
f"There is no album with an id of {album_id} in the library.",
)
return self.single_resource_document(
self.get_resource_object(self.lib, album)
@ -600,9 +596,7 @@ class ArtistDocument(AURADocument):
return self.error(
"404 Not Found",
"No artist with the requested id.",
"There is no artist with an id of {} in the library.".format(
artist_id
),
f"There is no artist with an id of {artist_id} in the library.",
)
return self.single_resource_document(artist_resource)
@ -727,9 +721,7 @@ class ImageDocument(AURADocument):
return self.error(
"404 Not Found",
"No image with the requested id.",
"There is no image with an id of {} in the library.".format(
image_id
),
f"There is no image with an id of {image_id} in the library.",
)
return self.single_resource_document(image_resource)
@ -775,9 +767,7 @@ def audio_file(track_id):
return AURADocument.error(
"404 Not Found",
"No track with the requested id.",
"There is no track with an id of {} in the library.".format(
track_id
),
f"There is no track with an id of {track_id} in the library.",
)
path = os.fsdecode(track.path)
@ -785,9 +775,8 @@ def audio_file(track_id):
return AURADocument.error(
"404 Not Found",
"No audio file for the requested track.",
(
"There is no audio file for track {} at the expected location"
).format(track_id),
f"There is no audio file for track {track_id} at the expected"
" location",
)
file_mimetype = guess_type(path)[0]
@ -795,10 +784,8 @@ def audio_file(track_id):
return AURADocument.error(
"500 Internal Server Error",
"Requested audio file has an unknown mimetype.",
(
"The audio file for track {} has an unknown mimetype. "
"Its file extension is {}."
).format(track_id, path.split(".")[-1]),
f"The audio file for track {track_id} has an unknown mimetype. "
f"Its file extension is {path.split('.')[-1]}.",
)
# Check that the Accept header contains the file's mimetype
@ -810,10 +797,8 @@ def audio_file(track_id):
return AURADocument.error(
"406 Not Acceptable",
"Unsupported MIME type or bitrate parameter in Accept header.",
(
"The audio file for track {} is only available as {} and "
"bitrate parameters are not supported."
).format(track_id, file_mimetype),
f"The audio file for track {track_id} is only available as"
f" {file_mimetype} and bitrate parameters are not supported.",
)
return send_file(
@ -896,9 +881,7 @@ def image_file(image_id):
return AURADocument.error(
"404 Not Found",
"No image with the requested id.",
"There is no image with an id of {} in the library".format(
image_id
),
f"There is no image with an id of {image_id} in the library",
)
return send_file(img_path)

View file

@ -110,9 +110,7 @@ class BadFiles(BeetsPlugin):
self._log.debug("checking path: {}", dpath)
if not os.path.exists(item.path):
ui.print_(
"{}: file does not exist".format(
ui.colorize("text_error", dpath)
)
f"{ui.colorize('text_error', dpath)}: file does not exist"
)
# Run the checker against the file if one is found
@ -141,25 +139,21 @@ class BadFiles(BeetsPlugin):
if status > 0:
error_lines.append(
"{}: checker exited with status {}".format(
ui.colorize("text_error", dpath), status
)
f"{ui.colorize('text_error', dpath)}: checker exited with"
f" status {status}"
)
for line in output:
error_lines.append(f" {line}")
elif errors > 0:
error_lines.append(
"{}: checker found {} errors or warnings".format(
ui.colorize("text_warning", dpath), errors
)
f"{ui.colorize('text_warning', dpath)}: checker found"
f" {status} errors or warnings"
)
for line in output:
error_lines.append(f" {line}")
elif self.verbose:
error_lines.append(
"{}: ok".format(ui.colorize("text_success", dpath))
)
error_lines.append(f"{ui.colorize('text_success', dpath)}: ok")
return error_lines
@ -180,9 +174,8 @@ class BadFiles(BeetsPlugin):
def on_import_task_before_choice(self, task, session):
if hasattr(task, "_badfiles_checks_failed"):
ui.print_(
"{} one or more files failed checks:".format(
ui.colorize("text_warning", "BAD")
)
f"{ui.colorize('text_warning', 'BAD')} one or more files failed"
" checks:"
)
for error in task._badfiles_checks_failed:
for error_line in error:

View file

@ -212,14 +212,10 @@ class BeatportClient:
try:
response = self.api.get(self._make_url(endpoint), params=kwargs)
except Exception as e:
raise BeatportAPIError(
"Error connecting to Beatport API: {}".format(e)
)
raise BeatportAPIError(f"Error connecting to Beatport API: {e}")
if not response:
raise BeatportAPIError(
"Error {0.status_code} for '{0.request.path_url}".format(
response
)
f"Error {response.status_code} for '{response.request.path_url}"
)
return response.json()["results"]
@ -275,15 +271,14 @@ class BeatportRelease(BeatportObject):
self.genre = data.get("genre")
if "slug" in data:
self.url = "https://beatport.com/release/{}/{}".format(
data["slug"], data["id"]
self.url = (
f"https://beatport.com/release/{data['slug']}/{data['id']}"
)
def __str__(self) -> str:
return "<BeatportRelease: {} - {} ({})>".format(
self.artists_str(),
self.name,
self.catalog_number,
return (
"<BeatportRelease: "
f"{self.artists_str()} - {self.name} ({self.catalog_number})>"
)
@ -311,9 +306,7 @@ class BeatportTrack(BeatportObject):
except ValueError:
pass
if "slug" in data:
self.url = "https://beatport.com/track/{}/{}".format(
data["slug"], data["id"]
)
self.url = f"https://beatport.com/track/{data['slug']}/{data['id']}"
self.track_number = data.get("trackNumber")
self.bpm = data.get("bpm")
self.initial_key = str((data.get("key") or {}).get("shortName"))

View file

@ -759,7 +759,7 @@ class Connection:
"""Create a new connection for the accepted socket `client`."""
self.server = server
self.sock = sock
self.address = "{}:{}".format(*sock.sock.getpeername())
self.address = ":".join(map(str, sock.sock.getpeername()))
def debug(self, message, kind=" "):
"""Log a debug message about this connection."""
@ -899,9 +899,7 @@ class MPDConnection(Connection):
return
except BPDIdleError as e:
self.idle_subscriptions = e.subsystems
self.debug(
"awaiting: {}".format(" ".join(e.subsystems)), kind="z"
)
self.debug(f"awaiting: {' '.join(e.subsystems)}", kind="z")
yield bluelet.call(self.server.dispatch_events())
@ -933,7 +931,7 @@ class ControlConnection(Connection):
func = command.delegate("ctrl_", self)
yield bluelet.call(func(*command.args))
except (AttributeError, TypeError) as e:
yield self.send("ERROR: {}".format(e.args[0]))
yield self.send(f"ERROR: {e.args[0]}")
except Exception:
yield self.send(
["ERROR: server error", traceback.format_exc().rstrip()]
@ -1011,7 +1009,7 @@ class Command:
# If the command accepts a variable number of arguments skip the check.
if wrong_num and not argspec.varargs:
raise TypeError(
'wrong number of arguments for "{}"'.format(self.name),
f'wrong number of arguments for "{self.name}"',
self.name,
)
@ -1110,10 +1108,8 @@ class Server(BaseServer):
self.lib = library
self.player = gstplayer.GstPlayer(self.play_finished)
self.cmd_update(None)
log.info("Server ready and listening on {}:{}".format(host, port))
log.debug(
"Listening for control signals on {}:{}".format(host, ctrl_port)
)
log.info(f"Server ready and listening on {host}:{port}")
log.debug(f"Listening for control signals on {host}:{ctrl_port}")
def run(self):
self.player.run()
@ -1142,9 +1138,7 @@ class Server(BaseServer):
pass
for tagtype, field in self.tagtype_map.items():
info_lines.append(
"{}: {}".format(tagtype, str(getattr(item, field)))
)
info_lines.append(f"{tagtype}: {getattr(item, field)}")
return info_lines
@ -1303,19 +1297,12 @@ class Server(BaseServer):
yield (
"bitrate: " + str(item.bitrate / 1000),
"audio: {}:{}:{}".format(
str(item.samplerate),
str(item.bitdepth),
str(item.channels),
),
f"audio: {item.samplerate}:{item.bitdepth}:{item.channels}",
)
(pos, total) = self.player.time()
yield (
"time: {}:{}".format(
str(int(pos)),
str(int(total)),
),
f"time: {int(pos)}:{int(total)}",
"elapsed: " + f"{pos:.3f}",
"duration: " + f"{total:.3f}",
)

View file

@ -55,8 +55,8 @@ def span_from_str(span_str):
years = [int(x) for x in re.findall(r"\d+", span_str)]
if not years:
raise ui.UserError(
"invalid range defined for year bucket '%s': no "
"year found" % span_str
"invalid range defined for year bucket '%s': no year found"
% span_str
)
try:
years = [normalize_year(x, years[0]) for x in years]
@ -125,11 +125,8 @@ def str2fmt(s):
"fromnchars": len(m.group("fromyear")),
"tonchars": len(m.group("toyear")),
}
res["fmt"] = "{}%s{}{}{}".format(
m.group("bef"),
m.group("sep"),
"%s" if res["tonchars"] else "",
m.group("after"),
res["fmt"] = (
f"{m['bef']}%s{m['sep']}{'%s' if res['tonchars'] else ''}{m['after']}"
)
return res

View file

@ -64,9 +64,7 @@ def get_format(fmt=None):
command = format_info["command"]
extension = format_info.get("extension", fmt)
except KeyError:
raise ui.UserError(
'convert: format {} needs the "command" field'.format(fmt)
)
raise ui.UserError(f'convert: format {fmt} needs the "command" field')
except ConfigTypeError:
command = config["convert"]["formats"][fmt].get(str)
extension = fmt
@ -77,8 +75,8 @@ def get_format(fmt=None):
command = config["convert"]["command"].as_str()
elif "opts" in keys:
# Undocumented option for backwards compatibility with < 1.3.1.
command = "ffmpeg -i $source -y {} $dest".format(
config["convert"]["opts"].as_str()
command = (
f"ffmpeg -i $source -y {config['convert']['opts'].as_str()} $dest"
)
if "extension" in keys:
extension = config["convert"]["extension"].as_str()
@ -125,18 +123,25 @@ class ConvertPlugin(BeetsPlugin):
"id3v23": "inherit",
"formats": {
"aac": {
"command": "ffmpeg -i $source -y -vn -acodec aac "
"-aq 1 $dest",
"command": (
"ffmpeg -i $source -y -vn -acodec aac -aq 1 $dest"
),
"extension": "m4a",
},
"alac": {
"command": "ffmpeg -i $source -y -vn -acodec alac $dest",
"command": (
"ffmpeg -i $source -y -vn -acodec alac $dest"
),
"extension": "m4a",
},
"flac": "ffmpeg -i $source -y -vn -acodec flac $dest",
"mp3": "ffmpeg -i $source -y -vn -aq 2 $dest",
"opus": "ffmpeg -i $source -y -vn -acodec libopus -ab 96k $dest",
"ogg": "ffmpeg -i $source -y -vn -acodec libvorbis -aq 3 $dest",
"opus": (
"ffmpeg -i $source -y -vn -acodec libopus -ab 96k $dest"
),
"ogg": (
"ffmpeg -i $source -y -vn -acodec libvorbis -aq 3 $dest"
),
"wma": "ffmpeg -i $source -y -vn -acodec wmav2 -vn $dest",
},
"max_bitrate": None,
@ -323,7 +328,7 @@ class ConvertPlugin(BeetsPlugin):
raise
except OSError as exc:
raise ui.UserError(
"convert: couldn't invoke '{}': {}".format(" ".join(args), exc)
f"convert: couldn't invoke {' '.join(args)!r}: {exc}"
)
if not quiet and not pretend:

View file

@ -552,7 +552,7 @@ class DiscogsPlugin(MetadataSourcePlugin):
idx, medium_idx, sub_idx = self.get_track_index(
subtracks[0]["position"]
)
position = "{}{}".format(idx or "", medium_idx or "")
position = f"{idx or ''}{medium_idx or ''}"
if tracklist and not tracklist[-1]["position"]:
# Assume the previous index track contains the track title.
@ -574,8 +574,8 @@ class DiscogsPlugin(MetadataSourcePlugin):
# option is set
if self.config["index_tracks"]:
for subtrack in subtracks:
subtrack["title"] = "{}: {}".format(
index_track["title"], subtrack["title"]
subtrack["title"] = (
f"{index_track['title']}: {subtrack['title']}"
)
tracklist.extend(subtracks)
else:

View file

@ -46,9 +46,7 @@ def edit(filename, log):
try:
subprocess.call(cmd)
except OSError as exc:
raise ui.UserError(
"could not run editor command {!r}: {}".format(cmd[0], exc)
)
raise ui.UserError(f"could not run editor command {cmd[0]!r}: {exc}")
def dump(arg):
@ -71,9 +69,7 @@ def load(s):
for d in yaml.safe_load_all(s):
if not isinstance(d, dict):
raise ParseError(
"each entry must be a dictionary; found {}".format(
type(d).__name__
)
f"each entry must be a dictionary; found {type(d).__name__}"
)
# Convert all keys to strings. They started out as strings,

View file

@ -35,8 +35,9 @@ def _confirm(objs, album):
to items).
"""
noun = "album" if album else "file"
prompt = "Modify artwork for {} {}{} (Y/n)?".format(
len(objs), noun, "s" if len(objs) > 1 else ""
prompt = (
"Modify artwork for"
f" {len(objs)} {noun}{'s' if len(objs) > 1 else ''} (Y/n)?"
)
# Show all the items or albums.
@ -110,9 +111,7 @@ class EmbedCoverArtPlugin(BeetsPlugin):
imagepath = normpath(opts.file)
if not os.path.isfile(syspath(imagepath)):
raise ui.UserError(
"image file {} not found".format(
displayable_path(imagepath)
)
f"image file {displayable_path(imagepath)} not found"
)
items = lib.items(args)
@ -137,7 +136,7 @@ class EmbedCoverArtPlugin(BeetsPlugin):
response = requests.get(opts.url, timeout=5)
response.raise_for_status()
except requests.exceptions.RequestException as e:
self._log.error("{}".format(e))
self._log.error(f"{e}")
return
extension = guess_extension(response.headers["Content-Type"])
if extension is None:
@ -149,7 +148,7 @@ class EmbedCoverArtPlugin(BeetsPlugin):
with open(tempimg, "wb") as f:
f.write(response.content)
except Exception as e:
self._log.error("Unable to save image: {}".format(e))
self._log.error(f"Unable to save image: {e}")
return
items = lib.items(args)
# Confirm with user.

View file

@ -38,9 +38,7 @@ def api_url(host, port, endpoint):
hostname_list.insert(0, "http://")
hostname = "".join(hostname_list)
joined = urljoin(
"{hostname}:{port}".format(hostname=hostname, port=port), endpoint
)
joined = urljoin(f"{hostname}:{port}", endpoint)
scheme, netloc, path, query_string, fragment = urlsplit(joined)
query_params = parse_qs(query_string)
@ -81,12 +79,12 @@ def create_headers(user_id, token=None):
headers = {}
authorization = (
'MediaBrowser UserId="{user_id}", '
f'MediaBrowser UserId="{user_id}", '
'Client="other", '
'Device="beets", '
'DeviceId="beets", '
'Version="0.0.0"'
).format(user_id=user_id)
)
headers["x-emby-authorization"] = authorization

View file

@ -89,8 +89,9 @@ class FishPlugin(BeetsPlugin):
"-o",
"--output",
default="~/.config/fish/completions/beet.fish",
help="where to save the script. default: "
"~/.config/fish/completions",
help=(
"where to save the script. default: ~/.config/fish/completions"
),
)
return [cmd]
@ -127,16 +128,12 @@ class FishPlugin(BeetsPlugin):
totstring += "" if nobasicfields else get_standard_fields(fields)
totstring += get_extravalues(lib, extravalues) if extravalues else ""
totstring += (
"\n"
+ "# ====== {} =====".format("setup basic beet completion")
+ "\n" * 2
"\n" + "# ====== setup basic beet completion =====" + "\n" * 2
)
totstring += get_basic_beet_options()
totstring += (
"\n"
+ "# ====== {} =====".format(
"setup field completion for subcommands"
)
+ "# ====== setup field completion for subcommands ====="
+ "\n"
)
totstring += get_subcommands(cmd_names_help, nobasicfields, extravalues)
@ -226,11 +223,7 @@ def get_subcommands(cmd_name_and_help, nobasicfields, extravalues):
for cmdname, cmdhelp in cmd_name_and_help:
cmdname = _escape(cmdname)
word += (
"\n"
+ "# ------ {} -------".format("fieldsetups for " + cmdname)
+ "\n"
)
word += "\n" + f"# ------ fieldsetups for {cmdname} -------" + "\n"
word += BL_NEED2.format(
("-a " + cmdname), ("-f " + "-d " + wrap(clean_whitespace(cmdhelp)))
)
@ -268,11 +261,7 @@ def get_all_commands(beetcmds):
name = _escape(name)
word += "\n"
word += (
("\n" * 2)
+ "# ====== {} =====".format("completions for " + name)
+ "\n"
)
word += ("\n" * 2) + f"# ====== completions for {name} =====" + "\n"
for option in cmd.parser._get_all_options()[1:]:
cmd_l = (
@ -332,7 +321,7 @@ def clean_whitespace(word):
def wrap(word):
# Need " or ' around strings but watch out if they're in the string
sptoken = '"'
if ('"') in word and ("'") in word:
if '"' in word and ("'") in word:
word.replace('"', sptoken)
return '"' + word + '"'

View file

@ -112,7 +112,7 @@ def apply_matches(d, log):
for item in d:
if not item.artist:
item.artist = artist
log.info("Artist replaced with: {}".format(item.artist))
log.info(f"Artist replaced with: {item.artist}")
# No artist field: remaining field is the title.
else:
@ -122,11 +122,11 @@ def apply_matches(d, log):
for item in d:
if bad_title(item.title):
item.title = str(d[item][title_field])
log.info("Title replaced with: {}".format(item.title))
log.info(f"Title replaced with: {item.title}")
if "track" in d[item] and item.track == 0:
item.track = int(d[item]["track"])
log.info("Track replaced with: {}".format(item.track))
log.info(f"Track replaced with: {item.track}")
# Plugin structure and hook into import process.

View file

@ -154,7 +154,7 @@ def search_pairs(item):
# examples include (live), (remix), and (acoustic).
r"(.+?)\s+[(].*[)]$",
# Remove any featuring artists from the title
r"(.*?) {}".format(plugins.feat_tokens(for_artist=False)),
rf"(.*?) {plugins.feat_tokens(for_artist=False)}",
# Remove part of title after colon ':' for songs with subtitles
r"(.+?)\s*:.*",
]

View file

@ -83,9 +83,7 @@ class MusicBrainzCollectionPlugin(BeetsPlugin):
collection = self.config["collection"].as_str()
if collection:
if collection not in collection_ids:
raise ui.UserError(
"invalid collection ID: {}".format(collection)
)
raise ui.UserError(f"invalid collection ID: {collection}")
return collection
# No specified collection. Just return the first collection ID

View file

@ -117,7 +117,7 @@ class MetaSyncPlugin(BeetsPlugin):
try:
cls = META_SOURCES[player]
except KeyError:
self._log.error("Unknown metadata source '{}'".format(player))
self._log.error(f"Unknown metadata source '{player}'")
try:
meta_source_instances[player] = cls(self.config, self._log)

View file

@ -68,9 +68,7 @@ class MusicBrainzAPIError(util.HumanReadableError):
super().__init__(reason, verb, tb)
def get_message(self):
return "{} in {} with query {}".format(
self._reasonstr(), self.verb, repr(self.query)
)
return f"{self._reasonstr()} in {self.verb} with query {self.query!r}"
RELEASE_INCLUDES = list(

View file

@ -43,7 +43,7 @@ def play(
"""
# Print number of tracks or albums to be played, log command to be run.
item_type += "s" if len(selection) > 1 else ""
ui.print_("Playing {} {}.".format(len(selection), item_type))
ui.print_(f"Playing {len(selection)} {item_type}.")
log.debug("executing command: {} {!r}", command_str, open_args)
try:
@ -179,9 +179,7 @@ class PlayPlugin(BeetsPlugin):
ui.print_(
ui.colorize(
"text_warning",
"You are about to queue {} {}.".format(
len(selection), item_type
),
f"You are about to queue {len(selection)} {item_type}.",
)
)

View file

@ -133,21 +133,16 @@ class PlaylistPlugin(beets.plugins.BeetsPlugin):
try:
self.update_playlist(playlist, base_dir)
except beets.util.FilesystemError:
self._log.error(
"Failed to update playlist: {}".format(
beets.util.displayable_path(playlist)
)
)
self._log.error("Failed to update playlist: {}", playlist)
def find_playlists(self):
"""Find M3U playlists in the playlist directory."""
playlist_dir = beets.util.syspath(self.playlist_dir)
try:
dir_contents = os.listdir(beets.util.syspath(self.playlist_dir))
dir_contents = os.listdir(playlist_dir)
except OSError:
self._log.warning(
"Unable to open playlist directory {}".format(
beets.util.displayable_path(self.playlist_dir)
)
"Unable to open playlist directory {}", self.playlist_dir
)
return
@ -195,9 +190,10 @@ class PlaylistPlugin(beets.plugins.BeetsPlugin):
if changes or deletions:
self._log.info(
"Updated playlist {} ({} changes, {} deletions)".format(
filename, changes, deletions
)
"Updated playlist {} ({} changes, {} deletions)",
filename,
changes,
deletions,
)
beets.util.copy(new_playlist, filename, replace=True)
beets.util.remove(new_playlist)

View file

@ -22,9 +22,7 @@ def get_music_section(
):
"""Getting the section key for the music library in Plex."""
api_endpoint = append_token("library/sections", token)
url = urljoin(
"{}://{}:{}".format(get_protocol(secure), host, port), api_endpoint
)
url = urljoin(f"{get_protocol(secure)}://{host}:{port}", api_endpoint)
# Sends request.
r = requests.get(
@ -54,9 +52,7 @@ def update_plex(host, port, token, library_name, secure, ignore_cert_errors):
)
api_endpoint = f"library/sections/{section_key}/refresh"
api_endpoint = append_token(api_endpoint, token)
url = urljoin(
"{}://{}:{}".format(get_protocol(secure), host, port), api_endpoint
)
url = urljoin(f"{get_protocol(secure)}://{host}:{port}", api_endpoint)
# Sends request and returns requests object.
r = requests.get(

View file

@ -70,9 +70,7 @@ def call(args: list[str], log: Logger, **kwargs: Any):
return command_output(args, **kwargs)
except subprocess.CalledProcessError as e:
log.debug(e.output.decode("utf8", "ignore"))
raise ReplayGainError(
"{} exited with status {}".format(args[0], e.returncode)
)
raise ReplayGainError(f"{args[0]} exited with status {e.returncode}")
def db_to_lufs(db: float) -> float:
@ -170,9 +168,8 @@ class RgTask:
# `track_gains` without throwing FatalReplayGainError
# => raise non-fatal exception & continue
raise ReplayGainError(
"ReplayGain backend `{}` failed for track {}".format(
self.backend_name, item
)
f"ReplayGain backend `{self.backend_name}` failed for track"
f" {item}"
)
self._store_track_gain(item, self.track_gains[0])
@ -191,10 +188,8 @@ class RgTask:
# `album_gain` without throwing FatalReplayGainError
# => raise non-fatal exception & continue
raise ReplayGainError(
"ReplayGain backend `{}` failed "
"for some tracks in album {}".format(
self.backend_name, self.album
)
f"ReplayGain backend `{self.backend_name}` failed "
f"for some tracks in album {self.album}"
)
for item, track_gain in zip(self.items, self.track_gains):
self._store_track_gain(item, track_gain)
@ -501,12 +496,10 @@ class FfmpegBackend(Backend):
if self._parse_float(b"M: " + line[1]) >= gating_threshold:
n_blocks += 1
self._log.debug(
"{}: {} blocks over {} LUFS".format(
item, n_blocks, gating_threshold
)
f"{item}: {n_blocks} blocks over {gating_threshold} LUFS"
)
self._log.debug("{}: gain {} LU, peak {}".format(item, gain, peak))
self._log.debug(f"{item}: gain {gain} LU, peak {peak}")
return Gain(gain, peak), n_blocks
@ -526,9 +519,7 @@ class FfmpegBackend(Backend):
if output[i].startswith(search):
return i
raise ReplayGainError(
"ffmpeg output: missing {} after line {}".format(
repr(search), start_line
)
f"ffmpeg output: missing {search!r} after line {start_line}"
)
def _parse_float(self, line: bytes) -> float:
@ -575,7 +566,7 @@ class CommandBackend(Backend):
# Explicit executable path.
if not os.path.isfile(self.command):
raise FatalReplayGainError(
"replaygain command does not exist: {}".format(self.command)
f"replaygain command does not exist: {self.command}"
)
else:
# Check whether the program is in $PATH.
@ -1229,10 +1220,8 @@ class ReplayGainPlugin(BeetsPlugin):
if self.backend_name not in BACKENDS:
raise ui.UserError(
"Selected ReplayGain backend {} is not supported. "
"Please select one of: {}".format(
self.backend_name, ", ".join(BACKENDS.keys())
)
f"Selected ReplayGain backend {self.backend_name} is not"
f" supported. Please select one of: {', '.join(BACKENDS)}"
)
# FIXME: Consider renaming the configuration option to 'peak_method'
@ -1240,10 +1229,9 @@ class ReplayGainPlugin(BeetsPlugin):
peak_method = self.config["peak"].as_str()
if peak_method not in PeakMethod.__members__:
raise ui.UserError(
"Selected ReplayGain peak method {} is not supported. "
"Please select one of: {}".format(
peak_method, ", ".join(PeakMethod.__members__)
)
f"Selected ReplayGain peak method {peak_method} is not"
" supported. Please select one of:"
f" {', '.join(PeakMethod.__members__)}"
)
# This only applies to plain old rg tags, r128 doesn't store peak
# values.
@ -1526,18 +1514,16 @@ class ReplayGainPlugin(BeetsPlugin):
if opts.album:
albums = lib.albums(args)
self._log.info(
"Analyzing {} albums ~ {} backend...".format(
len(albums), self.backend_name
)
f"Analyzing {len(albums)} albums ~"
f" {self.backend_name} backend..."
)
for album in albums:
self.handle_album(album, write, force)
else:
items = lib.items(args)
self._log.info(
"Analyzing {} tracks ~ {} backend...".format(
len(items), self.backend_name
)
f"Analyzing {len(items)} tracks ~"
f" {self.backend_name} backend..."
)
for item in items:
self.handle_track(item, write, force)
@ -1565,8 +1551,10 @@ class ReplayGainPlugin(BeetsPlugin):
dest="force",
action="store_true",
default=False,
help="analyze all files, including those that "
"already have ReplayGain metadata",
help=(
"analyze all files, including those that already have"
" ReplayGain metadata"
),
)
cmd.parser.add_option(
"-w",

View file

@ -138,10 +138,9 @@ class SmartPlaylistPlugin(BeetsPlugin):
if name in args
}
if not playlists:
unmatched = [name for name, _, _ in self._unmatched_playlists]
raise ui.UserError(
"No playlist matching any of {} found".format(
[name for name, _, _ in self._unmatched_playlists]
)
f"No playlist matching any of {unmatched} found"
)
self._matched_playlists = playlists
@ -331,8 +330,9 @@ class SmartPlaylistPlugin(BeetsPlugin):
for key, value in attr
]
attrs = "".join(al)
comment = "#EXTINF:{}{},{} - {}\n".format(
int(item.length), attrs, item.artist, item.title
comment = (
f"#EXTINF:{int(item.length)}{attrs},"
f"{item.artist} - {item.title}\n"
)
f.write(comment.encode("utf-8") + entry.uri + b"\n")
# Send an event when playlists were updated.

View file

@ -168,8 +168,9 @@ class SpotifyPlugin(
c_secret: str = self.config["client_secret"].as_str()
headers = {
"Authorization": "Basic {}".format(
base64.b64encode(f"{c_id}:{c_secret}".encode()).decode()
"Authorization": (
"Basic"
f" {base64.b64encode(f'{c_id}:{c_secret}'.encode()).decode()}"
)
}
response = requests.post(
@ -182,7 +183,7 @@ class SpotifyPlugin(
response.raise_for_status()
except requests.exceptions.HTTPError as e:
raise ui.UserError(
"Spotify authorization failed: {}\n{}".format(e, response.text)
f"Spotify authorization failed: {e}\n{response.text}"
)
self.access_token = response.json()["access_token"]
@ -236,7 +237,7 @@ class SpotifyPlugin(
if e.response.status_code == 401:
self._log.debug(
f"{self.data_source} access token has expired. "
f"Reauthenticating."
"Reauthenticating."
)
self._authenticate()
return self._handle_response(
@ -314,9 +315,7 @@ class SpotifyPlugin(
else:
raise ui.UserError(
"Invalid `release_date_precision` returned "
"by {} API: '{}'".format(
self.data_source, release_date_precision
)
f"by {self.data_source} API: '{release_date_precision}'"
)
tracks_data = album_data["tracks"]
@ -472,17 +471,17 @@ class SpotifyPlugin(
"-m",
"--mode",
action="store",
help='"open" to open {} with playlist, '
'"list" to print (default)'.format(self.data_source),
help=(
f'"open" to open {self.data_source} with playlist, '
'"list" to print (default)'
),
)
spotify_cmd.parser.add_option(
"-f",
"--show-failures",
action="store_true",
dest="show_failures",
help="list tracks that did not match a {} ID".format(
self.data_source
),
help=f"list tracks that did not match a {self.data_source} ID",
)
spotify_cmd.func = queries
@ -647,9 +646,7 @@ class SpotifyPlugin(
spotify_ids = [track_data["id"] for track_data in results]
if self.config["mode"].get() == "open":
self._log.info(
"Attempting to open {} with playlist".format(
self.data_source
)
f"Attempting to open {self.data_source} with playlist"
)
spotify_url = "spotify:trackset:Playlist:" + ",".join(
spotify_ids

View file

@ -168,9 +168,7 @@ class SubsonicPlaylistPlugin(BeetsPlugin):
params["v"] = "1.12.0"
params["c"] = "beets"
resp = requests.get(
"{}/rest/{}?{}".format(
self.config["base_url"].get(), endpoint, urlencode(params)
),
f"{self.config['base_url'].get()}/rest/{endpoint}?{urlencode(params)}",
timeout=10,
)
return resp

View file

@ -202,7 +202,7 @@ class ThumbnailsPlugin(BeetsPlugin):
artfile = os.path.split(album.artpath)[1]
with open(syspath(outfilename), "w") as f:
f.write("[Desktop Entry]\n")
f.write("Icon=./{}".format(artfile.decode("utf-8")))
f.write(f"Icon=./{artfile.decode('utf-8')}")
f.close()
self._log.debug("Wrote file {0}", displayable_path(outfilename))
@ -266,9 +266,7 @@ class GioURI(URIGetter):
g_file_ptr = self.libgio.g_file_new_for_path(path)
if not g_file_ptr:
raise RuntimeError(
"No gfile pointer received for {}".format(
displayable_path(path)
)
f"No gfile pointer received for {displayable_path(path)}"
)
try:

View file

@ -44,6 +44,6 @@ class TypesPlugin(BeetsPlugin):
mytypes[key] = types.DATE
else:
raise ConfigValueError(
"unknown type '{}' for the '{}' field".format(value, key)
f"unknown type '{value}' for the '{key}' field"
)
return mytypes

View file

@ -232,9 +232,7 @@ def _get_unique_table_field_values(model, field, sort_field):
raise KeyError
with g.lib.transaction() as tx:
rows = tx.query(
"SELECT DISTINCT '{}' FROM '{}' ORDER BY '{}'".format(
field, model._table, sort_field
)
f"SELECT DISTINCT '{field}' FROM '{model._table}' ORDER BY '{sort_field}'"
)
return [row[0] for row in rows]

View file

@ -280,6 +280,7 @@ select = [
"PT", # flake8-pytest-style
# "RUF", # ruff
# "UP", # pyupgrade
"UP032", # use f-string instead of format call
"TCH", # flake8-type-checking
"W", # pycodestyle
]

View file

@ -89,8 +89,8 @@ class CAAHelper:
MBID_RELASE = "rid"
MBID_GROUP = "rgid"
RELEASE_URL = "coverartarchive.org/release/{}".format(MBID_RELASE)
GROUP_URL = "coverartarchive.org/release-group/{}".format(MBID_GROUP)
RELEASE_URL = f"coverartarchive.org/release/{MBID_RELASE}"
GROUP_URL = f"coverartarchive.org/release-group/{MBID_GROUP}"
RELEASE_URL = "https://" + RELEASE_URL
GROUP_URL = "https://" + GROUP_URL
@ -305,10 +305,8 @@ class FSArtTest(UseThePlugin):
class CombinedTest(FetchImageTestCase, CAAHelper):
ASIN = "xxxx"
MBID = "releaseid"
AMAZON_URL = "https://images.amazon.com/images/P/{}.01.LZZZZZZZ.jpg".format(
ASIN
)
AAO_URL = "https://www.albumart.org/index_detail.php?asin={}".format(ASIN)
AMAZON_URL = f"https://images.amazon.com/images/P/{ASIN}.01.LZZZZZZZ.jpg"
AAO_URL = f"https://www.albumart.org/index_detail.php?asin={ASIN}"
def setUp(self):
super().setUp()

View file

@ -49,14 +49,12 @@ class ConvertMixin:
"""
if re.search("[^a-zA-Z0-9]", tag):
raise ValueError(
"tag '{}' must only contain letters and digits".format(tag)
f"tag '{tag}' must only contain letters and digits"
)
# A Python script that copies the file and appends a tag.
stub = os.path.join(_common.RSRC, b"convert_stub.py").decode("utf-8")
return "{} {} $source $dest {}".format(
shell_quote(sys.executable), shell_quote(stub), tag
)
return f"{shell_quote(sys.executable)} {shell_quote(stub)} $source $dest {tag}"
def file_endswith(self, path: Path, tag: str):
"""Check the path is a file and if its content ends with `tag`."""

View file

@ -144,9 +144,7 @@ class EmbedartCliTest(IOMixin, PluginMixin, FetchImageHelper, BeetsTestCase):
if os.path.isfile(syspath(tmp_path)):
os.remove(syspath(tmp_path))
self.fail(
"Artwork file {} was not deleted".format(
displayable_path(tmp_path)
)
f"Artwork file {displayable_path(tmp_path)} was not deleted"
)
def test_art_file_missing(self):

View file

@ -37,7 +37,7 @@ class IPFSPluginTest(PluginTestCase):
try:
if check_item.get("ipfs", with_album=False):
ipfs_item = os.fsdecode(os.path.basename(want_item.path))
want_path = "/ipfs/{}/{}".format(test_album.ipfs, ipfs_item)
want_path = f"/ipfs/{test_album.ipfs}/{ipfs_item}"
want_path = bytestring_path(want_path)
assert check_item.path == want_path
assert (

View file

@ -96,9 +96,7 @@ class PlayPluginTest(CleanupModulesMixin, PluginTestCase):
open_mock.assert_called_once_with(ANY, open_anything())
with open(open_mock.call_args[0][0][0], "rb") as f:
playlist = f.read().decode("utf-8")
assert (
f"{os.path.dirname(self.item.path.decode('utf-8'))}\n" == playlist
)
assert f"{self.item.filepath.parent}\n" == playlist
def test_raw(self, open_mock):
self.config["play"]["raw"] = True
@ -125,9 +123,7 @@ class PlayPluginTest(CleanupModulesMixin, PluginTestCase):
self.config["play"]["warning_threshold"] = 1
self.other_item = self.add_item(title="another NiceTitle")
expected_playlist = "{}\n{}".format(
self.item.path.decode("utf-8"), self.other_item.path.decode("utf-8")
)
expected_playlist = f"{self.item.filepath}\n{self.other_item.filepath}"
with control_stdin("a"):
self.run_and_assert(

View file

@ -91,14 +91,7 @@ class PlaylistQueryTest:
assert {i.title for i in results} == {"some item", "another item"}
def test_path_query_with_absolute_paths_in_playlist(self):
q = "playlist:{}".format(
quote(
os.path.join(
self.playlist_dir,
"absolute.m3u",
)
)
)
q = f"playlist:{quote(os.path.join(self.playlist_dir, 'absolute.m3u'))}"
results = self.lib.items(q)
assert {i.title for i in results} == {"some item", "another item"}
@ -108,14 +101,7 @@ class PlaylistQueryTest:
assert {i.title for i in results} == {"some item", "another item"}
def test_path_query_with_relative_paths_in_playlist(self):
q = "playlist:{}".format(
quote(
os.path.join(
self.playlist_dir,
"relative.m3u",
)
)
)
q = f"playlist:{quote(os.path.join(self.playlist_dir, 'relative.m3u'))}"
results = self.lib.items(q)
assert {i.title for i in results} == {"some item", "another item"}
@ -125,15 +111,7 @@ class PlaylistQueryTest:
assert set(results) == set()
def test_path_query_with_nonexisting_playlist(self):
q = "playlist:{}".format(
quote(
os.path.join(
self.playlist_dir,
self.playlist_dir,
"nonexisting.m3u",
)
)
)
q = f"playlist:{os.path.join(self.playlist_dir, 'nonexisting.m3u')!r}"
results = self.lib.items(q)
assert set(results) == set()
@ -141,20 +119,22 @@ class PlaylistQueryTest:
class PlaylistTestRelativeToLib(PlaylistQueryTest, PlaylistTestCase):
def setup_test(self):
with open(os.path.join(self.playlist_dir, "absolute.m3u"), "w") as f:
f.write(
"{}\n".format(os.path.join(self.music_dir, "a", "b", "c.mp3"))
)
f.write(
"{}\n".format(os.path.join(self.music_dir, "d", "e", "f.mp3"))
)
f.write(
"{}\n".format(os.path.join(self.music_dir, "nonexisting.mp3"))
f.writelines(
[
os.path.join(self.music_dir, "a", "b", "c.mp3") + "\n",
os.path.join(self.music_dir, "d", "e", "f.mp3") + "\n",
os.path.join(self.music_dir, "nonexisting.mp3") + "\n",
]
)
with open(os.path.join(self.playlist_dir, "relative.m3u"), "w") as f:
f.write("{}\n".format(os.path.join("a", "b", "c.mp3")))
f.write("{}\n".format(os.path.join("d", "e", "f.mp3")))
f.write("{}\n".format("nonexisting.mp3"))
f.writelines(
[
os.path.join("a", "b", "c.mp3") + "\n",
os.path.join("d", "e", "f.mp3") + "\n",
"nonexisting.mp3" + "\n",
]
)
self.config["playlist"]["relative_to"] = "library"
@ -162,20 +142,22 @@ class PlaylistTestRelativeToLib(PlaylistQueryTest, PlaylistTestCase):
class PlaylistTestRelativeToDir(PlaylistQueryTest, PlaylistTestCase):
def setup_test(self):
with open(os.path.join(self.playlist_dir, "absolute.m3u"), "w") as f:
f.write(
"{}\n".format(os.path.join(self.music_dir, "a", "b", "c.mp3"))
)
f.write(
"{}\n".format(os.path.join(self.music_dir, "d", "e", "f.mp3"))
)
f.write(
"{}\n".format(os.path.join(self.music_dir, "nonexisting.mp3"))
f.writelines(
[
os.path.join(self.music_dir, "a", "b", "c.mp3") + "\n",
os.path.join(self.music_dir, "d", "e", "f.mp3") + "\n",
os.path.join(self.music_dir, "nonexisting.mp3") + "\n",
]
)
with open(os.path.join(self.playlist_dir, "relative.m3u"), "w") as f:
f.write("{}\n".format(os.path.join("a", "b", "c.mp3")))
f.write("{}\n".format(os.path.join("d", "e", "f.mp3")))
f.write("{}\n".format("nonexisting.mp3"))
f.writelines(
[
os.path.join("a", "b", "c.mp3") + "\n",
os.path.join("d", "e", "f.mp3") + "\n",
"nonexisting.mp3" + "\n",
]
)
self.config["playlist"]["relative_to"] = self.music_dir
@ -183,40 +165,33 @@ class PlaylistTestRelativeToDir(PlaylistQueryTest, PlaylistTestCase):
class PlaylistTestRelativeToPls(PlaylistQueryTest, PlaylistTestCase):
def setup_test(self):
with open(os.path.join(self.playlist_dir, "absolute.m3u"), "w") as f:
f.write(
"{}\n".format(os.path.join(self.music_dir, "a", "b", "c.mp3"))
)
f.write(
"{}\n".format(os.path.join(self.music_dir, "d", "e", "f.mp3"))
)
f.write(
"{}\n".format(os.path.join(self.music_dir, "nonexisting.mp3"))
f.writelines(
[
os.path.join(self.music_dir, "a", "b", "c.mp3") + "\n",
os.path.join(self.music_dir, "d", "e", "f.mp3") + "\n",
os.path.join(self.music_dir, "nonexisting.mp3") + "\n",
]
)
with open(os.path.join(self.playlist_dir, "relative.m3u"), "w") as f:
f.write(
"{}\n".format(
f.writelines(
[
os.path.relpath(
os.path.join(self.music_dir, "a", "b", "c.mp3"),
start=self.playlist_dir,
)
)
)
f.write(
"{}\n".format(
+ "\n",
os.path.relpath(
os.path.join(self.music_dir, "d", "e", "f.mp3"),
start=self.playlist_dir,
)
)
)
f.write(
"{}\n".format(
+ "\n",
os.path.relpath(
os.path.join(self.music_dir, "nonexisting.mp3"),
start=self.playlist_dir,
)
)
+ "\n",
]
)
self.config["playlist"]["relative_to"] = "playlist"
@ -226,20 +201,22 @@ class PlaylistTestRelativeToPls(PlaylistQueryTest, PlaylistTestCase):
class PlaylistUpdateTest:
def setup_test(self):
with open(os.path.join(self.playlist_dir, "absolute.m3u"), "w") as f:
f.write(
"{}\n".format(os.path.join(self.music_dir, "a", "b", "c.mp3"))
)
f.write(
"{}\n".format(os.path.join(self.music_dir, "d", "e", "f.mp3"))
)
f.write(
"{}\n".format(os.path.join(self.music_dir, "nonexisting.mp3"))
f.writelines(
[
os.path.join(self.music_dir, "a", "b", "c.mp3") + "\n",
os.path.join(self.music_dir, "d", "e", "f.mp3") + "\n",
os.path.join(self.music_dir, "nonexisting.mp3") + "\n",
]
)
with open(os.path.join(self.playlist_dir, "relative.m3u"), "w") as f:
f.write("{}\n".format(os.path.join("a", "b", "c.mp3")))
f.write("{}\n".format(os.path.join("d", "e", "f.mp3")))
f.write("{}\n".format("nonexisting.mp3"))
f.writelines(
[
os.path.join("a", "b", "c.mp3") + "\n",
os.path.join("d", "e", "f.mp3") + "\n",
"nonexisting.mp3" + "\n",
]
)
self.config["playlist"]["auto"] = True
self.config["playlist"]["relative_to"] = "library"
@ -249,9 +226,7 @@ class PlaylistTestItemMoved(PlaylistUpdateTest, PlaylistTestCase):
def test_item_moved(self):
# Emit item_moved event for an item that is in a playlist
results = self.lib.items(
"path:{}".format(
quote(os.path.join(self.music_dir, "d", "e", "f.mp3"))
)
f"path:{quote(os.path.join(self.music_dir, 'd', 'e', 'f.mp3'))}"
)
item = results[0]
beets.plugins.send(
@ -265,9 +240,7 @@ class PlaylistTestItemMoved(PlaylistUpdateTest, PlaylistTestCase):
# Emit item_moved event for an item that is not in a playlist
results = self.lib.items(
"path:{}".format(
quote(os.path.join(self.music_dir, "x", "y", "z.mp3"))
)
f"path:{quote(os.path.join(self.music_dir, 'x', 'y', 'z.mp3'))}"
)
item = results[0]
beets.plugins.send(
@ -309,18 +282,14 @@ class PlaylistTestItemRemoved(PlaylistUpdateTest, PlaylistTestCase):
def test_item_removed(self):
# Emit item_removed event for an item that is in a playlist
results = self.lib.items(
"path:{}".format(
quote(os.path.join(self.music_dir, "d", "e", "f.mp3"))
)
f"path:{quote(os.path.join(self.music_dir, 'd', 'e', 'f.mp3'))}"
)
item = results[0]
beets.plugins.send("item_removed", item=item)
# Emit item_removed event for an item that is not in a playlist
results = self.lib.items(
"path:{}".format(
quote(os.path.join(self.music_dir, "x", "y", "z.mp3"))
)
f"path:{quote(os.path.join(self.music_dir, 'x', 'y', 'z.mp3'))}"
)
item = results[0]
beets.plugins.send("item_removed", item=item)

View file

@ -69,7 +69,7 @@ class RandomTest(TestHelper, unittest.TestCase):
# Print a histogram (useful for debugging).
if histogram:
for i in range(len(self.items)):
print("{:2d} {}".format(i, "*" * positions.count(i)))
print(f"{i:2d} {'*' * positions.count(i)}")
return self._stats(positions)
mean1, stdev1, median1 = experiment("artist")

View file

@ -204,9 +204,7 @@ class ReplayGainCliTest:
# This test is a lot less interesting if the backend cannot write
# both tag types.
self.skipTest(
"r128 tags for opus not supported on backend {}".format(
self.backend
)
f"r128 tags for opus not supported on backend {self.backend}"
)
album_rg = self._add_album(1)
@ -263,9 +261,7 @@ class ReplayGainCliTest:
def test_cli_writes_only_r128_tags(self):
if not self.has_r128_support:
self.skipTest(
"r128 tags for opus not supported on backend {}".format(
self.backend
)
f"r128 tags for opus not supported on backend {self.backend}"
)
album = self._add_album(2, ext="opus")
@ -299,9 +295,7 @@ class ReplayGainCliTest:
def test_r128_targetlevel_has_effect(self):
if not self.has_r128_support:
self.skipTest(
"r128 tags for opus not supported on backend {}".format(
self.backend
)
f"r128 tags for opus not supported on backend {self.backend}"
)
album = self._add_album(1, ext="opus")