Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
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))
@gingerwizard

This comment has been minimized.

Copy link
Owner Author

@gingerwizard gingerwizard commented Jun 8, 2020

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

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.