Purge owned relationships #452

project/models/__init__.py

@@ -31,11 +31,13 @@ from project.models.event_suggestion import (
 )
 from project.models.functions import sanitize_allday_instance
 from project.models.image import Image
+from project.models.iowned import IOwned
 from project.models.legacy import (
     FeaturedEventRejectionReason,
     FeaturedEventReviewStatus,
 )
 from project.models.location import Location
 from project.models.oauth import OAuth2AuthorizationCode, OAuth2Client, OAuth2Token
+from project.models.session_events import before_flush
 from project.models.settings import Settings
 from project.models.user import OAuth, Role, User, UserFavoriteEvents

project/models/admin_unit.py

@@ -203,11 +203,19 @@ class AdminUnit(db.Model, TrackableMixin):
     )
     location_id = deferred(db.Column(db.Integer, db.ForeignKey("location.id")))
     location = db.relationship(
-        "Location", uselist=False, single_parent=True, cascade="all, delete-orphan"
+        "Location",
+        uselist=False,
+        single_parent=True,
+        cascade="all, delete-orphan",
+        back_populates="adminunit",
     )
     logo_id = deferred(db.Column(db.Integer, db.ForeignKey("image.id")))
     logo = db.relationship(
-        "Image", uselist=False, single_parent=True, cascade="all, delete-orphan"
+        "Image",
+        uselist=False,
+        single_parent=True,
+        cascade="all, delete-orphan",
+        back_populates="adminunit",
     )
     url = deferred(Column(String(255)), group="detail")
     email = deferred(Column(Unicode(255)), group="detail")
@@ -313,16 +321,6 @@ class AdminUnit(db.Model, TrackableMixin):
             > 0
         )
 
-    def purge(self):
-        if self.logo and self.logo.is_empty():
-            self.logo_id = None
-
-
-@listens_for(AdminUnit, "before_insert")
-@listens_for(AdminUnit, "before_update")
-def before_saving_admin_unit(mapper, connect, self):
-    self.purge()
-
 
 @listens_for(AdminUnit.can_invite_other, "set")
 def set_admin_unit_can_invite_other(target, value, oldvalue, initiator):

project/models/event.py

@@ -192,4 +192,3 @@ class Event(db.Model, TrackableMixin, EventMixin):
 @listens_for(Event, "before_update")
 def before_saving_event(mapper, connect, self):
     self.validate()
-    self.purge_event_mixin()

project/models/event_mixin.py

@@ -56,9 +56,10 @@ class EventMixin(object):
     @declared_attr
     def photo(cls):
         return relationship(
-            "Image", uselist=False, single_parent=True, cascade="all, delete-orphan"
+            "Image",
+            uselist=False,
+            single_parent=True,
+            cascade="all, delete-orphan",
+            foreign_keys=[cls.photo_id],
+            back_populates=cls.__tablename__,
         )
-
-    def purge_event_mixin(self):
-        if self.photo and self.photo.is_empty():
-            self.photo_id = None

project/models/event_organizer.py

@@ -1,5 +1,4 @@
 from sqlalchemy import Column, Integer, String, Unicode, UniqueConstraint
-from sqlalchemy.event import listens_for
 from sqlalchemy.orm import deferred
 
 from project import db
@@ -17,22 +16,23 @@ class EventOrganizer(db.Model, TrackableMixin):
     fax = deferred(Column(Unicode(255)), group="detail")
     location_id = deferred(db.Column(db.Integer, db.ForeignKey("location.id")))
     location = db.relationship(
-        "Location", uselist=False, single_parent=True, cascade="all, delete-orphan"
+        "Location",
+        uselist=False,
+        single_parent=True,
+        cascade="all, delete-orphan",
+        back_populates="eventorganizer",
     )
     logo_id = deferred(db.Column(db.Integer, db.ForeignKey("image.id")))
     logo = db.relationship(
-        "Image", uselist=False, single_parent=True, cascade="all, delete-orphan"
+        "Image",
+        uselist=False,
+        single_parent=True,
+        cascade="all, delete-orphan",
+        back_populates="eventorganizer",
     )
     admin_unit_id = db.Column(db.Integer, db.ForeignKey("adminunit.id"), nullable=True)
 
 
-@listens_for(EventOrganizer, "before_insert")
-@listens_for(EventOrganizer, "before_update")
-def purge_event_organizer(mapper, connect, self):
-    if self.logo and self.logo.is_empty():
-        self.logo_id = None
-
-
 class EventCoOrganizers(db.Model):
     __tablename__ = "event_coorganizers"
     __table_args__ = (UniqueConstraint("event_id", "organizer_id"),)

project/models/event_place.py

@@ -1,5 +1,4 @@
 from sqlalchemy import Column, Integer, String, Unicode, UnicodeText, UniqueConstraint
-from sqlalchemy.event import listens_for
 
 from project import db
 from project.models.trackable_mixin import TrackableMixin
@@ -12,21 +11,20 @@ class EventPlace(db.Model, TrackableMixin):
     name = Column(Unicode(255), nullable=False)
     location_id = db.Column(db.Integer, db.ForeignKey("location.id"))
     location = db.relationship(
-        "Location", uselist=False, single_parent=True, cascade="all, delete-orphan"
+        "Location",
+        uselist=False,
+        single_parent=True,
+        cascade="all, delete-orphan",
+        back_populates="eventplace",
     )
     photo_id = db.Column(db.Integer, db.ForeignKey("image.id"))
     photo = db.relationship(
-        "Image", uselist=False, single_parent=True, cascade="all, delete-orphan"
+        "Image",
+        uselist=False,
+        single_parent=True,
+        cascade="all, delete-orphan",
+        back_populates="eventplace",
     )
     url = Column(String(255))
     description = Column(UnicodeText())
     admin_unit_id = db.Column(db.Integer, db.ForeignKey("adminunit.id"), nullable=True)
-
-
-@listens_for(EventPlace, "before_insert")
-@listens_for(EventPlace, "before_update")
-def purge_event_place(mapper, connect, self):
-    if self.location and self.location.is_empty():
-        self.location_id = None
-    if self.photo and self.photo.is_empty():
-        self.photo_id = None

project/models/event_suggestion.py

@@ -88,5 +88,4 @@ def purge_event_suggestion(mapper, connect, self):
         self.organizer_text = None
     if self.event_place_id is not None:
         self.event_place_text = None
-    self.purge_event_mixin()
     sanitize_allday_instance(self)

project/models/image.py

@@ -3,16 +3,23 @@ from sqlalchemy.orm import deferred
 
 from project import db
 from project.dateutils import gmt_tz
+from project.models.iowned import IOwned
 from project.models.trackable_mixin import TrackableMixin
 
 
-class Image(db.Model, TrackableMixin):
+class Image(db.Model, TrackableMixin, IOwned):
     __tablename__ = "image"
     id = Column(Integer(), primary_key=True)
     data = deferred(db.Column(db.LargeBinary))
     encoding_format = Column(String(80))
     copyright_text = Column(Unicode(255))
 
+    adminunit = db.relationship("AdminUnit", uselist=False)
+    event = db.relationship("Event", uselist=False)
+    eventorganizer = db.relationship("EventOrganizer", uselist=False)
+    eventplace = db.relationship("EventPlace", uselist=False)
+    eventsuggestion = db.relationship("EventSuggestion", uselist=False)
+
     def is_empty(self):
         return not self.data
 
@@ -25,3 +32,23 @@ class Image(db.Model, TrackableMixin):
 
     def get_file_extension(self):
         return self.encoding_format.split("/")[-1] if self.encoding_format else "png"
+
+    def before_flush(self, session, is_dirty):
+        if self.is_empty():
+            if self.adminunit:
+                self.adminunit.logo = None
+
+            if self.event:
+                self.event.photo = None
+
+            if self.eventorganizer:
+                self.eventorganizer.logo = None
+
+            if self.eventplace:
+                self.eventplace.photo = None
+
+            if self.eventsuggestion:
+                self.eventsuggestion.photo = None
+
+            if is_dirty:
+                session.delete(self)

project/models/iowned.py

@@ -0,0 +1,3 @@
+class IOwned:
+    def before_flush(self, is_dirty):  # pragma: no cover
+        raise NotImplementedError

project/models/location.py

@@ -3,10 +3,11 @@ from sqlalchemy import Column, Integer, Numeric, Unicode, and_
 from sqlalchemy.event import listens_for
 
 from project import db
+from project.models.iowned import IOwned
 from project.models.trackable_mixin import TrackableMixin
 
 
-class Location(db.Model, TrackableMixin):
+class Location(db.Model, TrackableMixin, IOwned):
     __tablename__ = "location"
     id = Column(Integer(), primary_key=True)
     street = Column(Unicode(255))
@@ -18,6 +19,10 @@ class Location(db.Model, TrackableMixin):
     longitude = Column(Numeric(19, 16))
     coordinate = Column(Geometry(geometry_type="POINT"))
 
+    adminunit = db.relationship("AdminUnit", uselist=False)
+    eventorganizer = db.relationship("EventOrganizer", uselist=False)
+    eventplace = db.relationship("EventPlace", uselist=False)
+
     def __init__(self, **kwargs):
         super(Location, self).__init__(**kwargs)
 
@@ -54,6 +59,20 @@ class Location(db.Model, TrackableMixin):
 
         db.session.commit()
 
+    def before_flush(self, session, is_dirty):
+        if self.is_empty():
+            if self.adminunit:
+                self.adminunit.location = None
+
+            if self.eventplace:
+                self.eventplace.location = None
+
+            if self.eventorganizer:
+                self.eventorganizer.location = None
+
+            if is_dirty:
+                session.delete(self)
+
 
 @listens_for(Location, "before_insert")
 @listens_for(Location, "before_update")

project/models/session_events.py

@@ -0,0 +1,15 @@
+from sqlalchemy import event
+from sqlalchemy.orm import Session
+
+from project.models import IOwned
+
+
+@event.listens_for(Session, "before_flush")
+def before_flush(session, flush_context, instances):
+    for instance in session.dirty:
+        if isinstance(instance, IOwned):
+            instance.before_flush(session, True)
+
+    for instance in session.new:
+        if isinstance(instance, IOwned):
+            instance.before_flush(session, False)

tests/seeder.py

@@ -717,3 +717,27 @@ class Seeder(object):
                 "Verein",
             )
             self.create_event(verein_admin_unit_id)
+
+    def create_location(
+        self,
+        street=None,
+        postalCode=None,
+        city=None,
+        latitude=None,
+        longitude=None,
+    ):
+        from project.models import Location
+
+        with self._app.app_context():
+            location = Location(
+                street=street,
+                postalCode=postalCode,
+                city=city,
+                latitude=latitude,
+                longitude=longitude,
+            )
+            self._db.session.add(location)
+            self._db.session.commit()
+            location_id = location.id
+
+        return location_id

tests/test_models.py

@@ -390,3 +390,150 @@ def test_event_is_favored_by_current_user(client, app, db, seeder):
 
         event = db.session.get(Event, event_id)
         assert event.is_favored_by_current_user() is False
+
+
+def test_purge_event_photo(client, app, db, seeder):
+    _, admin_unit_id = seeder.setup_base(log_in=False)
+    event_id = seeder.create_event(admin_unit_id)
+    first_image_id = seeder.upsert_default_image()
+    seeder.assign_image_to_event(event_id, first_image_id)
+
+    with app.app_context():
+        from project.models import Event, Image
+
+        event = db.session.get(Event, event_id)
+        assert event.photo is not None
+
+        event.photo.data = None
+        db.session.commit()
+
+        event = db.session.get(Event, event_id)
+        assert event.photo is None
+
+        image = db.session.get(Image, first_image_id)
+        assert image is None
+
+
+def test_purge_event_place_photo(client, app, db, seeder):
+    _, admin_unit_id = seeder.setup_base(log_in=False)
+    place_id = seeder.upsert_default_event_place(admin_unit_id)
+    first_image_id = seeder.upsert_default_image()
+    second_image_id = seeder.upsert_default_image()
+
+    with app.app_context():
+        from project.models import EventPlace, Image
+
+        place = db.session.get(EventPlace, place_id)
+        place.photo = db.session.get(Image, first_image_id)
+        db.session.commit()
+
+        assert place.photo is not None
+
+        place.photo = db.session.get(Image, second_image_id)
+        db.session.commit()
+
+        place = db.session.get(EventPlace, place_id)
+        assert place.photo is not None
+
+        image = db.session.get(Image, first_image_id)
+        assert image is None
+
+        image = db.session.get(Image, second_image_id)
+        assert image is not None
+
+        place.photo.data = None
+        db.session.commit()
+
+        place = db.session.get(EventPlace, place_id)
+        assert place.photo is None
+
+        image = db.session.get(Image, second_image_id)
+        assert image is None
+
+
+def test_purge_eventsuggestion_photo(client, app, db, seeder):
+    _, admin_unit_id = seeder.setup_base(log_in=False)
+    suggestion_id = seeder.create_event_suggestion(admin_unit_id)
+    image_id = seeder.upsert_default_image()
+
+    with app.app_context():
+        from project.models import EventSuggestion, Image
+
+        suggestion = db.session.get(EventSuggestion, suggestion_id)
+        suggestion.photo = db.session.get(Image, image_id)
+        db.session.commit()
+
+        assert suggestion.photo is not None
+
+        suggestion.photo.data = None
+        db.session.commit()
+
+        suggestion = db.session.get(EventSuggestion, suggestion_id)
+        assert suggestion.photo is None
+
+        image = db.session.get(Image, image_id)
+        assert image is None
+
+
+def test_purge_adminunit(client, app, db, seeder):
+    _, admin_unit_id = seeder.setup_base(log_in=False)
+    instance_id = admin_unit_id
+    image_id = seeder.upsert_default_image()
+    location_id = seeder.create_location(street="Street")
+
+    with app.app_context():
+        from project.models import AdminUnit, Image, Location
+
+        instance = db.session.get(AdminUnit, instance_id)
+        instance.logo = db.session.get(Image, image_id)
+        instance.location = db.session.get(Location, location_id)
+        db.session.commit()
+
+        assert instance.logo is not None
+        assert instance.location is not None
+
+        instance.logo.data = None
+        instance.location.street = None
+        db.session.commit()
+
+        instance = db.session.get(AdminUnit, instance_id)
+        assert instance.logo is None
+        assert instance.location is None
+
+        image = db.session.get(Image, image_id)
+        assert image is None
+
+        location = db.session.get(Location, location_id)
+        assert location is None
+
+
+def test_purge_eventorganizer(client, app, db, seeder):
+    _, admin_unit_id = seeder.setup_base(log_in=False)
+    instance_id = seeder.upsert_default_event_organizer(admin_unit_id)
+    image_id = seeder.upsert_default_image()
+    location_id = seeder.create_location(street="Street")
+
+    with app.app_context():
+        from project.models import EventOrganizer, Image, Location
+
+        instance = db.session.get(EventOrganizer, instance_id)
+        instance.logo = db.session.get(Image, image_id)
+        instance.location = db.session.get(Location, location_id)
+        db.session.commit()
+
+        assert instance.logo is not None
+        assert instance.location is not None
+
+        instance.logo.data = None
+        instance.location.street = None
+        db.session.commit()
+
+        instance = db.session.get(EventOrganizer, instance_id)
+        assert instance.logo is None
+        assert instance.location is None
+
+        image = db.session.get(Image, image_id)
+        assert image is None
+
+        location = db.session.get(Location, location_id)
+        assert location is None