diff options
Diffstat (limited to '')
| -rw-r--r-- | src/hircine/db/models.py | 379 |
1 files changed, 379 insertions, 0 deletions
diff --git a/src/hircine/db/models.py b/src/hircine/db/models.py new file mode 100644 index 0000000..575771b --- /dev/null +++ b/src/hircine/db/models.py @@ -0,0 +1,379 @@ +import os +from datetime import date, datetime, timezone +from typing import List, Optional + +from sqlalchemy import ( + DateTime, + ForeignKey, + MetaData, + TypeDecorator, + event, + func, + select, +) +from sqlalchemy.orm import ( + DeclarativeBase, + Mapped, + declared_attr, + deferred, + joinedload, + mapped_column, + relationship, + selectinload, +) + +from hircine.api import APIException +from hircine.api.responses import InvalidParameterError +from hircine.enums import Category, Censorship, Direction, Language, Layout, Rating + +naming_convention = { + "ix": "ix_%(column_0_label)s", + "ck": "ck_%(table_name)s_%(constraint_name)s", +} + + +class DateTimeUTC(TypeDecorator): + impl = DateTime + cache_ok = True + + def process_bind_param(self, value, dialect): + if value is not None: + if not value.tzinfo: + raise TypeError("tzinfo is required") + value = value.astimezone(timezone.utc).replace(tzinfo=None) + return value + + def process_result_value(self, value, dialect): + if value is not None: + value = value.replace(tzinfo=timezone.utc) + return value + + +class Base(DeclarativeBase): + metadata = MetaData(naming_convention=naming_convention) + + @declared_attr.directive + def __tablename__(cls) -> str: + return cls.__name__.lower() + + __mapper_args__ = {"eager_defaults": True} + + @classmethod + def load_update(cls, fields): + return [] + + +class MixinID: + id: Mapped[int] = mapped_column(primary_key=True) + + +class MixinName: + name: Mapped[str] = mapped_column(unique=True) + + @classmethod + def default_order(cls): + return [cls.name] + + +class MixinFavourite: + favourite: Mapped[bool] = mapped_column(insert_default=False) + + +class MixinOrganized: + organized: Mapped[bool] = mapped_column(insert_default=False) + + +class MixinBookmarked: + bookmarked: Mapped[bool] = mapped_column(insert_default=False) + + +class MixinCreatedAt: + created_at: Mapped[datetime] = mapped_column(DateTimeUTC, server_default=func.now()) + + +class MixinModifyDates(MixinCreatedAt): + updated_at: Mapped[datetime] = mapped_column(DateTimeUTC, server_default=func.now()) + + +class Archive(MixinID, MixinCreatedAt, MixinOrganized, Base): + hash: Mapped[str] = mapped_column(unique=True) + path: Mapped[str] = mapped_column(unique=True) + size: Mapped[int] + mtime: Mapped[datetime] = mapped_column(DateTimeUTC) + + cover_id: Mapped[int] = mapped_column(ForeignKey("image.id")) + cover: Mapped["Image"] = relationship(lazy="joined", innerjoin=True) + + pages: Mapped[List["Page"]] = relationship( + back_populates="archive", + order_by="(Page.index)", + cascade="save-update, merge, expunge, delete, delete-orphan", + ) + comics: Mapped[List["Comic"]] = relationship( + back_populates="archive", + cascade="save-update, merge, expunge, delete, delete-orphan", + ) + + page_count: Mapped[int] + + @property + def name(self): + return os.path.basename(self.path) + + @classmethod + def default_order(cls): + return [cls.path] + + @classmethod + def load_full(cls): + return [ + joinedload(cls.pages, innerjoin=True), + selectinload(cls.comics), + ] + + +class Image(MixinID, Base): + hash: Mapped[str] = mapped_column(unique=True) + width: Mapped[int] + height: Mapped[int] + + @property + def aspect_ratio(self): + return self.width / self.height + + +class Page(MixinID, Base): + path: Mapped[str] + index: Mapped[int] + + archive_id: Mapped[int] = mapped_column(ForeignKey("archive.id")) + archive: Mapped["Archive"] = relationship(back_populates="pages") + + image_id: Mapped[int] = mapped_column(ForeignKey("image.id")) + image: Mapped["Image"] = relationship(lazy="joined", innerjoin=True) + + comic_id: Mapped[Optional[int]] = mapped_column(ForeignKey("comic.id")) + + +class Comic( + MixinID, MixinModifyDates, MixinFavourite, MixinOrganized, MixinBookmarked, Base +): + title: Mapped[str] + original_title: Mapped[Optional[str]] + url: Mapped[Optional[str]] + language: Mapped[Optional[Language]] + date: Mapped[Optional[date]] + + direction: Mapped[Direction] = mapped_column(insert_default=Direction.LEFT_TO_RIGHT) + layout: Mapped[Layout] = mapped_column(insert_default=Layout.SINGLE) + rating: Mapped[Optional[Rating]] + category: Mapped[Optional[Category]] + censorship: Mapped[Optional[Censorship]] + + cover_id: Mapped[int] = mapped_column(ForeignKey("image.id")) + cover: Mapped["Image"] = relationship(lazy="joined", innerjoin=True) + + archive_id: Mapped[int] = mapped_column(ForeignKey("archive.id")) + archive: Mapped["Archive"] = relationship(back_populates="comics") + + pages: Mapped[List["Page"]] = relationship(order_by="(Page.index)") + page_count: Mapped[int] + + tags: Mapped[List["ComicTag"]] = relationship( + lazy="selectin", + cascade="save-update, merge, expunge, delete, delete-orphan", + passive_deletes=True, + ) + + artists: Mapped[List["Artist"]] = relationship( + secondary="comicartist", + lazy="selectin", + order_by="(Artist.name, Artist.id)", + passive_deletes=True, + ) + + characters: Mapped[List["Character"]] = relationship( + secondary="comiccharacter", + lazy="selectin", + order_by="(Character.name, Character.id)", + passive_deletes=True, + ) + + circles: Mapped[List["Circle"]] = relationship( + secondary="comiccircle", + lazy="selectin", + order_by="(Circle.name, Circle.id)", + passive_deletes=True, + ) + + worlds: Mapped[List["World"]] = relationship( + secondary="comicworld", + lazy="selectin", + order_by="(World.name, World.id)", + passive_deletes=True, + ) + + @classmethod + def default_order(cls): + return [cls.title] + + @classmethod + def load_full(cls): + return [ + joinedload(cls.archive, innerjoin=True), + joinedload(cls.pages, innerjoin=True), + ] + + @classmethod + def load_update(cls, fields): + if "pages" in fields: + return [joinedload(cls.pages, innerjoin=True)] + return [] + + +class Tag(MixinID, MixinModifyDates, MixinName, Base): + description: Mapped[Optional[str]] + namespaces: Mapped[List["Namespace"]] = relationship( + secondary="tagnamespaces", + passive_deletes=True, + order_by="(Namespace.sort_name, Namespace.name, Namespace.id)", + ) + + @classmethod + def load_full(cls): + return [selectinload(cls.namespaces)] + + @classmethod + def load_update(cls, fields): + if "namespaces" in fields: + return cls.load_full() + return [] + + +class Namespace(MixinID, MixinModifyDates, MixinName, Base): + sort_name: Mapped[Optional[str]] + + @classmethod + def default_order(cls): + return [cls.sort_name, cls.name] + + @classmethod + def load_full(cls): + return [] + + +class TagNamespaces(Base): + namespace_id: Mapped[int] = mapped_column( + ForeignKey("namespace.id", ondelete="CASCADE"), primary_key=True + ) + tag_id: Mapped[int] = mapped_column( + ForeignKey("tag.id", ondelete="CASCADE"), primary_key=True + ) + + +class ComicTag(Base): + comic_id: Mapped[int] = mapped_column( + ForeignKey("comic.id", ondelete="CASCADE"), primary_key=True + ) + namespace_id: Mapped[int] = mapped_column( + ForeignKey("namespace.id", ondelete="CASCADE"), primary_key=True + ) + tag_id: Mapped[int] = mapped_column( + ForeignKey("tag.id", ondelete="CASCADE"), primary_key=True + ) + + namespace: Mapped["Namespace"] = relationship( + lazy="joined", + innerjoin=True, + order_by="(Namespace.sort_name, Namespace.name, Namespace.id)", + ) + + tag: Mapped["Tag"] = relationship( + lazy="joined", + innerjoin=True, + order_by="(Tag.name, Tag.id)", + ) + + @property + def name(self): + return f"{self.namespace.name}:{self.tag.name}" + + @property + def id(self): + return f"{self.namespace.id}:{self.tag.id}" + + +class Artist(MixinID, MixinModifyDates, MixinName, Base): + pass + + +class ComicArtist(Base): + comic_id: Mapped[int] = mapped_column( + ForeignKey("comic.id", ondelete="CASCADE"), primary_key=True + ) + artist_id: Mapped[int] = mapped_column( + ForeignKey("artist.id", ondelete="CASCADE"), primary_key=True + ) + + +class Character(MixinID, MixinModifyDates, MixinName, Base): + pass + + +class ComicCharacter(Base): + comic_id: Mapped[int] = mapped_column( + ForeignKey("comic.id", ondelete="CASCADE"), primary_key=True + ) + character_id: Mapped[int] = mapped_column( + ForeignKey("character.id", ondelete="CASCADE"), primary_key=True + ) + + +class Circle(MixinID, MixinModifyDates, MixinName, Base): + pass + + +class ComicCircle(Base): + comic_id: Mapped[int] = mapped_column( + ForeignKey("comic.id", ondelete="CASCADE"), primary_key=True + ) + circle_id: Mapped[int] = mapped_column( + ForeignKey("circle.id", ondelete="CASCADE"), primary_key=True + ) + + +class World(MixinID, MixinModifyDates, MixinName, Base): + pass + + +class ComicWorld(Base): + comic_id: Mapped[int] = mapped_column( + ForeignKey("comic.id", ondelete="CASCADE"), primary_key=True + ) + world_id: Mapped[int] = mapped_column( + ForeignKey("world.id", ondelete="CASCADE"), primary_key=True + ) + + +def defer_relationship_count(relationship, secondary=False): + left, right = relationship.property.synchronize_pairs[0] + + return deferred( + select(func.count(right)) + .select_from(right.table) + .where(left == right) + .scalar_subquery() + ) + + +Comic.tag_count = defer_relationship_count(Comic.tags) + + +@event.listens_for(Comic.pages, "bulk_replace") +def on_comic_pages_bulk_replace(target, values, initiator): + if not values: + raise APIException( + InvalidParameterError(parameter="pages", text="cannot be empty") + ) + + target.page_count = len(values) |