models/entities: Use custom Set type for better typing

server/entertainment_decider/common.py

@@ -1,161 +0,0 @@
-from datetime import date, datetime
-import itertools
-import subprocess
-import sys
-from typing import (
-    IO,
-    Iterable,
-    Iterator,
-    List,
-    Literal,
-    Optional,
-    Sequence,
-    Tuple,
-    TypeVar,
-    Union,
-    overload,
-)
-
-
-def all_same(iterable: Iterable) -> bool:
-    it = iter(iterable)
-    first = next(it)
-    return all(first == elem for elem in it)
-
-
-def call(
-    args: Sequence[str],
-    check: bool = True,
-    stdin: Optional[IO] = None,
-) -> subprocess.CompletedProcess:
-    proc = subprocess.run(
-        args,
-        capture_output=True,
-        check=check,
-        text=True,
-        stdin=stdin,
-    )
-    return proc
-
-
-# source: https://peps.python.org/pep-0257/#handling-docstring-indentation
-def trim(docstring: str) -> str:
-    """
-    Trim strings like docstrings are trimmed following PEP 257
-    """
-    if not docstring:
-        return ""
-    # Convert tabs to spaces (following the normal Python rules)
-    # and split into a list of lines:
-    lines = docstring.expandtabs().splitlines()
-    # Determine minimum indentation (first line doesn't count):
-    indent = sys.maxsize
-    for line in lines[1:]:
-        stripped = line.lstrip()
-        if stripped:
-            indent = min(indent, len(line) - len(stripped))
-    # Remove indentation (first line is special):
-    trimmed = [lines[0].strip()]
-    if indent < sys.maxsize:
-        for line in lines[1:]:
-            trimmed.append(line[indent:].rstrip())
-    # Strip off trailing and leading blank lines:
-    while trimmed and not trimmed[-1]:
-        trimmed.pop()
-    while trimmed and not trimmed[0]:
-        trimmed.pop(0)
-    # Return a single string:
-    return "\n".join(trimmed)
-
-
-def update_bool_value(
-    old_value: bool, new_value: Union[bool, Literal["toggle"]]
-) -> bool:
-    if new_value == "toggle":
-        return not old_value
-    if type(new_value) != bool:
-        raise Exception(
-            f'Invalid type of new_value: Expected bool or literal "toggle", got type={type(new_value)!r}, value={new_value!r}'
-        )
-    return new_value
-
-
-T = TypeVar("T")
-
-
-def limit_iter(iter: Iterable[T], limit: int) -> List[T]:
-    return list(itertools.islice(iter, limit))
-
-
-class _IterFixer(Iterator[T]):
-    __it: Iterator[T]
-
-    def __init__(self, it: Iterator[T]) -> None:
-        super().__init__()
-        self.__it = it
-
-    def __iter__(self) -> Iterator[T]:
-        return self
-
-    def __next__(self) -> T:
-        return next(self.__it)
-
-
-def fix_iter(iterable: Iterable[T]) -> Iterable[T]:
-    return _IterFixer(iter(iterable))
-
-
-@overload
-def iter_lookahead(
-    iterable: Iterable[T],
-    get_first: Literal[False] = False,
-    get_last: Literal[False] = False,
-) -> Iterable[Tuple[T, T]]:
-    ...
-
-
-@overload
-def iter_lookahead(
-    iterable: Iterable[T],
-    get_first: Literal[True],
-    get_last: Literal[False] = False,
-) -> Iterable[Tuple[None, T] | Tuple[T, T]]:
-    ...
-
-
-@overload
-def iter_lookahead(
-    iterable: Iterable[T],
-    get_first: Literal[False] = False,
-    get_last: Literal[True] = True,  #  <- default only to satisfy python
-) -> Iterable[Tuple[T, T] | Tuple[T, None]]:
-    ...
-
-
-@overload
-def iter_lookahead(
-    iterable: Iterable[T],
-    get_first: Literal[True],
-    get_last: Literal[True],
-) -> Iterable[Tuple[None, T] | Tuple[T, T] | Tuple[T, None]]:
-    ...
-
-
-def iter_lookahead(
-    iterable: Iterable[T],
-    get_first: bool = False,
-    get_last: bool = False,
-) -> Iterable[Tuple[None, T] | Tuple[T, T] | Tuple[T, None]]:
-    it = iter(iterable)
-    last = next(it)
-    if get_first:
-        yield None, last
-    for cur in it:
-        yield last, cur
-        last = cur
-    if get_last:
-        yield last, None
-
-
-def date_to_datetime(d: date) -> datetime:
-    return datetime(d.year, d.month, d.day)

server/entertainment_decider/common/__init__.py

@@ -0,0 +1,36 @@
+from ._converter import (
+    date_to_datetime,
+    to_just_number,
+)
+from ._itertools import (
+    all_same,
+    fix_iter,
+    iter_lookahead,
+    limit_iter,
+)
+from ._search import (
+    search_source_by_keys,
+)
+from ._setting_handler import (
+    update_bool_value,
+)
+from ._string import (
+    trim,
+)
+from ._subprocess import (
+    call,
+)
+
+
+__all__ = [
+    "all_same",
+    "call",
+    "date_to_datetime",
+    "fix_iter",
+    "iter_lookahead",
+    "limit_iter",
+    "search_source_by_keys",
+    "to_just_number",
+    "trim",
+    "update_bool_value",
+]

server/entertainment_decider/common/_converter.py

@@ -0,0 +1,15 @@
+from datetime import date, datetime
+from typing import (
+    Optional,
+)
+
+
+def date_to_datetime(d: date) -> datetime:
+    return datetime(d.year, d.month, d.day)
+
+
+def to_just_number(name: str) -> Optional[int]:
+    try:
+        return int(name)
+    except ValueError:
+        return None

server/entertainment_decider/common/_itertools.py

@@ -0,0 +1,190 @@
+import itertools
+from typing import (
+    Iterable,
+    Iterator,
+    List,
+    Literal,
+    Tuple,
+    TypeVar,
+    overload,
+)
+
+
+T = TypeVar("T")
+
+
+# check Iterable for same values
+
+
+def all_same(iterable: Iterable) -> bool:
+    """
+    Returns True if all items of Iterable are equal, False otherwise.
+    """
+    it = iter(iterable)
+    first = next(it)
+    return all(first == elem for elem in it)
+
+
+# fix Iterables
+
+
+class _IterFixer(Iterator[T]):
+    __it: Iterator[T]
+
+    def __init__(self, it: Iterator[T]) -> None:
+        super().__init__()
+        self.__it = it
+
+    def __iter__(self) -> Iterator[T]:
+        # this method is the reason for this class
+        # some iterators miss it and so are not Iterable without this proxy
+        return self
+
+    def __next__(self) -> T:
+        return next(self.__it)
+
+
+def fix_iter(iterable: Iterable[T]) -> Iterable[T]:
+    """
+    wraps the iterator of iterable
+    into a simple proxy Iterator/Iterable implementation.
+
+    Required in weird edgecases,
+    where the iterator of an iterable is not an iterable.
+    """
+    return _IterFixer(iter(iterable))
+
+
+# lookahead on Iterables
+
+
+@overload
+def iter_lookahead(
+    iterable: Iterable[T],
+    get_first: Literal[False] = False,
+    get_last: Literal[False] = False,
+) -> Iterable[Tuple[T, T]]:
+    """
+    allows iterating over <iterable> while also getting the last/next item
+
+    The pairs are given as a Tuple and so can be split:
+    ```python
+    for last, cur in iter_lookahead(iterable): ...
+    ```
+
+    The first and last tuple will contain the first / last two entries, so that
+    - `iterable[0]` only appears in `last` and not in `cur`
+    - `iterable[-1]` only appears in `cur` and not in `last`
+
+    You can change that behavior with <get_first>/<get_last>.
+    """
+    ...
+
+
+@overload
+def iter_lookahead(
+    iterable: Iterable[T],
+    get_first: Literal[True],
+    get_last: Literal[False] = False,
+) -> Iterable[Tuple[None, T] | Tuple[T, T]]:
+    """
+    allows iterating over <iterable> while also getting the last/next item
+
+    The pairs are given as a Tuple and so can be split:
+    ```python
+    for last, cur in iter_lookahead(iterable): ...
+    ```
+
+    - The first tuple will look like `(None, cur)`,
+      so that `iterable[0]` will appear in both `cur` and `last`.
+    - The last tuple will contain the last two entries,
+      so that `iterable[-1]` only appears in `cur` and not in `last`.
+
+    You can change that behavior with <get_first>/<get_last>.
+    """
+    ...
+
+
+@overload
+def iter_lookahead(
+    iterable: Iterable[T],
+    get_first: Literal[False] = False,
+    get_last: Literal[True] = True,  #  <- default only to satisfy python
+) -> Iterable[Tuple[T, T] | Tuple[T, None]]:
+    """
+    allows iterating over <iterable> while also getting the last/next item
+
+    The pairs are given as a Tuple and so can be split:
+    ```python
+    for last, cur in iter_lookahead(iterable): ...
+    ```
+
+    - The first tuple will contain the first two entries,
+      so that `iterable[0]` only appears in `cur` and not in `last`.
+    - The last tuple will look like `(last, None)`,
+      so that `iterable[-1]` will appear in both `cur` and `last`.
+
+    You can change that behavior with <get_first>/<get_last>.
+    """
+    ...
+
+
+@overload
+def iter_lookahead(
+    iterable: Iterable[T],
+    get_first: Literal[True],
+    get_last: Literal[True],
+) -> Iterable[Tuple[None, T] | Tuple[T, T] | Tuple[T, None]]:
+    """
+    allows iterating over <iterable> while also getting the last/next item
+
+    The pairs are given as a Tuple and so can be split:
+    ```python
+    for last, cur in iter_lookahead(iterable): ...
+    ```
+
+    The first / last tuple will look like `(None, T)` / `(T, None)`,
+    so that both `iterable[0]` and `iterable[-1]`
+    will both appear in `cur` and `last`.
+
+    You can change that behavior with <get_first>/<get_last>.
+    """
+    ...
+
+
+def iter_lookahead(
+    iterable: Iterable[T],
+    get_first: bool = False,
+    get_last: bool = False,
+) -> Iterable[Tuple[None, T] | Tuple[T, T] | Tuple[T, None]]:
+    """
+    allows iterating over <iterable> while also getting the last/next item
+
+    The pairs are given as a Tuple and so can be split:
+    ```python
+    for last, cur in iter_lookahead(iterable): ...
+    ```
+
+    The behavior for the border cases can be configured.
+    If <get_first> is True, the first tuple will be `(None, last)`.
+    If <get_last> is True, the last tuple will be `(cur, None)`.
+    Otherwise, both values will be set at the borders respectively.
+    See the docstrings for the overloaded methods.
+    """
+    it = iter(iterable)
+    last = next(it)
+    if get_first:
+        yield None, last
+    for cur in it:
+        yield last, cur
+        last = cur
+    if get_last:
+        yield last, None
+
+
+# limit Iterables
+
+
+def limit_iter(iter: Iterable[T], limit: int) -> List[T]:
+    """gets the first <limit> elements of <iter> and puts them in a list"""
+    return list(itertools.islice(iter, limit))

server/entertainment_decider/common/_search.py

@@ -0,0 +1,24 @@
+from typing import (
+    Callable,
+    Optional,
+    Sequence,
+    TypeVar,
+)
+
+
+T = TypeVar("T")
+K = TypeVar("K")
+
+
+def search_source_by_keys(
+    source: Callable[[K], Optional[T]],
+    all_keys: Sequence[K],
+) -> Optional[T]:
+    """
+    tries <all_keys> in <source> and returns first result `!= None`, otherwise `None`
+    """
+    for key in all_keys:
+        item = source(key)
+        if item is not None:
+            return item
+    return None

server/entertainment_decider/common/_setting_handler.py

@@ -0,0 +1,16 @@
+from typing import (
+    Literal,
+)
+
+
+def update_bool_value(
+    old_value: bool,
+    new_value: bool | Literal["toggle"],
+) -> bool:
+    if new_value == "toggle":
+        return not old_value
+    if type(new_value) != bool:
+        raise Exception(
+            f'Invalid type of new_value: Expected bool or literal "toggle", got type={type(new_value)!r}, value={new_value!r}'
+        )
+    return new_value

server/entertainment_decider/common/_string.py

@@ -0,0 +1,31 @@
+import sys
+
+
+# source: https://peps.python.org/pep-0257/#handling-docstring-indentation
+def trim(docstring: str) -> str:
+    """
+    Trim strings like docstrings are trimmed following PEP 257
+    """
+    if not docstring:
+        return ""
+    # Convert tabs to spaces (following the normal Python rules)
+    # and split into a list of lines:
+    lines = docstring.expandtabs().splitlines()
+    # Determine minimum indentation (first line doesn't count):
+    indent = sys.maxsize
+    for line in lines[1:]:
+        stripped = line.lstrip()
+        if stripped:
+            indent = min(indent, len(line) - len(stripped))
+    # Remove indentation (first line is special):
+    trimmed = [lines[0].strip()]
+    if indent < sys.maxsize:
+        for line in lines[1:]:
+            trimmed.append(line[indent:].rstrip())
+    # Strip off trailing and leading blank lines:
+    while trimmed and not trimmed[-1]:
+        trimmed.pop()
+    while trimmed and not trimmed[0]:
+        trimmed.pop(0)
+    # Return a single string:
+    return "\n".join(trimmed)

server/entertainment_decider/common/_subprocess.py

@@ -0,0 +1,21 @@
+import subprocess
+from typing import (
+    IO,
+    Optional,
+    Sequence,
+)
+
+
+def call(
+    args: Sequence[str],
+    check: bool = True,
+    stdin: Optional[IO] = None,
+) -> subprocess.CompletedProcess:
+    proc = subprocess.run(
+        args,
+        capture_output=True,
+        check=check,
+        text=True,
+        stdin=stdin,
+    )
+    return proc

server/entertainment_decider/models/entities.py

@@ -184,13 +184,13 @@ class Tag(db.Entity, Tagable, TagProto["Tag"]):
 
     use_for_preferences: bool = orm.Required(bool, default=True)
 
-    tag_keys: Iterable[TagKey] = orm.Set(lambda: TagKey)
+    tag_keys: Set[TagKey] = orm.Set(lambda: TagKey)
 
-    super_tag_list: Iterable[Tag] = orm.Set(lambda: Tag, reverse="sub_tag_list")
-    sub_tag_list: Iterable[Tag] = orm.Set(lambda: Tag, reverse="super_tag_list")
+    super_tag_list: Set[Tag] = orm.Set(lambda: Tag, reverse="sub_tag_list")
+    sub_tag_list: Set[Tag] = orm.Set(lambda: Tag, reverse="super_tag_list")
 
-    collection_list: Iterable[MediaCollection] = orm.Set(lambda: MediaCollection)
-    media_list: Iterable[MediaElement] = orm.Set(lambda: MediaElement)
+    collection_list: Set[MediaCollection] = orm.Set(lambda: MediaCollection)
+    media_list: Set[MediaElement] = orm.Set(lambda: MediaElement)
 
     @property
     def orm_assigned_tags(self) -> Query[Tag]:
@@ -378,13 +378,13 @@ class MediaElement(db.Entity, UriHolder, Tagable):
     )
     length: int = orm.Optional(int)
 
-    tag_list: Iterable[Tag] = orm.Set(
+    tag_list: Set[Tag] = orm.Set(
         lambda: Tag,
     )
-    __uri_list: Iterable[MediaUriMapping] = orm.Set(
+    __uri_list: Set[MediaUriMapping] = orm.Set(
         lambda: MediaUriMapping,
     )
-    collection_links: Iterable[MediaCollectionLink] = orm.Set(
+    collection_links: Set[MediaCollectionLink] = orm.Set(
         lambda: MediaCollectionLink,
     )
 
@@ -731,13 +731,13 @@ class MediaCollection(db.Entity, UriHolder, Tagable):
         index=True,
     )
 
-    tag_list: Iterable[Tag] = orm.Set(
+    tag_list: Set[Tag] = orm.Set(
         lambda: Tag,
     )
-    __uri_set: Iterable[CollectionUriMapping] = orm.Set(
+    __uri_set: Set[CollectionUriMapping] = orm.Set(
         lambda: CollectionUriMapping,
     )
-    media_links: Iterable[MediaCollectionLink] = orm.Set(
+    media_links: Set[MediaCollectionLink] = orm.Set(
         MediaCollectionLink,
     )
     created_collections: Set[MediaCollection] = orm.Set(