def includeme()

in kinto-remote-settings/src/kinto_remote_settings/signer/__init__.py [0:0]
132 lines of code
13 McCabe index (conditional complexity)

def includeme(config):
    # Register heartbeat to check signer integration.
    config.registry.heartbeats["signer"] = heartbeat

    resources = load_signed_resources_configuration(config)

    settings = config.get_settings()

    global_settings = {
        k: v
        for k, v in config.registry.api_capabilities["signer"].items()
        if k in ("editors_group", "reviewers_group", "to_review_enabled")
    }

    # Since we have settings that can contain glob patterns, we refresh the settings
    # and exposed resources when a new collection is created.
    config.add_subscriber(
        lambda _: load_signed_resources_configuration(config),
        ResourceChanged,
        for_actions=(ACTIONS.CREATE,),
        for_resources=("collection",),
    )

    config.add_subscriber(on_review_approved, ReviewApproved)

    config.add_subscriber(
        functools.partial(listeners.set_work_in_progress_status, resources=resources),
        ResourceChanged,
        for_resources=("record",),
    )

    config.add_subscriber(
        functools.partial(
            listeners.check_collection_status, resources=resources, **global_settings
        ),
        ResourceChanged,
        for_actions=(ACTIONS.CREATE, ACTIONS.UPDATE),
        for_resources=("collection",),
    )

    config.add_subscriber(
        functools.partial(listeners.check_collection_tracking, resources=resources),
        ResourceChanged,
        for_actions=(ACTIONS.CREATE, ACTIONS.UPDATE),
        for_resources=("collection",),
    )

    config.add_subscriber(
        functools.partial(
            listeners.create_editors_reviewers_groups,
            resources=resources,
            editors_group=global_settings["editors_group"],
            reviewers_group=global_settings["reviewers_group"],
        ),
        ResourceChanged,
        for_actions=(ACTIONS.CREATE,),
        for_resources=("collection",),
    )

    config.add_subscriber(
        functools.partial(listeners.cleanup_preview_destination, resources=resources),
        ResourceChanged,
        for_actions=(ACTIONS.DELETE,),
        for_resources=("collection",),
    )

    config.add_subscriber(
        functools.partial(listeners.prevent_collection_delete, resources=resources),
        ResourceChanged,
        for_actions=(ACTIONS.DELETE,),
        for_resources=("collection",),
    )

    if not asbool(settings["signer.allow_floats"]):
        config.add_subscriber(
            functools.partial(listeners.prevent_float_value, resources=resources),
            ResourceChanged,
            for_actions=(ACTIONS.CREATE, ACTIONS.UPDATE),
            for_resources=("record",),
        )

    sign_data_listener = functools.partial(
        listeners.sign_collection_data, resources=resources, **global_settings
    )
    timed_listener = core_metrics.listener_with_timer(
        config, "plugins.signer", sign_data_listener
    )

    config.add_subscriber(
        timed_listener,
        ResourceChanged,
        for_actions=(ACTIONS.CREATE, ACTIONS.UPDATE),
        for_resources=("collection",),
    )

    def on_new_request(event):
        """Send the signer events in the before commit hook.
        This allows database operations done in subscribers to be automatically
        committed or rolledback.
        """
        # Since there is one transaction per batch, ignore subrequests.
        if hasattr(event.request, "parent"):
            return
        current = transaction.get()
        current.addBeforeCommitHook(listeners.send_signer_events, args=(event,))

    config.add_subscriber(on_new_request, NewRequest)

    try:
        from kinto_emailer import build_notification

        config.add_subscriber(build_notification, ReviewRequested)
        config.add_subscriber(build_notification, ReviewApproved)
        config.add_subscriber(build_notification, ReviewRejected)
    except ImportError:  # pragma: no cover
        pass

    # Automatically create resources on startup if option is enabled.
    def auto_create_resources(event, resources):
        storage = event.app.registry.storage
        permission = event.app.registry.permission
        write_principals = aslist(
            event.app.registry.settings["signer.auto_create_resources_principals"]
        )

        for resource in resources.values():
            perms = {"write": write_principals}
            bucket = resource["source"]["bucket"]
            collection = resource["source"]["collection"]

            bucket_uri = f"/buckets/{bucket}"
            utils.storage_create_raw(
                storage_backend=storage,
                permission_backend=permission,
                resource_name="bucket",
                parent_id="",
                object_uri=bucket_uri,
                object_id=bucket,
                permissions=perms,
            )

            # If resource is configured for specific collection, create it too.
            if collection:
                collection_uri = f"{bucket_uri}/collections/{collection}"
                utils.storage_create_raw(
                    storage_backend=storage,
                    permission_backend=permission,
                    resource_name="collection",
                    parent_id=bucket_uri,
                    object_uri=collection_uri,
                    object_id=collection,
                    permissions=perms,
                )

    # Create resources on startup (except when executing `migrate`).
    if (
        asbool(settings.get("signer.auto_create_resources", False))
        and not IS_RUNNING_MIGRATE
    ):
        config.add_subscriber(
            functools.partial(
                auto_create_resources,
                resources=resources,
            ),
            ApplicationCreated,
        )