diff --git a/src/playlist/data/base.py b/src/playlist/data/base.py index 3a70bce..b0a31fa 100644 --- a/src/playlist/data/base.py +++ b/src/playlist/data/base.py @@ -2,7 +2,6 @@ from __future__ import annotations import dataclasses -import datetime import functools import typing @@ -11,7 +10,6 @@ import marshmallow DataSub = typing.TypeVar("DataSub", bound="BaseData") -DataDict = typing.NewType("DataDict", dict[str, object]) class DataMeta(type): @@ -29,30 +27,6 @@ class DataMeta(type): """The marshmallow Schema object for this data class.""" return desert.schema(cls) - @property # type: ignore [no-redef,misc] - @functools.cache - def Dict(cls: DataMeta) -> DataDict: # noqa: N802 - """The TypedDict definition for this class. - - This provides a type definition for data to be converted to/from the - data class in dictionary form. - - Returns: - dict[str, object]: The TypedDict for this object. - """ # noqa: DAR203 - fields = { - name: str if type_ is datetime.datetime else type_ - for name, type_ in cls.__annotations__.items() - if name != "id" - } - return typing.cast( - DataDict, - typing.TypedDict( # type: ignore [operator] - f"{cls.__name__}Dict", - fields, - ), - ) - @dataclasses.dataclass class BaseData(metaclass=DataMeta): @@ -61,11 +35,11 @@ class BaseData(metaclass=DataMeta): __sa_dataclass_metadata_key__ = "sa" @classmethod - def load(cls: type[BaseData], data: DataDict) -> DataSub: + def load(cls: type[BaseData], data: dict[str, object]) -> 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) -> dict[str, object]: """Dump the class instance into a data dictionary.""" cls = type(self) - return typing.cast(DataDict, cls.Schema.dump(self)) + return typing.cast(dict[str, object], cls.Schema.dump(self)) diff --git a/src/playlist/data/models.py b/src/playlist/data/models.py index 5952f89..e100260 100644 --- a/src/playlist/data/models.py +++ b/src/playlist/data/models.py @@ -19,6 +19,7 @@ class Track(base.BaseData): id: int = dataclasses.field( init=False, + repr=False, metadata={"sa": sqlalchemy.Column(sqlalchemy.Integer, primary_key=True)}, ) plex_id: int = dataclasses.field( diff --git a/src/playlist/data/settings.py b/src/playlist/data/settings.py index 9253134..b7bfaa0 100644 --- a/src/playlist/data/settings.py +++ b/src/playlist/data/settings.py @@ -29,8 +29,7 @@ class CredentialSettings(base.BaseData): def create(cls: type[CredentialSettings]) -> CredentialSettings: # pragma: no cover from playlist.plex import server - data = server.get_creds() - return typing.cast(CredentialSettings, cls.load(data)) + return server.get_creds() @dataclasses.dataclass diff --git a/src/playlist/plex/server.py b/src/playlist/plex/server.py index 5dc142f..12b7792 100644 --- a/src/playlist/plex/server.py +++ b/src/playlist/plex/server.py @@ -78,8 +78,7 @@ async def gen_tracks( weight = len(batch) / batch_size times.append(process_time * weight) weights.append(weight) - for track_data in batch: - track: models.Track = models.Track.load(track_data) + for track in batch: durations.append(track.duration) yield track @@ -107,11 +106,11 @@ async def total_track_count() -> int: return num_tracks -def get_creds() -> settings.CredentialSettings.Dict: # type: ignore [name-defined] +def get_creds() -> settings.CredentialSettings: """Get the credentials to store in the Settings instance. Returns: - settings.CredentialSettings.Dict: The dictionary containing the baseurl and token string + settings.CredentialSettings: The dictionary containing the baseurl and token string to connect with. """ username = input("Plex Username: ") @@ -120,14 +119,14 @@ def get_creds() -> settings.CredentialSettings.Dict: # type: ignore [name-defin account = plexapi.myplex.MyPlexAccount(username, password) plex = account.resource(server).connect() - return settings.CredentialSettings.Dict(baseurl=plex._baseurl, token=plex._token) + return settings.CredentialSettings(baseurl=plex._baseurl, token=plex._token) async def _downloader( pos: int, size: int, loop: asyncio.AbstractEventLoop, -) -> tuple[float, list[models.Track.Dict]]: # type: ignore [name-defined] +) -> tuple[float, list[models.Track]]: """Download a batch of tracks from Plex, by running it inside a Process Pool. Args: @@ -136,7 +135,7 @@ async def _downloader( loop: The asyncio event loop to use. Returns: - tuple[float, list[models.Track.Dict]]: The time (in seconds) to process, and + tuple[float, list[models.Track]: The time (in seconds) to process, and the batch of Tracks that were retrieved, in dictionary form. """ async with throttle: @@ -155,7 +154,7 @@ async def _downloader( def _get_track_batch( size: int, pos: int, -) -> list[models.Track.Dict]: # type: ignore [name-defined] +) -> list[models.Track]: """Get a batch of Tracks from the Plex Server. Args: @@ -163,7 +162,7 @@ def _get_track_batch( pos: The index number of the batch to retrieve. Returns: - list[models.Track.Dict]: The process time and list of Tracks retrieved, + list[models.Track]: The process time and list of Tracks retrieved, in dictionary form. """ server = plexapi.server.PlexServer(**settings.get().creds.dump()) @@ -175,16 +174,16 @@ def _get_track_batch( return [_track_dump(track) for track in batch] -def _track_dump(track: plexapi.audio.Track) -> models.Track.Dict: # type: ignore [name-defined] +def _track_dump(track: plexapi.audio.Track) -> models.Track: """Convert a PlexAPI Audio Track object into a Track in dictionary form. Args: track: The PlexAPI Audio Track to convert from. Returns: - models.Track.Dict: The dictionary form for the Track object converted to. + models.Track: The dictionary form for the Track object converted to. """ - return models.Track.Dict( + return models.Track( plex_id=track.ratingKey, track_num=track.index, title=track.title, @@ -195,9 +194,9 @@ def _track_dump(track: plexapi.audio.Track) -> models.Track.Dict: # type: ignor duration=track.duration, rating=track.userRating, comments=track.summary, - added=str(track.addedAt), + added=track.addedAt, play_count=track.viewCount, - played=str(track.lastViewedAt) if track.lastViewedAt is not None else None, + played=track.lastViewedAt, ) diff --git a/tests/data/test_base.py b/tests/data/test_base.py index 7dd5fba..eb49319 100644 --- a/tests/data/test_base.py +++ b/tests/data/test_base.py @@ -3,7 +3,6 @@ import dataclasses import datetime import typing -import marshmallow import pytest from playlist.data import base @@ -21,62 +20,30 @@ class Dummy(base.BaseData): @pytest.fixture -def dummyannotations() -> dict[str, typing.Any]: - """Make a dummy annotation dict for testing.""" - return { - "name": str, - "some_id": int, - "calc": float, - "flag": bool, - "modified": str, - } - - -@pytest.fixture -def modified_date() -> datetime.datetime: - """Make a reusable datetime for testing.""" - return datetime.datetime.now() - - -@pytest.fixture -def dummydict(modified_date: datetime.datetime) -> dict[str, typing.Any]: - """Make a dummy dictionary for testing.""" - return { - "name": "Something", - "some_id": 1, - "calc": 0.1, - "flag": True, - "modified": "T".join(str(modified_date).split(" ")), - } - - -@pytest.fixture -def dummyobj(modified_date: datetime.datetime) -> Dummy: +def dummyobj() -> Dummy: """Make a dummy object for testing.""" return Dummy( name="Something", some_id=1, calc=0.1, flag=True, - modified=modified_date, + modified=datetime.datetime.now(), ) -def test_schema() -> None: - """Validate that .Schema works.""" - result = isinstance(Dummy.Schema, marshmallow.Schema) - - assert result +@pytest.fixture +def dummydict(dummyobj: Dummy) -> dict[str, object]: + """Make a dummy dictionary for testing.""" + return { + "name": dummyobj.name, + "some_id": dummyobj.some_id, + "calc": dummyobj.calc, + "flag": dummyobj.flag, + "modified": "T".join(str(dummyobj.modified).split(" ")), + } -def test_dict(dummyannotations: dict[str, typing.Any]) -> None: - """Validate that .Dict works.""" - result = Dummy.Dict.__annotations__ == dummyannotations - - assert result - - -def test_load(dummydict: base.DataDict, dummyobj: Dummy) -> None: +def test_load(dummydict: dict[str, object], dummyobj: Dummy) -> None: """Validate that .load() works.""" result: Dummy = Dummy.load(dummydict) assert result == dummyobj diff --git a/tests/plex/test_server.py b/tests/plex/test_server.py index cd55c39..85a85a4 100644 --- a/tests/plex/test_server.py +++ b/tests/plex/test_server.py @@ -88,7 +88,7 @@ def test_get_creds(mocker): # type: ignore [no-untyped-def] mock_plex._baseurl = "Not a valid URL" mock_plex._token = "Fake token" - expected = settings.CredentialSettings.Dict( + expected = settings.CredentialSettings( baseurl=mock_plex._baseurl, token=mock_plex._token, ) @@ -120,9 +120,9 @@ async def test_downloader(mocker): # type: ignore [no-untyped-def] @pytest.fixture -def fake_trackdict(fake_plextrack): # type: ignore [no-untyped-def] +def fake_track(fake_plextrack): # type: ignore [no-untyped-def] """Make a fake track dict object.""" - return models.Track.Dict( + return models.Track( plex_id=fake_plextrack.ratingKey, track_num=fake_plextrack.index, title=fake_plextrack.title, @@ -133,22 +133,20 @@ def fake_trackdict(fake_plextrack): # type: ignore [no-untyped-def] duration=fake_plextrack.duration, rating=fake_plextrack.userRating, comments=fake_plextrack.summary, - added=str(fake_plextrack.addedAt), + added=fake_plextrack.addedAt, play_count=fake_plextrack.viewCount, - played=str(fake_plextrack.lastViewedAt) - if fake_plextrack.lastViewedAt is not None - else None, + played=fake_plextrack.lastViewedAt, ) -def test_get_track_batch(mocker, fake_trackdict): # type: ignore [no-untyped-def] +def test_get_track_batch(mocker, fake_track): # type: ignore [no-untyped-def] """Test _get_track_batch function.""" mock_plexapi_server = mocker.patch("playlist.plex.server.plexapi.server") mock_server = mock_plexapi_server.PlexServer.return_value mock_search_tracks = mock_server.library.section.return_value.searchTracks mock_search_tracks.return_value = ["Not a track."] mock_track_dump = mocker.patch("playlist.plex.server._track_dump") - mock_track_dump.return_value = fake_trackdict + mock_track_dump.return_value = fake_track result = server._get_track_batch(1, 1) @@ -178,12 +176,11 @@ def fake_plextrack(): # type: ignore [no-untyped-def] return mock_track -def test_track_dump(fake_plextrack, fake_trackdict): # type: ignore [no-untyped-def] +def test_track_dump(fake_plextrack, fake_track): # type: ignore [no-untyped-def] """Test the _track_dump function.""" - print(models.Track.Dict.__annotations__) result = server._track_dump(fake_plextrack) - assert result == fake_trackdict + assert result == fake_track @pytest.mark.asyncio