def subscribe()

in Allura/allura/model/notification.py [0:0]

• 58 lines of code
• 11 McCabe index (conditional complexity)

    def subscribe(
            cls,
            user_id=None, project_id=None, app_config_id=None,
            artifact=None, topic=None,
            type='direct', n=1, unit='day'):
        if user_id is None:
            user_id = c.user._id
        if project_id is None:
            project_id = c.project._id
        if app_config_id is None:
            app_config_id = c.app.config._id
        tool_already_subscribed = cls.query.get(user_id=user_id,
                                                project_id=project_id,
                                                app_config_id=app_config_id,
                                                artifact_index_id=None)
        if tool_already_subscribed:
            return
        if artifact is None:
            artifact_title = 'All artifacts'
            artifact_url = None
            artifact_index_id = None
        else:
            i = artifact.index()
            artifact_title = h.get_first(i, 'title')
            artifact_url = artifact.url()
            artifact_index_id = i['id']
            artifact_already_subscribed = cls.query.get(user_id=user_id,
                                                        project_id=project_id,
                                                        app_config_id=app_config_id,
                                                        artifact_index_id=artifact_index_id)
            if artifact_already_subscribed:
                return
        d = dict(
            user_id=user_id, project_id=project_id, app_config_id=app_config_id,
            artifact_index_id=artifact_index_id, topic=topic)
        sess = session(cls)
        try:
            mbox = cls(
                type=type, frequency=dict(n=n, unit=unit),
                artifact_title=artifact_title,
                artifact_url=artifact_url,
                **d)
            sess.flush(mbox)
        except pymongo.errors.DuplicateKeyError:
            sess.expunge(mbox)
            mbox = cls.query.get(**d)
            mbox.artifact_title = artifact_title
            mbox.artifact_url = artifact_url
            mbox.type = type
            mbox.frequency.n = n
            mbox.frequency.unit = unit
            sess.flush(mbox)
        if not artifact_index_id:
            # Unsubscribe from individual artifacts when subscribing to the
            # tool
            for other_mbox in cls.query.find(dict(
                    user_id=user_id, project_id=project_id, app_config_id=app_config_id)):
                if other_mbox != mbox:
                    other_mbox.delete()
        return mbox