def cleanup_preview_destination()

in kinto-remote-settings/src/kinto_remote_settings/signer/listeners.py [0:0]

• 75 lines of code
• 15 McCabe index (conditional complexity)

def cleanup_preview_destination(event, resources):
    storage = event.request.registry.storage
    permission = event.request.registry.permission
    settings = event.request.registry.settings

    for impacted in event.impacted_objects:
        old_collection = impacted["old"]

        source_bid = event.payload["bucket_id"]
        source_bucket_uri = instance_uri(event.request, "bucket", id=source_bid)
        source_cid = old_collection["id"]

        resource, signer = pick_resource_and_signer(
            event.request,
            resources,
            bucket_id=source_bid,
            collection_id=source_cid,
        )
        if resource is None:
            continue

        # Delete groups (without tombstones if soft delete)
        should_hard_delete = asbool(
            settings["signer.hard_delete_destination_on_source_deletion"]
        )
        for obj_id in (f"{source_cid}-editors", f"{source_cid}-reviewers"):
            try:
                storage.delete(
                    resource_name="group",
                    parent_id=source_bucket_uri,
                    object_id=obj_id,
                    with_deleted=not should_hard_delete,
                )
            except ObjectNotFoundError:
                # Already deleted
                pass

        for k in ("preview", "destination"):
            if k not in resource:  # pragma: no cover
                continue
            bid = resource[k]["bucket"]
            cid = resource[k]["collection"]
            bucket_uri = instance_uri(event.request, "bucket", id=bid)
            collection_uri = instance_uri(
                event.request, "collection", bucket_id=bid, id=cid
            )

            if should_hard_delete:
                # Delete the records and the collection.
                storage.delete_all(
                    resource_name=None, parent_id=collection_uri, with_deleted=False
                )
                try:
                    storage.delete(
                        resource_name="collection",
                        parent_id=bucket_uri,
                        object_id=cid,
                        with_deleted=False,
                    )
                except ObjectNotFoundError:
                    # Already deleted
                    pass
                # Delete all tombstones and timestamps of this collection.
                storage.purge_deleted(resource_name=None, parent_id=collection_uri)
                # As well as individual permissions on the collection.
                permission.delete_object_permissions(collection_uri)
                # And delete everything related to this collection and underneath
                # (records, history, permissions on records, ...)
                storage.purge_deleted(
                    resource_name=None, parent_id=collection_uri + "/*"
                )
                permission.delete_object_permissions(collection_uri + "/*")
            else:
                # Delete records and leave tombstones.
                storage.delete_all(
                    resource_name="record", parent_id=collection_uri, with_deleted=True
                )

                updater = LocalUpdater(
                    signer=signer,
                    storage=storage,
                    permission=event.request.registry.permission,
                    source=resource["source"],
                    destination=resource[k],
                )

                # At this point, the DELETE event was sent for the source collection,
                # but the source records may not have been deleted yet (it happens in an
                # event listener too). That's why we don't copy the records otherwise it
                # will recreate the records that were just deleted.
                updater.sign_and_update_destination(
                    event.request,
                    source_attributes=old_collection,
                    next_source_status=None,
                    push_records=False,
                )