Add publications

src/infra/publication.py

@@ -0,0 +1,114 @@
+from datetime import timedelta
+import logging
+from sqlalchemy import String, Text, ForeignKey, Engine, Interval, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column, relationship, Session
+
+from src.infra.db import Base
+from src.services.publications import Publication, Order, Entry, Sequence, PublicationProfile, PublicationRepo
+
+
+class PublicationEntryModel(Base):
+    __tablename__ = 'publication_entry'
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    text: Mapped[str] = mapped_column(Text())
+    html: Mapped[str | None] = mapped_column(Text())
+
+    sequences: Mapped[list['PublicationSequenceModel']] = relationship(back_populates='entry')
+
+
+class PublicationSequenceModel(Base):
+    __tablename__ = 'publication_position'
+    id: Mapped[int] = mapped_column(primary_key=True)
+    position: Mapped[int] = mapped_column()
+    entry_id: Mapped[int] = mapped_column(ForeignKey('publication_entry.id'))
+    order_id: Mapped[int] = mapped_column(ForeignKey('publication_order.id'))
+    duration: Mapped[timedelta | None] = mapped_column(Interval())
+
+    __table_args__ = (
+        UniqueConstraint('order_id', 'entry_id'),
+        UniqueConstraint('order_id', 'position'),
+    )
+    order: Mapped['PublicationOrderModel'] = relationship(back_populates='sequence_entries')
+    entry: Mapped['PublicationEntryModel'] = relationship(back_populates='sequences')
+
+
+class PublicationOrderModel(Base):
+    __tablename__ = 'publication_order'
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    title: Mapped[str] = mapped_column(String(256))
+    sequence_entries: Mapped[list['PublicationSequenceModel']] = relationship(back_populates='order')
+
+    publication_id: Mapped[int] = mapped_column(ForeignKey('publication.id'))
+    publication: Mapped['PublicationModel'] = relationship(back_populates='orders')
+
+
+class PublicationModel(Base):
+    __tablename__ = 'publication'
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    title: Mapped[str] = mapped_column(String(256))
+    description: Mapped[str] = mapped_column(Text())
+    by_line: Mapped[str] = mapped_column(String(256))
+
+    orders: Mapped[list['PublicationOrderModel']] = relationship(back_populates='publication')
+
+
+def model_to_entry(db_pub_entry: PublicationEntryModel) -> Entry:
+    return Entry(id=db_pub_entry.id, text=db_pub_entry.text, html=db_pub_entry.html)
+
+
+def model_to_sequence(db_pub_position: PublicationSequenceModel) -> Sequence:
+    return Sequence(
+        position=db_pub_position.position,
+        entry=model_to_entry(db_pub_position.entry),
+        duration=db_pub_position.duration,
+    )
+
+
+def model_to_order(db_pub_order: PublicationOrderModel) -> Order:
+    return Order(
+        id=db_pub_order.id,
+        title=db_pub_order.title,
+        sequence_entries=list(map(model_to_sequence, db_pub_order.sequence_entries)),
+    )
+
+
+def model_to_publication(db_pub: PublicationModel) -> Publication:
+    return Publication(
+        id=db_pub.id,
+        title=db_pub.title,
+        description=db_pub.description,
+        by_line=db_pub.by_line,
+        orders=list(map(model_to_order, db_pub.orders)),
+    )
+
+
+# Useful because getting the full publication is going to require a lot of roundtrips to the database
+def model_to_publication_profile(db_pub: PublicationModel) -> PublicationProfile:
+    return PublicationProfile(id=db_pub.id, title=db_pub.title, description=db_pub.description)
+
+
+class PublicationRepoImpl(PublicationRepo):
+    def __init__(self, db: Engine):
+        self._db = db
+        self._logger = logging.getLogger('PublicationRepoImpl')
+
+    def get_publication_by_id(self, pub_id) -> Publication | None:
+        with Session(self._db) as session:
+            publication = session.get(PublicationModel, pub_id)
+
+        if publication:
+            return model_to_publication(publication)
+
+    def get_publication_profile_by_id(self, pub_id: int) -> PublicationProfile | None:
+        with Session(self._db) as session:
+            publication = session.get(PublicationModel, pub_id)
+
+        if publication:
+            return model_to_publication_profile(publication)
+
+    def get_all_publications(self) -> list[PublicationProfile]:
+        with Session(self._db) as session:
+            return list(map(model_to_publication_profile, session.query(PublicationModel).all()))

src/infra/users.py

@@ -2,7 +2,7 @@ import logging
 from argon2.exceptions import VerifyMismatchError
 from src.utils.secret import SecretBox
 from src.config.email import EmailAddress
-from sqlalchemy import String, Boolean, VARCHAR, Engine, select
+from sqlalchemy import String, VARCHAR, Engine, select
 from sqlalchemy.orm import mapped_column, Mapped, Session
 from argon2 import PasswordHasher
 from src.infra.db import Base
@@ -13,7 +13,7 @@ class UserModel(Base):
     __tablename__ = 'user'
     id: Mapped[int] = mapped_column(primary_key=True)
     email: Mapped[str] = mapped_column(String(254), unique=True)
-    email_confirmed: Mapped[bool] = mapped_column(Boolean())
+    email_confirmed: Mapped[bool] = mapped_column()
     password_hash: Mapped[str] = mapped_column(VARCHAR(255))
 
 

src/services/publications/__init__.py

@@ -0,0 +1,5 @@
+from .data import Publication, PublicationProfile, Order, Sequence, Entry
+from .service import PublicationService
+from .repo import PublicationRepo
+
+__all__ = ['Publication', 'PublicationProfile', 'Order', 'Sequence', 'Entry', 'PublicationService', 'PublicationRepo']

src/services/publications/data.py

@@ -0,0 +1,43 @@
+from datetime import timedelta
+from dataclasses import dataclass, field
+
+
+@dataclass
+class Entry:
+    id: int
+    text: str
+    html: str | None = None
+
+
+@dataclass
+class Sequence:
+    position: int
+    entry: Entry
+    duration: timedelta | None
+
+
+@dataclass
+class Order:
+    id: int
+    title: str
+    sequence_entries: list[Sequence]
+
+
+@dataclass
+class PublicationProfile:
+    id: int
+    title: str
+    description: str
+
+
+@dataclass
+class Publication:
+    id: int
+    title: str
+    description: str
+    by_line: str
+    orders: list[Order] = field(default_factory=list)
+
+    # Rather than using a subclass I think this makes sense
+    def to_profile(self) -> PublicationProfile:
+        return PublicationProfile(id=self.id, title=self.title, description=self.description)

src/services/publications/repo.py

@@ -0,0 +1,17 @@
+import abc
+
+from .data import Publication, PublicationProfile
+
+
+class PublicationRepo(abc.ABC):
+    @abc.abstractmethod
+    def get_publication_by_id(self, pub_id: int) -> Publication | None:
+        pass
+
+    @abc.abstractmethod
+    def get_publication_profile_by_id(self, pub_id: int) -> PublicationProfile | None:
+        pass
+
+    @abc.abstractmethod
+    def get_all_publications(self) -> list[PublicationProfile]:
+        pass

src/services/publications/service.py

@@ -0,0 +1,18 @@
+from src.services.publications.data import Publication, PublicationProfile
+from src.services.publications.repo import PublicationRepo
+
+
+class PublicationService:
+    """This is a bit of a shell since there's no real behavior other than lookup at this point"""
+
+    def __init__(self, repo: PublicationRepo):
+        self._repo = repo
+
+    def get_publication(self, publication_id: int) -> PublicationProfile | None:
+        return self._repo.get_publication_profile_by_id(publication_id)
+
+    def get_full_publication(self, publication_id: int) -> Publication | None:
+        return self._repo.get_publication_by_id(publication_id)
+
+    def get_all_publications(self) -> list[PublicationProfile]:
+        return self._repo.get_all_publications()