ingest_trello.py

import json
import re
from itertools import islice, chain
from elastic_workplace_search import Client
import os
import requests

#converts camel case fields from trello to workplace search's prefered snake case
pattern = re.compile(r'(?<!^)(?=[A-Z])')
default_user='ent_search'
s = requests.Session()
s.params.update({'key': os.getenv('TRELLO_API_KEY'), 'token': os.getenv('TRELLO_TOKEN')})
workplace_client = Client(os.getenv('WORKPLACE_SEARCH_TOKEN'), os.getenv('WORKPLACE_SEARCH_URL', 'http://localhost:3002'))
batch_size = int(os.getenv('BATCH_SIZE',10))

def get_boards(org_id):
    for board in s.get('https://api.trello.com/1/organizations/%s/boards' % org_id, params={'fields': 'id,idOrganization,name,desc,dateLastActivity,url,dateLastView'}).json():
        board['body'] = board['desc']
        board['title'] = board['name']
        board['type'] = 'board'
        yield board

def get_orgs():
    return s.get('https://api.trello.com/1/members/me/organizations', params={'fields': 'id,name,displayName,url'}).json()

def get_cards(board_id):
    for card in s.get('https://api.trello.com/1/boards/%s/cards' % board_id, params={'fields': 'id,dateLastActivity,desc,idBoard,name,url,labels,dueon,idList,dueComplete,closed,dateLastActivity'}).json():
        card['labels'] = [label['name'] for label in card['labels']]
        card['title'] = card['name']
        card['body'] = card['desc']
        card['type'] = 'card'
        yield card

def get_card_comments(card_id, card_url):
    for comment in s.get('https://api.trello.com/1/cards/%s/actions' % card_id, params={'filter':'commentCard', 'fields':'memberCreator,id,type,date,data'}).json():
        comment['body'] = comment['data']['text']
        comment['memberCreator'] = comment['memberCreator']['username']
        comment['url'] = '%s#%s' % (card_url, 'comment-%s' % comment['id'])
        comment['title'] = 'Comment by %s at %s' % (comment['memberCreator'], comment['date'])
        comment['type'] = 'comment'
        del comment['data']
        yield comment

def snake_case_fields(doc):
    if isinstance(doc,dict):
        return {pattern.sub('_', k).lower(): snake_case_fields(v) for k, v in doc.items()}
    elif isinstance(doc, list):
        return [snake_case_fields(sub_doc) for sub_doc in doc]
    return doc

def get_trello_objects():
    for org in get_orgs():
        print('Fetching boards for %s' % org['displayName'])
        for board in get_boards(org['id']):
            print('Processing board %s' % board['id'])
            #for each board
            yield snake_case_fields(board)
            #get the cards for the board
            print('Fetching cards for board %s' % board['id'])
            for card in get_cards(board['id']):
                yield snake_case_fields(card)
                for comment in get_card_comments(card['id'], card['url']):
                    yield snake_case_fields(comment)

def batch_docs(iterable, n=1):
    source_iter = iter(iterable)
    try:
        while True:
            batch_iter = islice(source_iter, n)
            yield chain([next(batch_iter)], batch_iter)
    except StopIteration:
        return

print('Ingesting Content...')
for batch in batch_docs(get_trello_objects(), n=batch_size):
    docs = list(batch)
    workplace_client.documents.index_documents(os.environ['WORKPLACE_SEARCH_KEY'], list(batch))
    print('Indexed %s docs' % len(docs))

Pre-requisites

• Python 3.8 & Pip
• Elastic Workplace Search client - pip install elastic-workplace-search
• Requests - pip install requests

Ensure the following environment variables are set:

• TRELLO_API_KEY - Your Trello API Key
• TRELLO_TOKEN - Trello App token
• WORKPLACE_SEARCH_TOKEN - Workplace Search Token
• WORKPLACE_SEARCH_KEY - Workplace Search Custom Source Key
• WORKPLACE_SEARCH_URL (Optional) - If different than http://localhost:3002
• BATCH_SIZE (Optional) - defaults to 10

Running

python ingest_trello.py