forgefed/deprecated/5. extending pagure database model/tasks/notification.py

import celery
import json
import pagure
import rdflib

from .. import activitypub
from .. import model
from .. import settings

from . import broker
from . import broker_url
from . import database_session

log = celery.utils.log.get_task_logger(__name__)
log.setLevel(settings.LOG_LEVEL)

# The following is a decorator that accepts a Pagure notification ID as input,
# for example "issue.new", and adds it to _USER_ACTIONS_ together with the
# decorated function. _USER_ACTIONS_ is a dictionary mapping Pagure notification
# IDs to a function to be executed when a new notification is received.
_USER_ACTIONS_ = {}
def action(notification_id):
    def closure(func):
        
        def decorator(*args, **kwargs):
            func(*args, **kwargs)
        
        global _USER_ACTIONS_
        _USER_ACTIONS_[notification_id] = decorator
        
        return decorator
    return closure

@broker.task
def handle_pagure_signal(notification_id, message):
    """
    This task receives notifications from Pagure about events that happen on
    the instance, creates a new activity, and schedules their delivery.
    """
    
    if notification_id not in _USER_ACTIONS_:
        log.debug('Unhandled user action {}.'.format(notification_id))
        return
    
    log.debug('New Pagure notification: {}\n{}'.format(notification_id,
                                                       json.dumps(message, indent=4, sort_keys=True)))
    
    with database_session() as database:
        
        # Handle the user action
        return _USER_ACTIONS_[notification_id](database, message)

@action('issue.new')
def new_issue(database, message):
    """
    A user has created a new Issue.
    """

    person = database \
                .query(model.Person) \
                .filter(model.Person.user == message['issue']['user']['name']) \
                .one_or_none()
    
    project = database \
        .query(model.Project) \
        .filter(model.Project.id == message['project']['id']) \
        .one_or_none()
    
    issue = database \
        .query(pagure.lib.model.Issue) \
        .filter(pagure.lib.model.Issue.id == message['issue']['id'],
                pagure.lib.model.Issue.project_id == message['project']['id']) \
        .one_or_none()
    
    # This should never raise an error otherwise there's a bug in Pagure
    assert person and issue and project, \
        "User, Issue, or Project doesn't exist."
    
    # Check if this tracker is only used to interact with a remote one
    same_as = database.query(database.SameAs) \
                      .filter(database.SameAs.local_uri == project.local_uri) \
                      .one_or_none()
    
    if same_as:
        log.debug('Sending new issue to remote tracker...')
        
        ticket = database \
            .query(model.Ticket) \
            .filter(model.Ticket.id == issue.id,
                    model.Ticket.project_id == issue.project_id) \
            .one_or_none()
        
        # Offer a new ticket to the remote tracker
        # person.offer(ticket.jsonld, to=same_as.remote_uri)

@action('issue.comment.added')
def new_issue_comment(database, message):
            # A user has commented on an issue
            
            # The Pagure notification contains *all* the comments of the issue
            # in an ordered list, so we extract the last one from the list of
            # comments.
            comment = pagure_db \
                .query(model.TicketComment) \
                .filter(model.TicketComment.id == message['issue']['comments'][-1]['id']) \
                .one_or_none()
            
            person = pagure_db \
                        .query(model.Person) \
                        .filter(model.Person.id == comment.user.id) \
                        .one_or_none()
            
            project = pagure_db \
                .query(model.Project) \
                .filter(model.Project.id == message['project']['id']) \
                .one_or_none()
            
            # Our local ticket
            ticket = pagure_db \
                .query(model.Ticket) \
                .filter(model.Ticket.id == message['issue']['id'],
                        model.Ticket.project_id == message['project']['id']) \
                .one_or_none()
            
            # Check if this tracker is only used to interact with a remote one
            same_as = forgefed_db.query(database.SameAs) \
                              .filter(database.SameAs.local_uri == project.local_uri) \
                              .one_or_none()
            
            if same_as:
                log.debug('Sending new comment to remote tracker...')
                
                assert ticket.remote_uri, 'Ticket not linked to remote ticket.'
                
                # Notify the remote tracker about the new comment
                person.create(comment.jsonld, to=same_as.remote_uri)
            
            else:
                log.debug('Not a remote tracker. Will send new comment to tracker followers '
                    + 'and watchers.')
                
                # Retrieve the pagure "watchlist"
                watchlist = pagure.lib.query.get_watch_list(pagure_db, ticket)
                actors = []
                
                for username in watchlist:
                    actor = pagure_db \
                        .query(model.Person) \
                        .filter(model.Person.user == username) \
                        .one_or_none()
                    
                    actors.append(actor.uri)
                
                # Send the Activity
                person.create(comment.jsonld, to=actors)

@action('issue.edit')
def edit_issue(database, message):
        if topic == 'issue.edit' \
        and 'status' in message['fields'] \
        and message['issue']['status'].upper() == 'CLOSED':
            # A user has closed an issue
            
            # The Ticket that was modified
            ticket = pagure_db \
                .query(model.Ticket) \
                .filter(model.Ticket.id == message['issue']['id'],
                        model.Ticket.project_id == message['project']['id']) \
                .one_or_none()
            
            # The user that edited the Ticket
            person = pagure_db \
                        .query(model.Person) \
                        .filter(model.Person.user == message['agent']) \
                        .one_or_none()
            
            project = pagure_db \
                .query(model.Project) \
                .filter(model.Project.id == message['project']['id']) \
                .one_or_none()
            
            # If the user has edited the Ticket of a remote tracker, we just
            # send the Activity to the tracker
            if ticket.is_remote:
                log.debug('Local user has closed the remote Ticket {}'.format(ticket.uri))
                
                person.resolve(ticket.uri, to=project.remote_uri)
            
            # otherwise we send the Activity to the Tickets' watchlist
            else:
                log.debug('Local user has closed the local Ticket {}'.format(ticket.uri))
                
                # Retrieve the pagure "watchlist"
                watchlist = pagure.lib.query.get_watch_list(pagure_db, ticket)
                actors = []
                
                for username in watchlist:
                    actor = pagure_db \
                        .query(model.Person) \
                        .filter(model.Person.user == username) \
                        .one_or_none()
                    
                    actors.append(actor.uri)
                
                # Send the Activity
                person.resolve(ticket.uri, to=actors)