mirror of
https://github.com/beetbox/beets.git
synced 2026-07-16 14:07:40 -04:00
typing: feed PathLike as database path
This commit is contained in:
@@ -15,6 +15,7 @@ from collections.abc import Mapping
|
||||
from contextlib import contextmanager
|
||||
from dataclasses import dataclass
|
||||
from functools import cached_property
|
||||
from pathlib import Path
|
||||
from sqlite3 import Connection, sqlite_version_info
|
||||
from typing import (
|
||||
TYPE_CHECKING,
|
||||
@@ -52,8 +53,7 @@ if TYPE_CHECKING:
|
||||
from sqlite3 import Connection
|
||||
from types import TracebackType
|
||||
|
||||
from beets.util import PathLike
|
||||
|
||||
from ..util import PathLike
|
||||
from .query import FieldQueryType, Query, SQLiteType
|
||||
from .sort import FieldSort, Sort
|
||||
|
||||
@@ -1085,6 +1085,8 @@ class Database:
|
||||
data is written in a transaction.
|
||||
"""
|
||||
|
||||
path: Path
|
||||
|
||||
def __init__(self, path: PathLike, timeout: float = 5.0) -> None:
|
||||
if sqlite3.threadsafety == 0:
|
||||
raise RuntimeError(
|
||||
@@ -1098,7 +1100,7 @@ class Database:
|
||||
if hasattr(sqlite3, "enable_callback_tracebacks"):
|
||||
sqlite3.enable_callback_tracebacks(True)
|
||||
|
||||
self.path = path
|
||||
self.path = Path(os.fsdecode(path))
|
||||
self.timeout = timeout
|
||||
|
||||
self._connections: dict[int, sqlite3.Connection] = {}
|
||||
|
||||
@@ -3,6 +3,7 @@ from __future__ import annotations
|
||||
import re
|
||||
from contextlib import contextmanager
|
||||
from functools import cached_property
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import platformdirs
|
||||
@@ -20,7 +21,7 @@ from .queries import parse_query_parts, parse_query_string
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from beets.dbcore import Results
|
||||
from beets.util import Replacements
|
||||
from beets.util import PathLike, Replacements
|
||||
from beets.util.pathformats import PathFormat
|
||||
|
||||
|
||||
@@ -61,7 +62,7 @@ class Library(dbcore.Database):
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
path="library.blb",
|
||||
path: PathLike = Path("library.blb"),
|
||||
directory: str | None = None,
|
||||
set_music_dir: bool = True,
|
||||
):
|
||||
|
||||
@@ -270,9 +270,9 @@ class TestHelper(RunMixin, PathsMixin, ConfigMixin):
|
||||
self.config["directory"] = str(self.lib_path)
|
||||
|
||||
dbpath = (
|
||||
util.bytestring_path(self.config["library"].as_filename())
|
||||
self.config["library"].as_path()
|
||||
if self.db_on_disk
|
||||
else ":memory:"
|
||||
else Path(":memory:")
|
||||
)
|
||||
self.lib = Library(dbpath, str(self.lib_path))
|
||||
|
||||
|
||||
@@ -767,9 +767,12 @@ def _setup() -> tuple[list[Subcommand], library.Library]:
|
||||
|
||||
|
||||
def _ensure_db_directory_exists(path):
|
||||
if path == b":memory:": # in memory db
|
||||
dbpath = os.fspath(path)
|
||||
if dbpath in (":memory:", b":memory:"): # in memory db
|
||||
return
|
||||
newpath = os.path.dirname(dbpath)
|
||||
if not newpath:
|
||||
return
|
||||
newpath = os.path.dirname(path)
|
||||
if not os.path.isdir(newpath):
|
||||
if input_yn(
|
||||
f"The database directory {util.displayable_path(newpath)} does not"
|
||||
@@ -780,7 +783,7 @@ def _ensure_db_directory_exists(path):
|
||||
|
||||
def _open_library(config: confuse.LazyConfig) -> library.Library:
|
||||
"""Create a new library instance from the configuration."""
|
||||
dbpath = util.bytestring_path(config["library"].as_filename())
|
||||
dbpath = config["library"].as_path()
|
||||
_ensure_db_directory_exists(dbpath)
|
||||
try:
|
||||
lib = library.Library(dbpath, config["directory"].as_filename())
|
||||
|
||||
@@ -200,8 +200,7 @@ class IPFSPlugin(BeetsPlugin):
|
||||
lib_name = args[1]
|
||||
else:
|
||||
lib_name = _hash
|
||||
lib_root = os.path.dirname(lib.path)
|
||||
remote_libs = os.path.join(lib_root, b"remotes")
|
||||
remote_libs = self._remote_libs_path(lib)
|
||||
if not os.path.exists(remote_libs):
|
||||
try:
|
||||
os.makedirs(remote_libs)
|
||||
@@ -255,9 +254,12 @@ class IPFSPlugin(BeetsPlugin):
|
||||
rlib = self.get_remote_lib(lib)
|
||||
return rlib.albums(args)
|
||||
|
||||
def _remote_libs_path(self, lib):
|
||||
lib_root = os.path.dirname(os.fsencode(lib.path))
|
||||
return os.path.join(lib_root, b"remotes")
|
||||
|
||||
def get_remote_lib(self, lib):
|
||||
lib_root = os.path.dirname(lib.path)
|
||||
remote_libs = os.path.join(lib_root, b"remotes")
|
||||
remote_libs = self._remote_libs_path(lib)
|
||||
path = os.path.join(remote_libs, b"joined.db")
|
||||
if not os.path.isfile(path):
|
||||
raise OSError
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
import os
|
||||
import shutil
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from tempfile import mkstemp
|
||||
from typing import ClassVar
|
||||
|
||||
@@ -90,6 +91,28 @@ class DatabaseFixtureTwoModels(dbcore.Database):
|
||||
_models = (ModelFixture2, AnotherModelFixture)
|
||||
|
||||
|
||||
class TestDatabasePath:
|
||||
@pytest.mark.parametrize(
|
||||
"path", [":memory:", b":memory:", Path(":memory:")]
|
||||
)
|
||||
def test_path_inputs_are_stored_as_path(self, path):
|
||||
db = DatabaseFixture1(path)
|
||||
try:
|
||||
assert db.path == Path(":memory:")
|
||||
assert isinstance(db.path, Path)
|
||||
finally:
|
||||
db._connection().close()
|
||||
|
||||
def test_bytes_filesystem_path_opens_decoded_path(self, tmp_path):
|
||||
db_path = tmp_path / "library.db"
|
||||
db = DatabaseFixture1(os.fsencode(db_path))
|
||||
try:
|
||||
assert db.path == db_path
|
||||
assert db_path.exists()
|
||||
finally:
|
||||
db._connection().close()
|
||||
|
||||
|
||||
class ModelFixtureWithGetters(dbcore.Model):
|
||||
@classmethod
|
||||
def _getters(cls):
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import os
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
from beets import util
|
||||
from beets import library, util
|
||||
from beets.test import _common
|
||||
from beets.test.helper import PluginTestCase
|
||||
from beetsplug.ipfs import IPFSPlugin
|
||||
@@ -35,6 +35,21 @@ class IPFSPluginTest(PluginTestCase):
|
||||
found = True
|
||||
assert found
|
||||
|
||||
def test_get_remote_lib_accepts_library_path(self):
|
||||
self.lib.path = self.temp_dir_path / "library.db"
|
||||
remote_dir = self.temp_dir_path / "remotes"
|
||||
remote_dir.mkdir()
|
||||
|
||||
remote_lib = library.Library(remote_dir / "joined.db")
|
||||
remote_lib._close()
|
||||
|
||||
ipfs = IPFSPlugin()
|
||||
added_lib = ipfs.get_remote_lib(self.lib)
|
||||
try:
|
||||
assert added_lib.path == remote_dir / "joined.db"
|
||||
finally:
|
||||
added_lib._close()
|
||||
|
||||
def mk_test_album(self):
|
||||
items = [_common.item() for _ in range(3)]
|
||||
items[0].title = "foo bar"
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
import os
|
||||
import unittest
|
||||
from copy import deepcopy
|
||||
from pathlib import Path
|
||||
from random import random
|
||||
|
||||
import pytest
|
||||
@@ -74,6 +75,10 @@ class InputMethodsTest(IOMixin, unittest.TestCase):
|
||||
|
||||
|
||||
class ParentalDirCreation(IOMixin, BeetsTestCase):
|
||||
def test_memory_path_skips_creation_prompt(self):
|
||||
ui._ensure_db_directory_exists(Path(":memory:"))
|
||||
assert not self.io.getoutput()
|
||||
|
||||
def test_create_yes(self):
|
||||
non_exist_path = _common.os.fsdecode(
|
||||
os.path.join(self.temp_dir, b"nonexist", str(random()).encode())
|
||||
|
||||
Reference in New Issue
Block a user