Receive webhooks from GitHub and create or update pages in Notion

sync.py

import os
from uuid import uuid1
from datetime import datetime
from wsgiref.simple_server import make_server
from pyramid.config import Configurator
from pyramid.view import view_config, view_defaults, notfound_view_config
from pyramid.response import Response
from notion.client import NotionClient
from notion.collection import NotionDate

import beeline

beeline.init(
    writekey=os.environ['HONEYCOMB_KEY'],
    dataset='GitHub -> Notion sync',
    service_name='github-notion-sync-webhook'
)
honeycomb_client = beeline.get_beeline().client
    
ENDPOINT = "webhook"

NOTION_CLIENT = NotionClient(token_v2=os.environ["NOTION_TOKEN"])
TABLE = NOTION_CLIENT.get_block(os.environ["TARGET_PAGE"])

# Querying in NotionPy is currently broken due to a Notion API update, so do
# the query directly.
query = {'collectionId': os.environ["TARGET_COLLECTION"],
   'collectionViewId': os.environ["TARGET_VIEW"],
   'loader': {'limit': 2,
              'loadContentCover': True,
              'searchQuery': None,
              'type': 'table',
              'userLocale': 'en',
              'userTimeZone': 'Asia/Bangkok'},
   'query': {}}


def update_labels(labels):
    schema = TABLE.collection.get("schema")
    label_schema = next(
        (v for k, v in schema.items() if v["name"] == 'labels'), None
    )
    changed = False
    for label in labels:
        exists = next(
            (o for o in label_schema["options"] if o["value"] == label["name"]), None
        )
        if exists:
            continue
        changed = True
        label_schema["options"].append(
            {"id": str(uuid1()), "value": label["name"], "color": "gray"}
        )
    if changed:
        TABLE.collection.set("schema", schema)


def update_issue(row, issue):
    changed = []
    for k,v in issue.items():
        if v and (k in ('user', 'assignee', 'closed_by')):
            v = v['login']           
        elif k == 'assignees':
            v = ', '.join([a['login'] for a in v])
        elif v and (k == 'milestone'):
            v = v['title']
            # print(f'Update milestone: {v}')
        elif v and (k == 'pull_request'):
            v = v['html_url']
        elif k == 'labels':
            update_labels(v)
            v = [l['name'] for l in v]
        elif k == 'locked':
            v = True if v == 'True' else False
        elif v and (k in ('created_at', 'updated_at', 'closed_at')):
            # v = datetime.strptime(v, "%Y-%m-%d %H:%M:%S")
            # `_rawData` in PyGithub
            v = datetime.strptime(v, "%Y-%m-%dT%H:%M:%SZ")
        elif v is not None:
            v = str(v)
        try:
            old_value = row.get_property(k)
        except AttributeError:
            print('AttributeError on', k)
            continue
        if type(old_value) is NotionDate:
            old_value = old_value.start
            if v:
                v = v.replace(second=0)
        if v != old_value:
            if v or old_value:
              changed.append(k)
              row.set_property(k, v)
        # if k == 'milestone':
        #   print(f'old: {old_value}, new: {v}, changed: {changed}')
    return changed


@view_defaults(
    route_name=ENDPOINT, renderer="json", request_method="POST"
)
class PayloadView(object):
    """
    View receiving of Github payload. By default, this view is fired only if
    the request is JSON and method POST.
    """

    def __init__(self, request):
        self.request = request
        # Payload from Github, it's a dict
        self.payload = self.request.json

    @view_config(header="X-Github-Event:issues")
    def payload_issues(self):
        honeycomb_event = honeycomb_client.new_event()
        print(f'Syncing {self.payload["issue"]["html_url"]} -- {self.payload["issue"]["title"]}')
        TABLE.refresh()
        if self.payload['action'] in ['opened', 'closed', 'edited', 'labeled', 'unlabeled', 'reopened', 'assigned', 'unassigned', 'milestoned', 'demilestoned']:
            # if self.payload['action'] in ['milestoned', 'demilestoned']:
            #     print(f"{self.payload}")
            print(f'Action: {self.payload["action"]}')
            honeycomb_event.add({'action': self.payload["action"]})
            query['loader']['searchQuery'] = self.payload['issue']['node_id']
            try:
                result = NOTION_CLIENT.post('queryCollection', query).json()
            except:
                print(query)
                honeycomb_event.add({'query': query, 'error': 'Notion query failed'})
                honeycomb_event.send()
                raise
            ids = result['result']['blockIds']
            print(query)
            print(result['result'])
            honeycomb_event.add({'query': query})
            honeycomb_event.add({'result': result})
            if ids:
                row = NOTION_CLIENT.get_block(ids[0])
            else:
                # Maybe someone deleted it in Notion?
                print(f'Creating {self.payload["issue"]["title"]}')
                honeycomb_event.add({'create': self.payload["issue"]["title"]})
                row = TABLE.collection.add_row(update_views=False)
        else:
            print(f"Unhandled action: {self.payload['action']}")
            return Response("success")
        changes = update_issue(row, self.payload['issue'])
        if changes:
            print(f'  Updated https://notion.so/{row.id.replace("-", "")}: {", ".join(changes)}.')
            honeycomb_event.add({'changes': ", ".join(changes)})
        honeycomb_event.send()
        return Response("success")
      
    # TODO: 
    #     @view_config(header="X-Github-Event:pull_request")
    # ...

    @view_config(header="X-Github-Event:ping")
    def payload_else(self):
        print("Pinged! Webhook created with id {}!".format(self.payload["hook"]["id"]))
        return {"status": 200}

@notfound_view_config()
def notfound(request):
   #  print(request)
    return Response('Not Found', status='404 Not Found')


if __name__ == "__main__":
    config = Configurator()
    config.add_route(ENDPOINT, "/{}".format(ENDPOINT))
    config.scan()

    app = config.make_wsgi_app()
    server = make_server("0.0.0.0", 5000, app)
    server.serve_forever()