Itmens/lib.itmens
Itmens/lib.itmens
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 3

more typehints

Browse source 
wip

wip
This commit is contained in:
Your Name 2024-05-27 15:44:12 -04:00 • committed by Henri Dickson
parent bc79c957de
commit a2ffff1760
29 changed files with 247 additions and 88 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
catalog/book/models.py
View file

@ -18,6 +18,7 @@ work data seems asymmetric (a book links to a work, but may not listed in that w
"""
from os.path import exists
from typing import TYPE_CHECKING
from django.core.validators import MaxValueValidator, MinValueValidator
from django.db import models
@ -62,6 +63,8 @@ class EditionSchema(EditionInSchema, BaseSchema):
class Edition(Item):
if TYPE_CHECKING:
works: "models.ManyToManyField[Work, Edition]"
category = ItemCategory.Book
url_path = "book"
@ -164,17 +167,17 @@ class Edition(Item):
return detect_isbn_asin(lookup_id_value)
return super().lookup_id_cleanup(lookup_id_type, lookup_id_value)
def merge_to(self, to_item: "Edition | None"):
def merge_to(self, to_item: "Edition | None"): # type: ignore[reportIncompatibleMethodOverride]
super().merge_to(to_item)
if to_item:
for work in self.works.all():
to_item.works.add(work)
self.works.clear()
def delete(self, using=None, soft=True, *args, **kwargs):
def delete(self, using=None, keep_parents=False, soft=True, *args, **kwargs):
if soft:
self.works.clear()
return super().delete(using, soft, *args, **kwargs)
return super().delete(using, soft, keep_parents, *args, **kwargs)
def update_linked_items_from_external_resource(self, resource):
"""add Work from resource.metadata['work'] if not yet"""
@ -279,7 +282,7 @@ class Work(Item):
]
return [(i.value, i.label) for i in id_types]
def merge_to(self, to_item: "Work | None"):
def merge_to(self, to_item: "Work | None"): # type: ignore[reportIncompatibleMethodOverride]
super().merge_to(to_item)
if to_item:
for edition in self.editions.all():
@ -293,10 +296,10 @@ class Work(Item):
to_item.other_title += [self.title] # type: ignore
to_item.save()
def delete(self, using=None, soft=True, *args, **kwargs):
def delete(self, using=None, keep_parents=False, soft=True, *args, **kwargs):
if soft:
self.editions.clear()
return super().delete(using, soft, *args, **kwargs)
return super().delete(using, keep_parents, soft, *args, **kwargs)
class Series(Item):

8
catalog/collection/models.py
View file

@ -1,7 +1,13 @@
from catalog.common import *
from typing import TYPE_CHECKING
from catalog.common import Item, ItemCategory
class Collection(Item):
if TYPE_CHECKING:
from journal.models import Collection as JournalCollection
journal_item: "JournalCollection"
category = ItemCategory.Collection
@property

5
catalog/common/mixins.py
View file

@ -13,10 +13,11 @@ class SoftDeleteMixin:
def clear(self):
pass
def delete(self, using=None, soft=True, *args, **kwargs):
def delete(self, using=None, keep_parents=False, soft=True, *args, **kwargs):
if soft:
self.clear()
self.is_deleted = True
self.save(using=using) # type: ignore
return 0, {}
else:
return super().delete(using=using, *args, **kwargs) # type: ignore
return super().delete(using=using, keep_parents=keep_parents, *args, **kwargs) # type: ignore

107
catalog/common/models.py
View file

@ -1,7 +1,7 @@
import re
import uuid
from functools import cached_property
from typing import TYPE_CHECKING, Any, Iterable, Type, cast
from typing import TYPE_CHECKING, Any, Iterable, Self, Type, cast
from auditlog.context import disable_auditlog
from auditlog.models import AuditlogHistoryField, LogEntry
@ -9,7 +9,8 @@ from django.conf import settings
from django.contrib.contenttypes.models import ContentType
from django.core.files.uploadedfile import SimpleUploadedFile
from django.db import connection, models
from django.db.models import QuerySet
from django.db.models import QuerySet, Value
from django.template.defaultfilters import default
from django.utils import timezone
from django.utils.baseconv import base62
from django.utils.translation import gettext_lazy as _
@ -19,12 +20,14 @@ from polymorphic.models import PolymorphicModel
from catalog.common import jsondata
from .mixins import SoftDeleteMixin
from .utils import DEFAULT_ITEM_COVER, item_cover_path, resource_cover_path
if TYPE_CHECKING:
from journal.models import Collection
from users.models import User
from .sites import ResourceContent
class SiteName(models.TextChoices):
Unknown = "unknown", _("Unknown")
@ -76,9 +79,9 @@ class IdType(models.TextChoices):
Bandcamp = "bandcamp", _("Bandcamp")
Spotify_Album = "spotify_album", _("Spotify Album")
Spotify_Show = "spotify_show", _("Spotify Podcast")
Discogs_Release = "discogs_release", ("Discogs Release")
Discogs_Master = "discogs_master", ("Discogs Master")
MusicBrainz = "musicbrainz", ("MusicBrainz ID")
Discogs_Release = "discogs_release", _("Discogs Release")
Discogs_Master = "discogs_master", _("Discogs Master")
MusicBrainz = "musicbrainz", _("MusicBrainz ID")
# DoubanBook_Author = "doubanbook_author", _("Douban Book Author")
# DoubanCelebrity = "doubanmovie_celebrity", _("Douban Movie Celebrity")
# Goodreads_Author = "goodreads_author", _("Goodreads Author")
@ -168,14 +171,16 @@ class PrimaryLookupIdDescriptor(object): # TODO make it mixin of Field
def __init__(self, id_type: IdType):
self.id_type = id_type
def __get__(self, instance, cls=None):
def __get__(
self, instance: "Item | None", cls: type[Any] | None = None
) -> str | Self | None:
if instance is None:
return self
if self.id_type != instance.primary_lookup_id_type:
return None
return instance.primary_lookup_id_value
def __set__(self, instance, id_value):
def __set__(self, instance: "Item", id_value: str | None):
if id_value:
instance.primary_lookup_id_type = self.id_type
instance.primary_lookup_id_value = id_value
@ -246,12 +251,16 @@ class ItemSchema(BaseSchema, ItemInSchema):
pass
class Item(PolymorphicModel, SoftDeleteMixin):
class Item(PolymorphicModel):
if TYPE_CHECKING:
external_resources: QuerySet["ExternalResource"]
collections: QuerySet["Collection"]
merged_from_items: QuerySet["Item"]
merged_to_item_id: int
category: ItemCategory # subclass must specify this
url_path = "item" # subclass must specify this
type = None # subclass must specify this
child_class = None # subclass may specify this to allow link to parent item
parent_class = None # subclass may specify this to allow create child item
category: ItemCategory # subclass must specify this
uid = models.UUIDField(default=uuid.uuid4, editable=False, db_index=True)
title = models.CharField(_("title"), max_length=1000, default="")
brief = models.TextField(_("description"), blank=True, default="")
@ -288,6 +297,24 @@ class Item(PolymorphicModel, SoftDeleteMixin):
]
]
def delete(
self,
using: Any = None,
keep_parents: bool = False,
soft: bool = True,
*args: tuple[Any, ...],
**kwargs: dict[str, Any],
) -> tuple[int, dict[str, int]]:
if soft:
self.clear()
self.is_deleted = True
self.save(using=using)
return 0, {}
else:
return super().delete(
using=using, keep_parents=keep_parents, *args, **kwargs
)
@cached_property
def history(self):
# can't use AuditlogHistoryField bc it will only return history with current content type
@ -324,7 +351,7 @@ class Item(PolymorphicModel, SoftDeleteMixin):
return lookup_id_type, lookup_id_value.strip()
@classmethod
def get_best_lookup_id(cls, lookup_ids: dict[IdType, str]) -> tuple[IdType, str]:
def get_best_lookup_id(cls, lookup_ids: dict[str, str]) -> tuple[str, str]:
"""get best available lookup id, ideally commonly used"""
for t in IdealIdTypes:
if lookup_ids.get(t):
@ -406,7 +433,7 @@ class Item(PolymorphicModel, SoftDeleteMixin):
res.item = to_item
res.save()
def recast_to(self, model: "type[Item]") -> "Item":
def recast_to(self, model: "type[Any]") -> "Item":
logger.warning(f"recast item {self} to {model}")
if isinstance(self, model):
return self
@ -453,7 +480,7 @@ class Item(PolymorphicModel, SoftDeleteMixin):
return self.title
@classmethod
def get_by_url(cls, url_or_b62: str) -> "Item | None":
def get_by_url(cls, url_or_b62: str) -> "Self | None":
b62 = url_or_b62.strip().split("/")[-1]
if len(b62) not in [21, 22]:
r = re.search(r"[A-Za-z0-9]{21,22}", url_or_b62)
@ -469,7 +496,7 @@ class Item(PolymorphicModel, SoftDeleteMixin):
# prefix = id_type.strip().lower() + ':'
# return next((x[len(prefix):] for x in self.lookup_ids if x.startswith(prefix)), None)
def update_lookup_ids(self, lookup_ids):
def update_lookup_ids(self, lookup_ids: list[tuple[str, str]]):
for t, v in lookup_ids:
if t in IdealIdTypes and self.primary_lookup_id_type not in IdealIdTypes:
self.primary_lookup_id_type = t
@ -484,25 +511,25 @@ class Item(PolymorphicModel, SoftDeleteMixin):
] # list of metadata keys to copy from resource to item
@classmethod
def copy_metadata(cls, metadata):
def copy_metadata(cls, metadata: dict[str, Any]) -> dict[str, Any]:
return dict(
(k, v)
for k, v in metadata.items()
if k in cls.METADATA_COPY_LIST and v is not None
)
def has_cover(self):
return self.cover and self.cover != DEFAULT_ITEM_COVER
def has_cover(self) -> bool:
return bool(self.cover) and self.cover != DEFAULT_ITEM_COVER
@property
def cover_image_url(self):
def cover_image_url(self) -> str | None:
return (
f"{settings.SITE_INFO['site_url']}{self.cover.url}"
f"{settings.SITE_INFO['site_url']}{self.cover.url}" # type:ignore
if self.cover and self.cover != DEFAULT_ITEM_COVER
else None
)
def merge_data_from_external_resources(self, ignore_existing_content=False):
def merge_data_from_external_resources(self, ignore_existing_content: bool = False):
"""Subclass may override this"""
lookup_ids = []
for p in self.external_resources.all():
@ -517,7 +544,7 @@ class Item(PolymorphicModel, SoftDeleteMixin):
self.cover = p.cover
self.update_lookup_ids(list(set(lookup_ids)))
def update_linked_items_from_external_resource(self, resource):
def update_linked_items_from_external_resource(self, resource: "ExternalResource"):
"""Subclass should override this"""
pass
@ -575,6 +602,9 @@ class ItemLookupId(models.Model):
class ExternalResource(models.Model):
if TYPE_CHECKING:
required_resources: list[dict[str, str]]
related_resources: list[dict[str, str]]
item = models.ForeignKey(
Item, null=True, on_delete=models.SET_NULL, related_name="external_resources"
)
@ -598,15 +628,21 @@ class ExternalResource(models.Model):
scraped_time = models.DateTimeField(null=True)
created_time = models.DateTimeField(auto_now_add=True)
edited_time = models.DateTimeField(auto_now=True)
required_resources = jsondata.ArrayField(
models.CharField(), null=False, blank=False, default=list
) # links required to generate Item from this resource, e.g. parent TVShow of TVSeason
) # type: ignore
""" links required to generate Item from this resource, e.g. parent TVShow of TVSeason """
related_resources = jsondata.ArrayField(
models.CharField(), null=False, blank=False, default=list
) # links related to this resource which may be fetched later, e.g. sub TVSeason of TVShow
) # type: ignore
"""links related to this resource which may be fetched later, e.g. sub TVSeason of TVShow"""
prematched_resources = jsondata.ArrayField(
models.CharField(), null=False, blank=False, default=list
) # links to help match an existing Item from this resource
)
"""links to help match an existing Item from this resource"""
class Meta:
unique_together = [["id_type", "id_value"]]
@ -645,7 +681,7 @@ class ExternalResource(models.Model):
return n or domain
return self.site_name.label
def update_content(self, resource_content):
def update_content(self, resource_content: "ResourceContent"):
self.other_lookup_ids = resource_content.lookup_ids
self.metadata = resource_content.metadata
if resource_content.cover_image and resource_content.cover_image_extention:
@ -662,13 +698,15 @@ class ExternalResource(models.Model):
def ready(self):
return bool(self.metadata and self.scraped_time)
def get_all_lookup_ids(self):
def get_all_lookup_ids(self) -> dict[str, str]:
d = self.other_lookup_ids.copy()
d[self.id_type] = self.id_value
d = {k: v for k, v in d.items() if bool(v)}
return d
def get_lookup_ids(self, default_model):
def get_lookup_ids(
self, default_model: type[Item] | None = None
) -> list[tuple[str, str]]:
lookup_ids = self.get_all_lookup_ids()
model = self.get_item_model(default_model)
bt, bv = model.get_best_lookup_id(lookup_ids)
@ -677,23 +715,30 @@ class ExternalResource(models.Model):
ids = [(bt, bv)] + ids
return ids
def get_item_model(self, default_model: type[Item]) -> type[Item]:
def get_item_model(self, default_model: type[Item] | None) -> type[Item]:
model = self.metadata.get("preferred_model")
if model:
m = ContentType.objects.filter(
app_label="catalog", model=model.lower()
).first()
if m:
return cast(Item, m).model_class()
mc: type[Item] | None = m.model_class() # type: ignore
if not mc:
raise ValueError(
f"preferred model {model} does not exist in ContentType"
)
return mc
else:
raise ValueError(f"preferred model {model} does not exist")
if not default_model:
raise ValueError("no default preferred model specified")
return default_model
_CONTENT_TYPE_LIST = None
def item_content_types():
def item_content_types() -> dict[type[Item], int]:
global _CONTENT_TYPE_LIST
if _CONTENT_TYPE_LIST is None:
_CONTENT_TYPE_LIST = {}

1
catalog/jobs/discover.py
View file

@ -1,5 +1,6 @@
from datetime import timedelta
from django.conf import settings
from django.core.cache import cache
from django.db.models import Count, F
from django.utils import timezone

9
catalog/management/commands/catalog.py
View file

@ -1,5 +1,6 @@
import pprint
from django.contrib.contenttypes.models import ContentType
from django.core.management.base import BaseCommand
from django.db.models import Count, F
@ -111,12 +112,14 @@ class Command(BaseCommand):
else:
self.stdout.write(f"! no season {i} : {i.absolute_url}?skipcheck=1")
if self.fix:
i.recast_to(i.merged_to_item.__class__)
i.recast_to(i.merged_to_item.__class__) # type:ignore
self.stdout.write(f"Checking TVSeason is child of other class...")
for i in TVSeason.objects.filter(show__isnull=False).exclude(
show__polymorphic_ctype_id=tvshow_ct_id
):
if not i.show:
continue
self.stdout.write(f"! {i.show} : {i.show.absolute_url}?skipcheck=1")
if self.fix:
i.show = None
@ -124,6 +127,8 @@ class Command(BaseCommand):
self.stdout.write(f"Checking deleted item with child TV Season...")
for i in TVSeason.objects.filter(show__is_deleted=True):
if not i.show:
continue
self.stdout.write(f"! {i.show} : {i.show.absolute_url}?skipcheck=1")
if self.fix:
i.show.is_deleted = False
@ -131,6 +136,8 @@ class Command(BaseCommand):
self.stdout.write(f"Checking merged item with child TV Season...")
for i in TVSeason.objects.filter(show__merged_to_item__isnull=False):
if not i.show:
continue
self.stdout.write(f"! {i.show} : {i.show.absolute_url}?skipcheck=1")
if self.fix:
i.show = i.show.merged_to_item

49
catalog/models.py
View file

@ -98,3 +98,52 @@ def init_catalog_audit_log():
)
# logger.debug(f"Catalog audit log initialized for {item_content_types().values()}")
__all__ = [
"Edition",
"EditionInSchema",
"EditionSchema",
"Series",
"Work",
"CatalogCollection",
"AvailableItemCategory",
"ExternalResource",
"IdType",
"Item",
"ItemCategory",
"ItemInSchema",
"ItemSchema",
"ItemType",
"SiteName",
"item_categories",
"item_content_types",
"Game",
"GameInSchema",
"GameSchema",
"Movie",
"MovieInSchema",
"MovieSchema",
"Album",
"AlbumInSchema",
"AlbumSchema",
"Performance",
"PerformanceProduction",
"PerformanceProductionSchema",
"PerformanceSchema",
"Podcast",
"PodcastEpisode",
"PodcastInSchema",
"PodcastSchema",
"TVEpisode",
"TVEpisodeSchema",
"TVSeason",
"TVSeasonInSchema",
"TVSeasonSchema",
"TVShow",
"TVShowInSchema",
"TVShowSchema",
"Indexer",
"init_catalog_search_models",
"init_catalog_audit_log",
]

19
catalog/performance/models.py
View file

@ -1,4 +1,5 @@
from functools import cached_property
from typing import TYPE_CHECKING
from django.db import models
from django.utils.translation import gettext_lazy as _
@ -100,6 +101,8 @@ def _crew_by_role(crew):
class Performance(Item):
if TYPE_CHECKING:
productions: models.QuerySet["PerformanceProduction"]
type = ItemType.Performance
child_class = "PerformanceProduction"
category = ItemCategory.Performance
@ -371,10 +374,10 @@ class PerformanceProduction(Item):
]
@property
def parent_item(self):
def parent_item(self) -> Performance | None: # type:ignore
return self.show
def set_parent_item(self, value):
def set_parent_item(self, value: Performance | None): # type:ignore
self.show = value
@classmethod
@ -389,23 +392,23 @@ class PerformanceProduction(Item):
return f"{self.show.title if self.show else '♢'} {self.title}"
@property
def cover_image_url(self):
def cover_image_url(self) -> str | None:
return (
self.cover.url
self.cover.url # type:ignore
if self.cover and self.cover != DEFAULT_ITEM_COVER
else self.show.cover_image_url
if self.show
else None
)
def update_linked_items_from_external_resource(self, resource):
def update_linked_items_from_external_resource(self, resource: ExternalResource):
for r in resource.required_resources:
if r["model"] == "Performance":
resource = ExternalResource.objects.filter(
res = ExternalResource.objects.filter(
id_type=r["id_type"], id_value=r["id_value"]
).first()
if resource and resource.item:
self.show = resource.item
if res and res.item:
self.show = res.item
@cached_property
def crew_by_role(self):

10
catalog/podcast/models.py
View file

@ -1,3 +1,5 @@
from typing import TYPE_CHECKING
from django.db import models
from django.utils.translation import gettext_lazy as _
@ -26,6 +28,8 @@ class PodcastSchema(PodcastInSchema, BaseSchema):
class Podcast(Item):
if TYPE_CHECKING:
episodes: models.QuerySet["PodcastEpisode"]
category = ItemCategory.Podcast
child_class = "PodcastEpisode"
url_path = "podcast"
@ -107,10 +111,10 @@ class PodcastEpisode(Item):
]
@property
def parent_item(self):
def parent_item(self) -> Podcast | None: # type:ignore
return self.program
def set_parent_item(self, value):
def set_parent_item(self, value: Podcast | None): # type:ignore
self.program = value
@property
@ -123,7 +127,7 @@ class PodcastEpisode(Item):
self.program.cover_image_url if self.program else None
)
def get_url_with_position(self, position=None):
def get_url_with_position(self, position: int | str | None = None):
return (
self.url
if position is None or position == ""

1
catalog/sites/fedi.py
View file

@ -1,5 +1,6 @@
import re
from django.conf import settings
from django.core.validators import URLValidator
from loguru import logger

1
catalog/sites/rss.py
View file

@ -5,6 +5,7 @@ from datetime import datetime
import bleach
import podcastparser
from django.conf import settings
from django.core.cache import cache
from django.core.exceptions import ValidationError
from django.core.validators import URLValidator

18
catalog/tv/models.py
View file

@ -26,9 +26,13 @@ For now, we follow Douban convention, but keep an eye on it in case it breaks it
"""
import re
from functools import cached_property
from typing import TYPE_CHECKING, overload
from auditlog.diff import ForeignKey
from auditlog.models import QuerySet
from django.db import models
from django.utils.translation import gettext_lazy as _
from typing_extensions import override
from catalog.common import (
BaseSchema,
@ -90,6 +94,8 @@ class TVEpisodeSchema(ItemSchema):
class TVShow(Item):
if TYPE_CHECKING:
seasons: QuerySet["TVSeason"]
type = ItemType.TVShow
child_class = "TVSeason"
category = ItemCategory.TV
@ -249,6 +255,8 @@ class TVShow(Item):
class TVSeason(Item):
if TYPE_CHECKING:
episodes: models.QuerySet["TVEpisode"]
type = ItemType.TVSeason
category = ItemCategory.TV
url_path = "tv/season"
@ -424,10 +432,10 @@ class TVSeason(Item):
return self.episodes.all().order_by("episode_number")
@property
def parent_item(self):
def parent_item(self) -> TVShow | None: # type:ignore
return self.show
def set_parent_item(self, value):
def set_parent_item(self, value: TVShow | None): # type:ignore
self.show = value
@property
@ -462,10 +470,10 @@ class TVEpisode(Item):
)
@property
def parent_item(self):
def parent_item(self) -> TVSeason | None: # type:ignore
return self.season
def set_parent_item(self, value):
def set_parent_item(self, value: TVSeason | None): # type:ignore
self.season = value
@classmethod
@ -476,7 +484,7 @@ class TVEpisode(Item):
]
return [(i.value, i.label) for i in id_types]
def update_linked_items_from_external_resource(self, resource):
def update_linked_items_from_external_resource(self, resource: ExternalResource):
for w in resource.required_resources:
if w["model"] == "TVSeason":
p = ExternalResource.objects.filter(

2
catalog/views.py
View file

@ -129,7 +129,7 @@ def retrieve(request, item_path, item_uuid):
def episode_data(request, item_uuid):
item = get_object_or_404(Item, uid=get_uuid_or_404(item_uuid))
item = get_object_or_404(Podcast, uid=get_uuid_or_404(item_uuid))
qs = item.episodes.all().order_by("-pub_date")
if request.GET.get("last"):
qs = qs.filter(pub_date__lt=request.GET.get("last"))

18
catalog/views_edit.py
View file

@ -224,13 +224,13 @@ def assign_parent(request, item_path, item_uuid):
@require_http_methods(["POST"])
@login_required
def remove_unused_seasons(request, item_path, item_uuid):
item = get_object_or_404(Item, uid=get_uuid_or_404(item_uuid))
item = get_object_or_404(TVShow, uid=get_uuid_or_404(item_uuid))
sl = list(item.seasons.all())
for s in sl:
if not s.journal_exists():
s.delete()
ol = [s.id for s in sl]
nl = [s.id for s in item.seasons.all()]
ol = [s.pk for s in sl]
nl = [s.pk for s in item.seasons.all()]
discord_send(
"audit",
f"{item.absolute_url}\n{ol} ➡ {nl}\nby [@{request.user.username}]({request.user.absolute_url})",
@ -244,7 +244,7 @@ def remove_unused_seasons(request, item_path, item_uuid):
@require_http_methods(["POST"])
@login_required
def fetch_tvepisodes(request, item_path, item_uuid):
item = get_object_or_404(Item, uid=get_uuid_or_404(item_uuid))
item = get_object_or_404(TVSeason, uid=get_uuid_or_404(item_uuid))
if item.class_name != "tvseason" or not item.imdb or item.season_number is None:
raise BadRequest(_("TV Season with IMDB id and season number required."))
item.log_action({"!fetch_tvepisodes": ["", ""]})
@ -257,7 +257,7 @@ def fetch_tvepisodes(request, item_path, item_uuid):
def fetch_episodes_for_season_task(item_uuid, user):
with set_actor(user):
season = Item.get_by_url(item_uuid)
season = TVSeason.get_by_url(item_uuid)
if not season:
return
episodes = season.episode_uuids
@ -313,8 +313,8 @@ def merge(request, item_path, item_uuid):
@require_http_methods(["POST"])
@login_required
def link_edition(request, item_path, item_uuid):
item = get_object_or_404(Item, uid=get_uuid_or_404(item_uuid))
new_item = Item.get_by_url(request.POST.get("target_item_url"))
item = get_object_or_404(Edition, uid=get_uuid_or_404(item_uuid))
new_item = Edition.get_by_url(request.POST.get("target_item_url"))
if (
not new_item
or new_item.is_deleted
@ -325,7 +325,7 @@ def link_edition(request, item_path, item_uuid):
if item.class_name != "edition" or new_item.class_name != "edition":
raise BadRequest(_("Cannot link items other than editions"))
if request.POST.get("sure", 0) != "1":
new_item = Item.get_by_url(request.POST.get("target_item_url"))
new_item = Edition.get_by_url(request.POST.get("target_item_url")) # type: ignore
return render(
request,
"catalog_merge.html",
@ -345,7 +345,7 @@ def link_edition(request, item_path, item_uuid):
@require_http_methods(["POST"])
@login_required
def unlink_works(request, item_path, item_uuid):
item = get_object_or_404(Item, uid=get_uuid_or_404(item_uuid))
item = get_object_or_404(Edition, uid=get_uuid_or_404(item_uuid))
if not request.user.is_staff and item.journal_exists():
raise PermissionDenied(_("Insufficient permission"))
item.unlink_from_all_works()

2
journal/management/commands/collection.py
View file

@ -47,7 +47,7 @@ class Command(BaseCommand):
{
"title": member.item.title,
"url": member.item.absolute_url,
"note": member.note,
"note": member.note, # type:ignore
}
)
print(json.dumps(data, indent=2))

3
journal/models/collection.py
View file

@ -1,5 +1,6 @@
import re
from functools import cached_property
from typing import TYPE_CHECKING
from django.db import models
from django.utils.translation import gettext_lazy as _
@ -40,6 +41,8 @@ class CollectionMember(ListMember):
class Collection(List):
if TYPE_CHECKING:
members: models.QuerySet[CollectionMember]
url_path = "collection"
MEMBER_CLASS = CollectionMember
catalog_item = models.OneToOneField(

8
journal/models/common.py
View file

@ -21,6 +21,8 @@ from .mixins import UserOwnedObjectMixin
if TYPE_CHECKING:
from takahe.models import Post
from .like import Like
class VisibilityType(models.IntegerChoices):
Public = 0, _("Public")
@ -112,6 +114,8 @@ def q_item_in_category(item_category: ItemCategory):
class Piece(PolymorphicModel, UserOwnedObjectMixin):
if TYPE_CHECKING:
likes: models.QuerySet["Like"]
url_path = "p" # subclass must specify this
uid = models.UUIDField(default=uuid.uuid4, editable=False, db_index=True)
local = models.BooleanField(default=True)
@ -257,7 +261,7 @@ class Content(Piece):
owner = models.ForeignKey(APIdentity, on_delete=models.PROTECT)
visibility = models.PositiveSmallIntegerField(
default=0
) # 0: Public / 1: Follower only / 2: Self only
) # 0: Public / 1: Follower only / 2: Self only # type:ignore
created_time = models.DateTimeField(default=timezone.now)
edited_time = models.DateTimeField(auto_now=True)
metadata = models.JSONField(default=dict)
@ -275,7 +279,7 @@ class Debris(Content):
class_name = CharField(max_length=50)
@classmethod
def create_from_piece(cls, c: Piece):
def create_from_piece(cls, c: Content):
return cls.objects.create(
class_name=c.__class__.__name__,
owner=c.owner,

17
journal/models/itemlist.py
View file

@ -1,4 +1,5 @@
from functools import cached_property
from typing import TYPE_CHECKING
import django.dispatch
from django.db import models
@ -18,10 +19,14 @@ class List(Piece):
List (abstract model)
"""
if TYPE_CHECKING:
MEMBER_CLASS: "type[ListMember]"
members: "models.QuerySet[ListMember]"
items: "models.ManyToManyField[Item, List]"
owner = models.ForeignKey(APIdentity, on_delete=models.PROTECT)
visibility = models.PositiveSmallIntegerField(
default=0
) # 0: Public / 1: Follower only / 2: Self only
) # 0: Public / 1: Follower only / 2: Self only # type:ignore
created_time = models.DateTimeField(default=timezone.now)
edited_time = models.DateTimeField(auto_now=True)
metadata = models.JSONField(default=dict)
@ -29,7 +34,6 @@ class List(Piece):
class Meta:
abstract = True
MEMBER_CLASS: Piece
# MEMBER_CLASS = None # subclass must override this
# subclass must add this:
# items = models.ManyToManyField(Item, through='ListMember')
@ -76,9 +80,10 @@ class List(Piece):
ml = self.ordered_members
p = {"parent": self}
p.update(params)
lm = ml.last()
member = self.MEMBER_CLASS.objects.create(
owner=self.owner,
position=ml.last().position + 1 if ml.count() else 1,
position=lm.position + 1 if lm else 1,
item=item,
**p,
)
@ -96,7 +101,7 @@ class List(Piece):
def update_member_order(self, ordered_member_ids):
for m in self.members.all():
try:
i = ordered_member_ids.index(m.id)
i = ordered_member_ids.index(m.pk)
if m.position != i + 1:
m.position = i + 1
m.save()
@ -142,10 +147,12 @@ class ListMember(Piece):
parent = models.ForeignKey('List', related_name='members', on_delete=models.CASCADE)
"""
if TYPE_CHECKING:
parent: models.ForeignKey["ListMember", "List"]
owner = models.ForeignKey(APIdentity, on_delete=models.PROTECT)
visibility = models.PositiveSmallIntegerField(
default=0
) # 0: Public / 1: Follower only / 2: Self only
) # 0: Public / 1: Follower only / 2: Self only # type:ignore[reportAssignmentType]
created_time = models.DateTimeField(default=timezone.now)
edited_time = models.DateTimeField(auto_now=True)
metadata = models.JSONField(default=dict)

2
journal/models/like.py
View file

@ -12,7 +12,7 @@ class Like(Piece): # TODO remove
owner = models.ForeignKey(APIdentity, on_delete=models.PROTECT)
visibility = models.PositiveSmallIntegerField(
default=0
) # 0: Public / 1: Follower only / 2: Self only
) # 0: Public / 1: Follower only / 2: Self only # type: ignore
created_time = models.DateTimeField(default=timezone.now)
edited_time = models.DateTimeField(auto_now=True)
target = models.ForeignKey(Piece, on_delete=models.CASCADE, related_name="likes")

6
journal/models/mark.py
View file

@ -42,7 +42,7 @@ class Mark:
self.item = item
@cached_property
def shelfmember(self) -> ShelfMember:
def shelfmember(self) -> ShelfMember | None:
return self.owner.shelf_manager.locate_item(self.item)
@property
@ -198,7 +198,7 @@ class Mark:
last_visibility = self.visibility if last_shelf_type else None
if shelf_type is None: # TODO change this use case to DEFERRED status
# take item off shelf
if last_shelf_type:
if self.shelfmember:
Takahe.delete_posts(self.shelfmember.all_post_ids)
self.shelfmember.log_and_delete()
if self.comment:
@ -207,7 +207,7 @@ class Mark:
self.rating.delete()
return
# create/update shelf member and shelf log if necessary
if last_shelf_type == shelf_type:
if self.shelfmember and last_shelf_type == shelf_type:
shelfmember_changed = False
log_entry = self.shelfmember.ensure_log_entry()
if metadata is not None and metadata != self.shelfmember.metadata:

4
journal/models/mixins.py
View file

@ -18,7 +18,9 @@ class UserOwnedObjectMixin:
"""
if TYPE_CHECKING:
owner: ForeignKey[APIdentity, Piece]
owner: ForeignKey[Piece, APIdentity]
# owner: ForeignKey[APIdentity, Piece]
owner_id: int
visibility: int
def is_visible_to(

6
journal/models/shelf.py
View file

@ -272,6 +272,9 @@ _SHELF_LABELS = [
class ShelfMember(ListMember):
if TYPE_CHECKING:
parent: models.ForeignKey["ShelfMember", "Shelf"]
parent = models.ForeignKey(
"Shelf", related_name="members", on_delete=models.CASCADE
)
@ -375,6 +378,9 @@ class Shelf(List):
Shelf
"""
if TYPE_CHECKING:
members: models.QuerySet[ShelfMember]
class Meta:
unique_together = [["owner", "shelf_type"]]

3
journal/models/tag.py
View file

@ -1,5 +1,6 @@
import re
from functools import cached_property
from typing import TYPE_CHECKING
from django.core.validators import MaxValueValidator, MinValueValidator, RegexValidator
from django.db import connection, models
@ -14,6 +15,8 @@ from .itemlist import List, ListMember
class TagMember(ListMember):
if TYPE_CHECKING:
parent: models.ForeignKey["TagMember", "Tag"]
parent = models.ForeignKey("Tag", related_name="members", on_delete=models.CASCADE)
class Meta:

2
journal/views/collection.py
View file

@ -247,7 +247,7 @@ def collection_update_item_note(request: AuthedHttpRequest, collection_uuid, ite
)
return collection_retrieve_items(request, collection_uuid, True)
else:
member = collection.get_member_for_item(item)
member: CollectionMember = collection.get_member_for_item(item) # type:ignore
return render(
request,
"collection_update_item_note.html",

2
journal/views/common.py
View file

@ -99,7 +99,7 @@ def render_list(
if year:
year = int(year)
queryset = queryset.filter(created_time__year=year)
paginator = Paginator(queryset, PAGE_SIZE)
paginator = Paginator(queryset, PAGE_SIZE) # type:ignore
page_number = int(request.GET.get("page", default=1))
members = paginator.get_page(page_number)
pagination = PageLinksGenerator(page_number, paginator.num_pages, request.GET)

5
mastodon/api.py
View file

@ -291,16 +291,17 @@ def get_related_acct_list(site, token, api):
)
url = None
if response.status_code == 200:
r: list[dict[str, str]] = response.json()
results.extend(
map(
lambda u: (
lambda u: ( # type: ignore
u["acct"]
if u["acct"].find("@") != -1
else u["acct"] + "@" + site
)
if "acct" in u
else u,
response.json(),
r,
)
)
if "Link" in response.headers:

4
pyproject.toml
View file

@ -1,5 +1,5 @@
[tool.pyright]
exclude = [ "media", ".venv", ".git", "playground", "catalog/*/tests.py", "neodb", "**/migrations", "**/sites/douban_*", "neodb-takahe" ]
exclude = [ "media", ".venv", ".git", "playground", "catalog/*/tests.py", "journal/tests.py", "neodb", "**/migrations", "**/sites/douban_*", "neodb-takahe" ]
reportIncompatibleVariableOverride = false
reportUnusedImport = false
reportUnknownVariableType = false
@ -10,6 +10,8 @@ reportUnknownArgumentType = false
reportAny = false
reportImplicitOverride = false
reportUninitializedInstanceVariable = false
reportMissingTypeStubs = false
reportIgnoreCommentWithoutRule = false
[tool.djlint]
ignore="T002,T003,H005,H006,H019,H020,H021,H023,H030,H031,D018"

1
takahe/models.py
View file

@ -869,6 +869,7 @@ class Post(models.Model):
A post (status, toot) that is either local or remote.
"""
if TYPE_CHECKING:
interactions: "models.QuerySet[PostInteraction]"
attachments: "models.QuerySet[PostAttachment]"

3
takahe/utils.py
View file

@ -533,7 +533,8 @@ class Takahe:
@staticmethod
def post_collection(collection: "Collection"):
existing_post = collection.latest_post
user = collection.owner.user
owner: APIdentity = collection.owner
user = owner.user
if not user:
raise ValueError(f"Cannot find user for collection {collection}")
visibility = Takahe.visibility_n2t(