Replace sqlalchemy backref parameter with back_populates (#2194)

The sqlalchemy backref parameter is considered legacy see: https://docs.sqlalchemy.org/en/14/orm/backref.html. Using this construct means that mypy (and IDE) can't infer the type of the dynamically generated parameter.

src/palace/manager/feed/acquisition.py

@@ -212,7 +212,7 @@ def _create_entry(
         cls,
         work: Work,
         active_licensepool: LicensePool | None,
-        edition: Edition,
+        edition: Edition | None,
         identifier: Identifier,
         annotator: Annotator,
     ) -> WorkEntry:
@@ -926,10 +926,11 @@ def single_entry(cls, work: tuple[Identifier, Work], annotator: Annotator) -> Wo
         if error_status:
             return cls.error_message(identifier, error_status, error_message or "")
 
-        if active_licensepool:
-            edition = active_licensepool.presentation_edition
-        else:
-            edition = _work.presentation_edition
+        edition = (
+            active_licensepool.presentation_edition
+            if active_licensepool
+            else _work.presentation_edition
+        )
         try:
             return cls._create_entry(
                 _work, active_licensepool, edition, identifier, annotator

src/palace/manager/feed/annotator/base.py

@@ -236,9 +236,7 @@ def content(cls, work: Work | None) -> str:
                 and work.summary.representation
                 and work.summary.representation.content
             ):
-                content = work.summary.representation.content
-                if isinstance(content, bytes):
-                    content = content.decode("utf-8")
+                content = work.summary.representation.content.decode("utf-8")
                 work.summary_text = content
                 summary = work.summary_text
         return summary

src/palace/manager/feed/annotator/circulation.py

@@ -365,7 +365,7 @@ def annotate_work_entry(
         updated: datetime.datetime | None = None,
     ) -> None:
         work = entry.work
-        identifier = entry.identifier or work.presentation_edition.primary_identifier
+        identifier = entry.identifier
         active_license_pool = entry.license_pool or self.active_licensepool_for(work)
         # If OpenSearch included a more accurate last_update_time,
         # use it instead of Work.last_update_time
@@ -875,7 +875,7 @@ def annotate_work_entry(
             return
 
         work = entry.work
-        identifier = entry.identifier or work.presentation_edition.primary_identifier
+        identifier = entry.identifier
 
         permalink_uri, permalink_type = self.permalink_for(identifier)
         # TODO: Do not force OPDS types

src/palace/manager/feed/annotator/loan_and_hold.py

@@ -97,11 +97,13 @@ def annotate_work_entry(
         active_license_pool = entry.license_pool
         work = entry.work
         edition = work.presentation_edition
-        identifier = edition.primary_identifier
+        identifier = edition.primary_identifier if edition else None
         # Only OPDS for Distributors should get the time tracking link
         # And only if there is an active loan for the work
         if (
-            edition.medium == EditionConstants.AUDIO_MEDIUM
+            edition
+            and identifier
+            and edition.medium == EditionConstants.AUDIO_MEDIUM
             and active_license_pool
             and active_license_pool.should_track_playtime is True
             and work in self.active_loans_by_work

src/palace/manager/sqlalchemy/model/admin.py

@@ -40,7 +40,7 @@ class Admin(Base, HasSessionCache):
 
     # An Admin may have many roles.
     roles: Mapped[list[AdminRole]] = relationship(
-        "AdminRole", backref="admin", cascade="all, delete-orphan", uselist=True
+        "AdminRole", back_populates="admin", cascade="all, delete-orphan", uselist=True
     )
 
     # Token age is max 30 minutes, in seconds
@@ -290,6 +290,7 @@ class AdminRole(Base, HasSessionCache):
 
     id = Column(Integer, primary_key=True)
     admin_id = Column(Integer, ForeignKey("admins.id"), nullable=False, index=True)
+    admin: Mapped[Admin] = relationship("Admin", back_populates="roles")
     library_id = Column(Integer, ForeignKey("libraries.id"), nullable=True, index=True)
     library: Mapped[Library] = relationship("Library", back_populates="adminroles")
     role = Column(Unicode, nullable=False, index=True)

src/palace/manager/sqlalchemy/model/circulationevent.py

@@ -1,14 +1,20 @@
 # CirculationEvent
-
+from __future__ import annotations
 
 import logging
+from typing import TYPE_CHECKING
 
 from sqlalchemy import Column, DateTime, ForeignKey, Index, Integer, String, Unicode
+from sqlalchemy.orm import Mapped, relationship
 
 from palace.manager.sqlalchemy.model.base import Base
 from palace.manager.sqlalchemy.util import get_one_or_create
 from palace.manager.util.datetime_helpers import utc_now
 
+if TYPE_CHECKING:
+    from palace.manager.sqlalchemy.model.library import Library
+    from palace.manager.sqlalchemy.model.licensing import LicensePool
+
 
 class CirculationEvent(Base):
     """Changes to a license pool's circulation status.
@@ -25,6 +31,9 @@ class CirculationEvent(Base):
 
     # One LicensePool can have many circulation events.
     license_pool_id = Column(Integer, ForeignKey("licensepools.id"), index=True)
+    license_pool: Mapped[LicensePool | None] = relationship(
+        "LicensePool", back_populates="circulation_events"
+    )
 
     type = Column(String(32), index=True)
     start = Column(DateTime(timezone=True), index=True)
@@ -36,6 +45,9 @@ class CirculationEvent(Base):
     # The Library associated with the event, if it happened in the
     # context of a particular Library and we know which one.
     library_id = Column(Integer, ForeignKey("libraries.id"), index=True, nullable=True)
+    library: Mapped[Library | None] = relationship(
+        "Library", back_populates="circulation_events"
+    )
 
     # The geographic location associated with the event. This string
     # may mean different things for different libraries. It might be a

src/palace/manager/sqlalchemy/model/classification.py

@@ -132,6 +132,7 @@ class Subject(Base):
 
     # Each Subject may claim affinity with one Genre.
     genre_id = Column(Integer, ForeignKey("genres.id"), index=True)
+    genre: Mapped[Genre] = relationship("Genre", back_populates="subjects")
 
     # A locked Subject has been reviewed by a human and software will
     # not mess with it without permission.
@@ -346,11 +347,15 @@ class Classification(Base):
     __tablename__ = "classifications"
     id = Column(Integer, primary_key=True)
     identifier_id = Column(Integer, ForeignKey("identifiers.id"), index=True)
-    identifier: Mapped[Identifier | None]
+    identifier: Mapped[Identifier | None] = relationship(
+        "Identifier", back_populates="classifications"
+    )
     subject_id = Column(Integer, ForeignKey("subjects.id"), index=True)
     subject: Mapped[Subject] = relationship("Subject", back_populates="classifications")
     data_source_id = Column(Integer, ForeignKey("datasources.id"), index=True)
-    data_source: Mapped[DataSource | None]
+    data_source: Mapped[DataSource | None] = relationship(
+        "DataSource", back_populates="classifications"
+    )
 
     # How much weight the data source gives to this classification.
     weight = Column(Integer)
@@ -481,7 +486,7 @@ class Genre(Base, HasSessionCache):
     name = Column(Unicode, unique=True, index=True)
 
     # One Genre may have affinity with many Subjects.
-    subjects: Mapped[list[Subject]] = relationship("Subject", backref="genre")
+    subjects: Mapped[list[Subject]] = relationship("Subject", back_populates="genre")
 
     # One Genre may participate in many WorkGenre assignments.
     works = association_proxy("work_genres", "work")
@@ -490,7 +495,9 @@ class Genre(Base, HasSessionCache):
         "WorkGenre", back_populates="genre", cascade="all, delete-orphan"
     )
 
-    lane_genres: Mapped[list[LaneGenre]] = relationship("LaneGenre", backref="genre")
+    lane_genres: Mapped[list[LaneGenre]] = relationship(
+        "LaneGenre", back_populates="genre"
+    )
 
     def __repr__(self):
         if classifier.genres.get(self.name):

src/palace/manager/sqlalchemy/model/collection.py

@@ -129,13 +129,13 @@ class Collection(Base, HasSessionCache, RedisKeyMixin):
     )
 
     catalog: Mapped[list[Identifier]] = relationship(
-        "Identifier", secondary="collections_identifiers", backref="collections"
+        "Identifier", secondary="collections_identifiers", back_populates="collections"
     )
 
     # A Collection can be associated with multiple CoverageRecords
     # for Identifiers in its catalog.
     coverage_records: Mapped[list[CoverageRecord]] = relationship(
-        "CoverageRecord", backref="collection", cascade="all"
+        "CoverageRecord", back_populates="collection", cascade="all"
     )
 
     # A collection may be associated with one or more custom lists.
@@ -144,7 +144,7 @@ class Collection(Base, HasSessionCache, RedisKeyMixin):
     # the list and they won't be added back, so the list doesn't
     # necessarily match the collection.
     customlists: Mapped[list[CustomList]] = relationship(
-        "CustomList", secondary="collections_customlists", backref="collections"
+        "CustomList", secondary="collections_customlists", back_populates="collections"
     )
 
     export_marc_records = Column(Boolean, default=False, nullable=False)

src/palace/manager/sqlalchemy/model/coverage.py

@@ -347,6 +347,9 @@ class CoverageRecord(Base, BaseCoverageRecord):
     # coverage has taken place. This is currently only applicable
     # for Metadata Wrangler coverage.
     collection_id = Column(Integer, ForeignKey("collections.id"), nullable=True)
+    collection: Mapped[Collection | None] = relationship(
+        "Collection", back_populates="coverage_records"
+    )
 
     __table_args__ = (
         Index(

src/palace/manager/sqlalchemy/model/customlist.py

@@ -31,6 +31,8 @@
 
 if TYPE_CHECKING:
     from palace.manager.sqlalchemy.model.collection import Collection
+    from palace.manager.sqlalchemy.model.edition import Edition
+    from palace.manager.sqlalchemy.model.lane import Lane
     from palace.manager.sqlalchemy.model.library import Library
 
 
@@ -43,7 +45,6 @@ class CustomList(Base):
     INIT = "init"
     UPDATED = "updated"
     REPOPULATE = "repopulate"
-    auto_update_status_enum = Enum(INIT, UPDATED, REPOPULATE, name="auto_update_status")
 
     __tablename__ = "customlists"
     id = Column(Integer, primary_key=True)
@@ -59,6 +60,7 @@ class CustomList(Base):
     updated = Column(DateTime(timezone=True), index=True)
     responsible_party = Column(Unicode)
     library_id = Column(Integer, ForeignKey("libraries.id"), index=True, nullable=True)
+    library: Mapped[Library] = relationship("Library", back_populates="custom_lists")
 
     # How many titles are in this list? This is calculated and
     # cached when the list contents change.
@@ -80,11 +82,17 @@ class CustomList(Base):
     auto_update_query = Column(Unicode, nullable=True)  # holds json data
     auto_update_facets = Column(Unicode, nullable=True)  # holds json data
     auto_update_last_update = Column(DateTime, nullable=True)
-    auto_update_status: Mapped[str] = Column(auto_update_status_enum, default=INIT)  # type: ignore[assignment]
+    auto_update_status: Mapped[str] = Column(
+        Enum(INIT, UPDATED, REPOPULATE, name="auto_update_status"), default=INIT
+    )
+
+    collections: Mapped[list[Collection]] = relationship(
+        "Collection", secondary="collections_customlists", back_populates="customlists"
+    )
 
-    # Typing specific
-    collections: list[Collection]
-    library: Library
+    lane: Mapped[list[Lane]] = relationship(
+        "Lane", secondary="lanes_customlists", back_populates="customlists"
+    )
 
     __table_args__ = (
         UniqueConstraint("data_source_id", "foreign_identifier"),
@@ -369,7 +377,15 @@ class CustomListEntry(Base):
     )
 
     edition_id = Column(Integer, ForeignKey("editions.id"), index=True)
+    edition: Mapped[Edition | None] = relationship(
+        "Edition", back_populates="custom_list_entries"
+    )
+
     work_id = Column(Integer, ForeignKey("works.id"), index=True)
+    work: Mapped[Work | None] = relationship(
+        "Work", back_populates="custom_list_entries"
+    )
+
     featured = Column(Boolean, nullable=False, default=False)
     annotation = Column(Unicode)
 

src/palace/manager/sqlalchemy/model/datasource.py

@@ -64,10 +64,14 @@ class DataSource(Base, HasSessionCache, DataSourceConstants):
     )
 
     # One DataSource can provide many Hyperlinks.
-    links: Mapped[list[Hyperlink]] = relationship("Hyperlink", backref="data_source")
+    links: Mapped[list[Hyperlink]] = relationship(
+        "Hyperlink", back_populates="data_source"
+    )
 
     # One DataSource can provide many Resources.
-    resources: Mapped[list[Resource]] = relationship("Resource", backref="data_source")
+    resources: Mapped[list[Resource]] = relationship(
+        "Resource", back_populates="data_source"
+    )
 
     # One DataSource can generate many Measurements.
     measurements: Mapped[list[Measurement]] = relationship(
@@ -76,7 +80,7 @@ class DataSource(Base, HasSessionCache, DataSourceConstants):
 
     # One DataSource can provide many Classifications.
     classifications: Mapped[list[Classification]] = relationship(
-        "Classification", backref="data_source"
+        "Classification", back_populates="data_source"
     )
 
     # One DataSource can have many associated Credentials.
@@ -92,7 +96,7 @@ class DataSource(Base, HasSessionCache, DataSourceConstants):
     # One DataSource can provide many LicensePoolDeliveryMechanisms.
     delivery_mechanisms: Mapped[list[LicensePoolDeliveryMechanism]] = relationship(
         "LicensePoolDeliveryMechanism",
-        backref="data_source",
+        back_populates="data_source",
         foreign_keys="LicensePoolDeliveryMechanism.data_source_id",
     )
 

src/palace/manager/sqlalchemy/model/devicetokens.py

@@ -2,7 +2,7 @@
 
 from sqlalchemy import Column, Enum, ForeignKey, Index, Integer, Unicode
 from sqlalchemy.exc import IntegrityError
-from sqlalchemy.orm import Mapped, backref, relationship
+from sqlalchemy.orm import Mapped, relationship
 
 from palace.manager.core.exceptions import BasePalaceException
 from palace.manager.sqlalchemy.model.base import Base
@@ -32,9 +32,7 @@ class DeviceToken(Base):
         index=True,
         nullable=False,
     )
-    patron: Mapped[Patron] = relationship(
-        "Patron", backref=backref("device_tokens", passive_deletes=True)
-    )
+    patron: Mapped[Patron] = relationship("Patron", back_populates="device_tokens")
 
     token_type_enum = Enum(
         DeviceTokenTypes.FCM_ANDROID, DeviceTokenTypes.FCM_IOS, name="token_types"

src/palace/manager/sqlalchemy/model/edition.py

@@ -41,6 +41,7 @@
 
 if TYPE_CHECKING:
     from palace.manager.sqlalchemy.model.customlist import CustomListEntry
+    from palace.manager.sqlalchemy.model.resource import Resource
     from palace.manager.sqlalchemy.model.work import Work
 
 
@@ -85,17 +86,19 @@ class Edition(Base, EditionConstants):
     # it. Through the Equivalency class, it is associated with a
     # (probably huge) number of other identifiers.
     primary_identifier_id = Column(Integer, ForeignKey("identifiers.id"), index=True)
-    primary_identifier: Identifier  # for typing
+    primary_identifier: Mapped[Identifier] = relationship(
+        "Identifier", back_populates="primarily_identifies"
+    )
 
     # An Edition may be the presentation edition for a single Work. If it's not
     # a presentation edition for a work, work will be None.
     work: Mapped[Work] = relationship(
-        "Work", uselist=False, backref="presentation_edition"
+        "Work", uselist=False, back_populates="presentation_edition"
     )
 
     # An Edition may show up in many CustomListEntries.
     custom_list_entries: Mapped[list[CustomListEntry]] = relationship(
-        "CustomListEntry", backref="edition"
+        "CustomListEntry", back_populates="edition"
     )
 
     # An Edition may be the presentation edition for many LicensePools.
@@ -146,6 +149,9 @@ class Edition(Base, EditionConstants):
         ForeignKey("resources.id", use_alter=True, name="fk_editions_summary_id"),
         index=True,
     )
+    cover: Mapped[Resource | None] = relationship(
+        "Resource", back_populates="cover_editions", foreign_keys=[cover_id]
+    )
     # These two let us avoid actually loading up the cover Resource
     # every time.
     cover_full_url = Column(Unicode)