Replace bytestring_path with builtin os.fsencode

Same `os.fsdecode`, this has only been available since Python 3.2 which
was after this code was written.

Author: snejus

beets/art.py

@@ -22,7 +22,7 @@
 
 import mediafile
 
-from beets.util import bytestring_path, displayable_path, syspath
+from beets.util import displayable_path, syspath
 from beets.util.artresizer import ArtResizer
 
 
@@ -183,7 +183,7 @@ def check_art_similarity(
 
 def extract(log, outpath, item):
     art = get_art(log, item)
-    outpath = bytestring_path(outpath)
+    outpath = os.fsencode(outpath)
     if not art:
         log.info("No album art present in {0}, skipping.", item)
         return
@@ -193,7 +193,7 @@ def extract(log, outpath, item):
     if not ext:
         log.warning("Unknown image type in {0}.", displayable_path(item.path))
         return
-    outpath += bytestring_path("." + ext)
+    outpath += os.fsencode("." + ext)
 
     log.info(
         "Extracting album art from: {0} to: {1}",

beets/importer.py

@@ -1800,7 +1800,7 @@ def albums_in_dir(path):
                 start_collapsing = True
                 subdir_pat = None
                 for subdir in dirs:
-                    subdir = util.bytestring_path(subdir)
+                    subdir = os.fsencode(subdir)
                     # The first directory dictates the pattern for
                     # the remaining directories.
                     if not subdir_pat:

beets/library.py

@@ -28,14 +28,7 @@
 import beets
 from beets import dbcore, logging, plugins, util
 from beets.dbcore import Results, types
-from beets.util import (
-    MoveOperation,
-    bytestring_path,
-    lazy_property,
-    normpath,
-    samefile,
-    syspath,
-)
+from beets.util import MoveOperation, lazy_property, normpath, samefile, syspath
 from beets.util.functemplate import Template, template
 
 # To use the SQLite "blob" type, it doesn't suffice to provide a byte
@@ -214,12 +207,12 @@ def format(self, value):
         return util.displayable_path(value)
 
     def parse(self, string):
-        return normpath(bytestring_path(string))
+        return normpath(os.fsencode(string))
 
     def normalize(self, value):
         if isinstance(value, str):
             # Paths stored internally as encoded bytes.
-            return bytestring_path(value)
+            return os.fsencode(value)
 
         elif isinstance(value, BLOB_TYPE):
             # We unwrap buffers to bytes.
@@ -681,7 +674,7 @@ def __setitem__(self, key, value):
         # Encode unicode paths and read buffers.
         if key == "path":
             if isinstance(value, str):
-                value = bytestring_path(value)
+                value = os.fsencode(value)
             elif isinstance(value, BLOB_TYPE):
                 value = bytes(value)
         elif key == "album_id":
@@ -1402,7 +1395,7 @@ def art_destination(self, image, item_dir=None):
         items, so the album must contain at least one item or
         item_dir must be provided.
         """
-        image = bytestring_path(image)
+        image = os.fsencode(image)
         item_dir = item_dir or self.item_dir()
 
         filename_tmpl = template(beets.config["art_filename"].as_str())
@@ -1414,12 +1407,12 @@ def art_destination(self, image, item_dir=None):
         subpath = util.sanitize_path(
             subpath, replacements=self._db.replacements
         )
-        subpath = bytestring_path(subpath)
+        subpath = os.fsencode(subpath)
 
         _, ext = os.path.splitext(image)
         dest = os.path.join(item_dir, subpath + ext)
 
-        return bytestring_path(dest)
+        return os.fsencode(dest)
 
     def set_art(self, path, copy=True):
         """Set the album's cover art to the image at the given path.
@@ -1429,7 +1422,7 @@ def set_art(self, path, copy=True):
 
         Send an 'art_set' event with `self` as the sole argument.
         """
-        path = bytestring_path(path)
+        path = os.fsencode(path)
         oldart = self.artpath
         artdest = self.art_destination(path)
 
@@ -1568,7 +1561,7 @@ def __init__(
         timeout = beets.config["timeout"].as_number()
         super().__init__(path, timeout=timeout)
 
-        self.directory = bytestring_path(normpath(directory))
+        self.directory = os.fsencode(normpath(directory))
         self.path_formats = path_formats
         self.replacements = replacements
 

beets/test/_common.py

@@ -30,7 +30,7 @@
 from beets import util  # noqa: E402
 from beets import importer, logging  # noqa: E402
 from beets.ui import commands  # noqa: E402
-from beets.util import bytestring_path, syspath  # noqa: E402
+from beets.util import syspath  # noqa: E402
 
 beetsplug.__path__ = [
     os.path.abspath(
@@ -44,7 +44,7 @@
 ]
 
 # Test resources path.
-RSRC = util.bytestring_path(
+RSRC = os.fsencode(
     os.path.abspath(
         os.path.join(
             os.path.dirname(__file__),
@@ -207,7 +207,7 @@ def setUp(self):
 
         # Direct paths to a temporary directory. Tests can also use this
         # temporary directory.
-        self.temp_dir = util.bytestring_path(tempfile.mkdtemp())
+        self.temp_dir = os.fsencode(tempfile.mkdtemp())
 
         beets.config["statefile"] = os.fsdecode(
             os.path.join(self.temp_dir, b"state.pickle")
@@ -399,7 +399,7 @@ def create_temp_dir(self):
         """Create a temporary directory and assign it into `self.temp_dir`.
         Call `remove_temp_dir` later to delete it.
         """
-        self.temp_dir = bytestring_path(tempfile.mkdtemp())
+        self.temp_dir = os.fsencode(tempfile.mkdtemp())
 
     def remove_temp_dir(self):
         """Delete the temporary directory created by `create_temp_dir`."""

beets/test/helper.py

@@ -49,7 +49,7 @@
 from beets.library import Album, Item, Library
 from beets.test import _common
 from beets.ui.commands import TerminalImportSession
-from beets.util import MoveOperation, bytestring_path, syspath
+from beets.util import MoveOperation, syspath
 
 
 class LogCapture(logging.Handler):
@@ -182,7 +182,7 @@ def setup_beets(self, disk=False):
         self.config["directory"] = os.fsdecode(self.libdir)
 
         if disk:
-            dbpath = util.bytestring_path(self.config["library"].as_filename())
+            dbpath = os.fsencode(self.config["library"].as_filename())
         else:
             dbpath = ":memory:"
         self.lib = Library(dbpath, self.libdir)
@@ -241,7 +241,7 @@ def create_importer(self, item_count=1, album_count=1):
 
         album_no = 0
         while album_count:
-            album = util.bytestring_path(f"album {album_no}")
+            album = os.fsencode(f"album {album_no}")
             album_dir = os.path.join(import_dir, album)
             if os.path.exists(syspath(album_dir)):
                 album_no += 1
@@ -254,7 +254,7 @@ def create_importer(self, item_count=1, album_count=1):
             while album_item_count:
                 title = f"track {track_no}"
                 src = os.path.join(_common.RSRC, b"full.mp3")
-                title_file = util.bytestring_path(f"{title}.mp3")
+                title_file = os.fsencode(f"{title}.mp3")
                 dest = os.path.join(album_dir, title_file)
                 if os.path.exists(syspath(dest)):
                     track_no += 1
@@ -341,7 +341,7 @@ def add_item_fixture(self, **values):
         item = self.create_item(**values)
         extension = item["format"].lower()
         item["path"] = os.path.join(
-            _common.RSRC, util.bytestring_path("min." + extension)
+            _common.RSRC, os.fsencode("min." + extension)
         )
         item.add(self.lib)
         item.move(operation=MoveOperation.COPY)
@@ -356,7 +356,7 @@ def add_item_fixtures(self, ext="mp3", count=1):
         """Add a number of items with files to the database."""
         # TODO base this on `add_item()`
         items = []
-        path = os.path.join(_common.RSRC, util.bytestring_path("full." + ext))
+        path = os.path.join(_common.RSRC, os.fsencode("full." + ext))
         for i in range(count):
             item = Item.from_path(path)
             item.album = f"\u00e4lbum {i}"  # Check unicode paths
@@ -380,7 +380,7 @@ def add_album_fixture(
         items = []
         path = os.path.join(
             _common.RSRC,
-            util.bytestring_path(f"{fname}.{ext}"),
+            os.fsencode(f"{fname}.{ext}"),
         )
         for discnumber in range(1, disc_count + 1):
             for i in range(track_count):
@@ -407,17 +407,17 @@ def create_mediafile_fixture(self, ext="mp3", images=[]):
         specified extension a cover art image is added to the media
         file.
         """
-        src = os.path.join(_common.RSRC, util.bytestring_path("full." + ext))
+        src = os.path.join(_common.RSRC, os.fsencode("full." + ext))
         handle, path = mkstemp()
-        path = bytestring_path(path)
+        path = os.fsencode(path)
         os.close(handle)
         shutil.copyfile(syspath(src), syspath(path))
 
         if images:
             mediafile = MediaFile(path)
             imgs = []
             for img_ext in images:
-                file = util.bytestring_path(f"image-2x3.{img_ext}")
+                file = os.fsencode(f"image-2x3.{img_ext}")
                 img_path = os.path.join(_common.RSRC, file)
                 with open(img_path, "rb") as f:
                     imgs.append(Image(f.read()))
@@ -467,7 +467,7 @@ def create_temp_dir(self):
         `self.temp_dir`. Call `remove_temp_dir` later to delete it.
         """
         temp_dir = mkdtemp()
-        self.temp_dir = util.bytestring_path(temp_dir)
+        self.temp_dir = os.fsencode(temp_dir)
 
     def remove_temp_dir(self):
         """Delete the temporary directory created by `create_temp_dir`."""
@@ -544,7 +544,7 @@ def _create_import_dir(self, count=3):
         for i in range(count):
             # Copy files
             medium_path = os.path.join(
-                album_path, bytestring_path("track_%d.mp3" % (i + 1))
+                album_path, os.fsencode("track_%d.mp3" % (i + 1))
             )
             shutil.copy(syspath(resource_path), syspath(medium_path))
             medium = MediaFile(medium_path)

beets/ui/__init__.py

@@ -1753,7 +1753,7 @@ def _ensure_db_directory_exists(path):
 
 def _open_library(config):
     """Create a new library instance from the configuration."""
-    dbpath = util.bytestring_path(config["library"].as_filename())
+    dbpath = os.fsencode(config["library"].as_filename())
     _ensure_db_directory_exists(dbpath)
     try:
         lib = library.Library(

beets/util/__init__.py

@@ -160,7 +160,7 @@ def normpath(path: bytes) -> bytes:
     """
     path = syspath(path, prefix=False)
     path = os.path.normpath(os.path.abspath(os.path.expanduser(path)))
-    return bytestring_path(path)
+    return os.fsencode(path)
 
 
 def ancestry(path: bytes) -> List[str]:
@@ -199,8 +199,8 @@ def sorted_walk(
     warning messages are logged there when a directory cannot be listed.
     """
     # Make sure the paths aren't Unicode strings.
-    path = bytestring_path(path)
-    ignore = [bytestring_path(i) for i in ignore]
+    path = os.fsencode(path)
+    ignore = [os.fsencode(i) for i in ignore]
 
     # Get all the directories and files at this level.
     try:
@@ -216,7 +216,7 @@ def sorted_walk(
     dirs = []
     files = []
     for base in contents:
-        base = bytestring_path(base)
+        base = os.fsencode(base)
 
         # Skip ignored filenames.
         skip = False
@@ -322,8 +322,8 @@ def prune_dirs(
         if not os.path.exists(directory):
             # Directory gone already.
             continue
-        clutter: List[bytes] = [bytestring_path(c) for c in clutter]
-        match_paths = [bytestring_path(d) for d in os.listdir(directory)]
+        clutter: List[bytes] = [os.fsencode(c) for c in clutter]
+        match_paths = [os.fsencode(d) for d in os.listdir(directory)]
         try:
             if fnmatch_all(match_paths, clutter):
                 # Directory contains only clutter (or nothing).
@@ -380,29 +380,7 @@ def _fsencoding() -> str:
     return encoding
 
 
-def bytestring_path(path: Bytes_or_String) -> bytes:
-    """Given a path, which is either a bytes or a unicode, returns a str
-    path (ensuring that we never deal with Unicode pathnames). Path should be
-    bytes but has safeguards for strings to be converted.
-    """
-    # Pass through bytestrings.
-    if isinstance(path, bytes):
-        return path
-
-    # On Windows, remove the magic prefix added by `syspath`. This makes
-    # ``bytestring_path(syspath(X)) == X``, i.e., we can safely
-    # round-trip through `syspath`.
-    if os.path.__name__ == "ntpath" and path.startswith(WINDOWS_MAGIC_PREFIX):
-        path = path[len(WINDOWS_MAGIC_PREFIX) :]
-
-    # Try to encode with default encodings, but fall back to utf-8.
-    try:
-        return path.encode(_fsencoding())
-    except (UnicodeError, LookupError):
-        return path.encode("utf-8")
-
-
-PATH_SEP: bytes = bytestring_path(os.sep)
+PATH_SEP: bytes = os.fsencode(os.sep)
 
 
 def displayable_path(
@@ -522,8 +500,8 @@ def move(path: bytes, dest: bytes, replace: bool = False):
         os.replace(syspath(path), syspath(dest))
     except OSError:
         # Copy the file to a temporary destination.
-        basename = os.path.basename(bytestring_path(dest))
-        dirname = os.path.dirname(bytestring_path(dest))
+        basename = os.path.basename(os.fsencode(dest))
+        dirname = os.path.dirname(os.fsencode(dest))
         tmp = tempfile.NamedTemporaryFile(
             suffix=syspath(b".beets", prefix=False),
             prefix=syspath(b"." + basename, prefix=False),
@@ -738,7 +716,7 @@ def _legalize_stage(
 
     # Encode for the filesystem.
     if not fragment:
-        path = bytestring_path(path)  # type: ignore
+        path = os.fsencode(path)  # type: ignore
 
     # Preserve extension.
     path += extension.lower()

beets/util/artresizer.py

@@ -26,7 +26,7 @@
 from urllib.parse import urlencode
 
 from beets import logging, util
-from beets.util import bytestring_path, displayable_path, syspath
+from beets.util import displayable_path, syspath
 
 PROXY_URL = "https://images.weserv.nl/"
 
@@ -54,7 +54,7 @@ def temp_file_for(path):
     """
     ext = os.path.splitext(path)[1]
     with NamedTemporaryFile(suffix=os.fsdecode(ext), delete=False) as f:
-        return bytestring_path(f.name)
+        return os.fsencode(f.name)
 
 
 class LocalBackendNotAvailableError(Exception):