DarkHelm.org/plex-playlist-old
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Got the Typeguard thing to finally stop complaining.

Browse Source 
Signed-off-by: Cliff Hill <xlorep@darkhelm.org>
This commit is contained in:
Xlorep DarkHelm
2021-10-05 16:57:39 -04:00
parent ed93d324a8
commit 85c16b67a4
6 changed files with 40 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
src/playlist/data/base.py
View File

@@ -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))

1
src/playlist/data/models.py
View File

@@ -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(

3
src/playlist/data/settings.py
View File

@@ -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

27
src/playlist/plex/server.py
View File

@@ -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,
)

59
tests/data/test_base.py
View File

@@ -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 <data class>.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 <data class>.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 <data class>.load() works."""
result: Dummy = Dummy.load(dummydict)
assert result == dummyobj

21
tests/plex/test_server.py
View File

@@ -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