More refactoring for cleanliness.
Signed-off-by: Cliff Hill <xlorep@darkhelm.org>
This commit is contained in:
@@ -4,15 +4,13 @@ from __future__ import annotations
|
||||
import dataclasses
|
||||
import datetime
|
||||
import functools
|
||||
import pathlib
|
||||
import typing
|
||||
|
||||
import desert # type: ignore [import]
|
||||
import marshmallow.schema # type: ignore [import]
|
||||
import yaml
|
||||
|
||||
|
||||
DataSub = typing.NewType("DataSub", "BaseData")
|
||||
DataSub = typing.TypeVar("DataSub")
|
||||
DataDict = typing.TypeVar("DataDict")
|
||||
|
||||
|
||||
@@ -49,42 +47,17 @@ class DataMeta(type):
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class BaseData(metaclass=DataMeta):
|
||||
class BaseData(typing.Generic[DataSub], metaclass=DataMeta):
|
||||
"""Base class for data classes in the application."""
|
||||
|
||||
__sa_dataclass_metadata_key__ = "sa"
|
||||
|
||||
@classmethod
|
||||
def load(cls: type[BaseData], data: DataDict) -> DataSub:
|
||||
def load(cls: type[BaseData[DataSub]], data: DataDict) -> DataSub:
|
||||
"""Load the given data dictionary into a class instance."""
|
||||
return typing.cast(DataSub, cls.Schema.load(data))
|
||||
|
||||
def dump(self: BaseData) -> DataDict:
|
||||
def dump(self: BaseData[DataSub]) -> DataDict:
|
||||
"""Dump the class instance into a data dictionary."""
|
||||
cls = type(self)
|
||||
return typing.cast(DataDict, cls.Schema.dump(self))
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class YAMLData(BaseData):
|
||||
"""Data class base that contains functionality to read/write as YAML."""
|
||||
|
||||
@classmethod
|
||||
def yaml_read(cls: type[YAMLData], filepath: pathlib.Path) -> DataSub:
|
||||
"""Read the given YAML file and convert it into an object."""
|
||||
with filepath.open() as fp:
|
||||
data = yaml.safe_load(fp)
|
||||
return cls.load(data)
|
||||
|
||||
def yaml_write(self: YAMLData, filepath: pathlib.Path) -> None:
|
||||
"""Write this object as the given YAML file."""
|
||||
data: type(self).Dict = self.dump() # type: ignore [valid-type]
|
||||
with filepath.open(mode="w") as fp:
|
||||
fp.write(yaml.dump(data))
|
||||
|
||||
@classmethod
|
||||
def yaml_create(cls: type[YAMLData], filepath: pathlib.Path) -> DataSub:
|
||||
"""Reload the YAML file with this object."""
|
||||
data = cls()
|
||||
data.yaml_write(filepath)
|
||||
return typing.cast(DataSub, data)
|
||||
|
||||
@@ -25,6 +25,7 @@ class Default(enum.Enum):
|
||||
|
||||
PROCESS_POOL = concurrent.futures.ProcessPoolExecutor()
|
||||
MAX_PROCESSES = PROCESS_POOL._max_workers # type: ignore [attr-defined]
|
||||
SETTINGS_FILENAME = "settings.yaml"
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
|
||||
@@ -12,7 +12,7 @@ mapper_registry = sqlalchemy.orm.registry()
|
||||
|
||||
@mapper_registry.mapped
|
||||
@dataclasses.dataclass
|
||||
class Track(base.BaseData):
|
||||
class Track(base.BaseData["Track"]):
|
||||
"""Model defining a Track object."""
|
||||
|
||||
__tablename__ = "tracks"
|
||||
|
||||
@@ -3,19 +3,19 @@ from __future__ import annotations
|
||||
|
||||
import dataclasses
|
||||
import datetime
|
||||
import typing
|
||||
import functools
|
||||
import pathlib
|
||||
|
||||
import yaml
|
||||
|
||||
from playlist.data import base
|
||||
from playlist.data import const
|
||||
from playlist.data import yaml_base
|
||||
|
||||
__all__ = ["get", "write"]
|
||||
|
||||
SETTINGS_FILE = "settings.yaml"
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class CredentialSettings(base.BaseData):
|
||||
class CredentialSettings(base.BaseData["CredentialSettings"]):
|
||||
"""Credentials component for Settings object."""
|
||||
|
||||
baseurl: str
|
||||
@@ -32,11 +32,11 @@ class CredentialSettings(base.BaseData):
|
||||
servername = input("Plex Server: ")
|
||||
|
||||
data = server.get_creds(username, password, servername)
|
||||
return typing.cast(CredentialSettings, cls.load(data))
|
||||
return cls.load(data)
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class DownloadSettings(base.BaseData):
|
||||
class DownloadSettings(base.BaseData["DownloadSettings"]):
|
||||
"""DownloadSettingser options for Settings object."""
|
||||
|
||||
batch_size: int
|
||||
@@ -51,7 +51,7 @@ class DownloadSettings(base.BaseData):
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class TrackSettings(base.BaseData):
|
||||
class TrackSettings(base.BaseData["TrackSettings"]):
|
||||
"""Track options for Settings object."""
|
||||
|
||||
duration: float
|
||||
@@ -62,7 +62,7 @@ class TrackSettings(base.BaseData):
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class PlaylistSettings(base.BaseData):
|
||||
class PlaylistSettings(base.BaseData["PlaylistSettings"]):
|
||||
"""Playlist options for Settings object."""
|
||||
|
||||
playtime: datetime.timedelta
|
||||
@@ -77,7 +77,7 @@ class PlaylistSettings(base.BaseData):
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class Settings(yaml_base.YAMLBase):
|
||||
class Settings(base.BaseData["Settings"]):
|
||||
"""Settings object, loaded from settings.yaml file."""
|
||||
|
||||
creds: CredentialSettings = dataclasses.field(
|
||||
@@ -91,12 +91,48 @@ class Settings(yaml_base.YAMLBase):
|
||||
default_factory=PlaylistSettings.create,
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def yaml_read(
|
||||
cls: type[Settings],
|
||||
filepath: pathlib.Path,
|
||||
) -> Settings:
|
||||
"""Read the given YAML file and convert it into an object."""
|
||||
with filepath.open() as fp:
|
||||
data = yaml.safe_load(fp)
|
||||
return cls.load(data)
|
||||
|
||||
def yaml_write(self: Settings, filepath: pathlib.Path) -> None:
|
||||
"""Write this object as the given YAML file."""
|
||||
data: type(self).Dict = self.dump() # type: ignore [valid-type]
|
||||
with filepath.open(mode="w") as fp:
|
||||
fp.write(yaml.dump(data))
|
||||
|
||||
@classmethod
|
||||
def yaml_create(
|
||||
cls: type[Settings],
|
||||
filepath: pathlib.Path,
|
||||
) -> Settings:
|
||||
"""Create the YAML file with this object."""
|
||||
data = cls()
|
||||
data.yaml_write(filepath)
|
||||
return data
|
||||
|
||||
|
||||
@functools.cache
|
||||
def get() -> Settings:
|
||||
"""Get the Settings object instance."""
|
||||
return typing.cast(Settings, Settings.get(SETTINGS_FILE))
|
||||
instance: Settings
|
||||
filepath = const.PATHS.CONFIG / const.SETTINGS_FILENAME
|
||||
try:
|
||||
instance = Settings.yaml_read(filepath)
|
||||
except FileNotFoundError:
|
||||
const.PATHS.CONFIG.mkdir(parents=True, exist_ok=True)
|
||||
instance = Settings.yaml_create(filepath)
|
||||
|
||||
return instance
|
||||
|
||||
|
||||
def write() -> None:
|
||||
"""Write the Settings object instance."""
|
||||
Settings.write(SETTINGS_FILE)
|
||||
filepath = const.PATHS.CONFIG / const.SETTINGS_FILENAME
|
||||
get().yaml_write(filepath)
|
||||
|
||||
@@ -12,38 +12,44 @@ from playlist.data import base
|
||||
from playlist.data import const
|
||||
|
||||
|
||||
YAMLSub = typing.NewType("YAMLSub", "YAMLBase")
|
||||
YAMLSub = typing.TypeVar("YAMLSub")
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class YAMLBase(base.BaseData):
|
||||
class YAMLBase(base.BaseData[YAMLSub]):
|
||||
"""Data class base that contains functionality to read/write as YAML."""
|
||||
|
||||
@classmethod
|
||||
def yaml_read(cls: type[YAMLBase], filepath: pathlib.Path) -> YAMLSub:
|
||||
def yaml_read(
|
||||
cls: type[YAMLBase[YAMLSub]],
|
||||
filepath: pathlib.Path,
|
||||
) -> YAMLSub:
|
||||
"""Read the given YAML file and convert it into an object."""
|
||||
with filepath.open() as fp:
|
||||
data = yaml.safe_load(fp)
|
||||
return typing.cast(YAMLSub, cls.load(data))
|
||||
|
||||
def yaml_write(self: YAMLBase, filepath: pathlib.Path) -> None:
|
||||
def yaml_write(self: YAMLBase[YAMLSub], filepath: pathlib.Path) -> None:
|
||||
"""Write this object as the given YAML file."""
|
||||
data: type(self).Dict = self.dump() # type: ignore [valid-type]
|
||||
with filepath.open(mode="w") as fp:
|
||||
fp.write(yaml.dump(data))
|
||||
|
||||
@classmethod
|
||||
def yaml_create(cls: type[YAMLBase], filepath: pathlib.Path) -> YAMLSub:
|
||||
"""Reload the YAML file with this object."""
|
||||
def yaml_create(
|
||||
cls: type[YAMLBase[YAMLSub]],
|
||||
filepath: pathlib.Path,
|
||||
) -> YAMLSub:
|
||||
"""Create the YAML file with this object."""
|
||||
data = cls()
|
||||
data.yaml_write(filepath)
|
||||
return typing.cast(YAMLSub, data)
|
||||
|
||||
@functools.cache
|
||||
@classmethod
|
||||
def get(cls: type[YAMLBase], filename: str) -> YAMLSub:
|
||||
def get(cls: type[YAMLBase[YAMLSub]], filename: str) -> YAMLSub:
|
||||
"""Get the data from the given YAML file."""
|
||||
instance: YAMLBase
|
||||
instance: YAMLSub
|
||||
filepath = const.PATHS.CONFIG / filename
|
||||
try:
|
||||
instance = cls.yaml_read(filepath)
|
||||
@@ -54,7 +60,7 @@ class YAMLBase(base.BaseData):
|
||||
return instance
|
||||
|
||||
@classmethod
|
||||
def write(cls: type[YAMLBase], filename: str) -> None:
|
||||
def write(cls: type[YAMLBase[YAMLSub]], filename: str) -> None:
|
||||
"""Store the current data to the YAML file."""
|
||||
filepath = const.PATHS.CONFIG / filename
|
||||
cls.get(filepath).yaml_write(filepath)
|
||||
cls.get(filepath).yaml_write(filepath) # type: ignore [attr-defined]
|
||||
|
||||
Reference in New Issue
Block a user