Skip to content

Instantly share code, notes, and snippets.

@waisbrot

waisbrot/__init__.py

Created March 23, 2018 01:11
Show Gist options
  • Save waisbrot/6309146011606d3bba7cdac68b95b34c to your computer and use it in GitHub Desktop.
Save waisbrot/6309146011606d3bba7cdac68b95b34c to your computer and use it in GitHub Desktop.
Download ZIP
Nationbuilder demo
Raw
__init__.py
from flask import Flask, url_for, redirect, request, render_template, Markup
from .nbapi import NBReal, NBMock, RequestError
import os
import json
import sys
from traceback import format_tb
app = Flask(__name__)
if os.getenv('NB_MOCK', default='false').lower() == 'true':
api = NBMock()
else:
api = NBReal(os.environ.get('NATION'), os.environ.get('API_KEY'))
@app.route('/')
def root():
menu_items = [
{'name': 'People', 'url': url_for('people_base')},
{'name': 'Webhooks', 'url': url_for('webhooks_base')},
{'name': 'Contact', 'url': url_for('contact_base')},
]
return render_template('root.html', links=menu_items)
@app.route('/contact/', methods=['GET'])
def contact_base():
people = api.sample_people()
types = api.sample_contact_types()
return render_template('contact.html', people=people, types=types)
@app.route('/contact/create/', methods=['POST'])
def contact_create():
contact = {
'type_id': int(request.form['type_id']),
'person_id': int(request.form['person_id']),
}
try:
result = api.create_contact(contact)
pretty_json = json.dumps(result, sort_keys=True, indent=4)
return render_template('contact_result.html', contact=result, pretty_json=pretty_json)
except RequestError:
return _error_page(return_url=url_for('contact_base'))
@app.route('/people/', methods=['GET'])
def people_base():
return render_template('people.html', people=api.sample_people())
def _error_page(return_url):
(_, error, stack) = sys.exc_info()
traceback = ''.join(format_tb(stack))
return render_template('error.html', error_message=error.message, error_raw=traceback, return_url=return_url)
@app.route('/people/create/', methods=['POST'])
def people_create():
person = {
'first_name': request.form['first'],
'last_name': request.form['last'],
'email': request.form['email']
}
try:
result = api.create_person(person)
pretty_json = json.dumps(result, sort_keys=True, indent=4)
return render_template('person.html', person=result, pretty_json=pretty_json)
except RequestError:
return _error_page(return_url=url_for('people_base'))
@app.route('/people/update/', methods=['POST'])
def people_update():
person = {
'id': int(request.form['id']),
'note': request.form['note']
}
try:
result = api.update_person(person)
pretty_json = json.dumps(result, sort_keys=True, indent=4)
return render_template('person.html', person=result, pretty_json=pretty_json)
except RequestError:
return _error_page(return_url=url_for('people_base'))
@app.route('/people/delete/', methods=['POST'])
def people_delete():
person_id = int(request.form['id'])
try:
api.delete_person(person_id)
return redirect(url_for('people_base'))
except RequestError:
return _error_page(return_url=url_for('people_base'))
@app.route('/webhooks/', methods=['GET'])
def webhooks_base():
return render_template('webhooks.html', webhooks=api.sample_webhooks())
@app.route('/webhooks/create/', methods=['POST'])
def webhooks_create():
hook = {
'version': 4,
'url': request.form['url'],
'event': request.form['event'],
}
try:
result = api.create_webhook(hook)
pretty_json = json.dumps(result, sort_keys=True, indent=4)
return render_template('wehbook.html', hook=result, pretty_json=pretty_json)
except RequestError:
return _error_page(return_url=url_for('webhooks_base'))
Raw
nbapi.py
"""
Skeleton implementation of NB API class plus mocking
"""
from abc import ABC, abstractmethod
import logging
from uuid import uuid1 as uuid
import datetime
log = logging.getLogger(__name__)
class RequestError(Exception):
def __init__(self, message):
self.message = message
class NBAPI(ABC):
'''
ABC to assert that the mock and real implementations are compatible
'''
@abstractmethod
def sample_people(self):
pass
@abstractmethod
def create_person(self, data):
pass
@abstractmethod
def update_person(self, data):
pass
@abstractmethod
def delete_person(self, pid):
pass
@abstractmethod
def sample_webhooks(self):
pass
@abstractmethod
def create_webook(self, data):
pass
@abstractmethod
def create_contact(self, data):
pass
@abstractmethod
def sample_contact_types(self):
pass
class NBReal(NBAPI):
'''
Actual API that makes live requests
'''
def __init__(self, nation, api_key):
self.base_url = 'https://{}.nationbuilder.com/api/v1'.format(nation)
self.params = {'access_token': api_key}
@staticmethod
def _assert_response_ok(response):
if response.status_code >= 300 or response.status_code < 200:
raise RequestError('Request failed. Server sent {}: {}'.format(action, response.status_code, response.text))
def _post(self, ep, data):
response = requests.post(base_url + ep, params=self.params, json=data)
self._assert_response_ok(response)
return response
def _put(self, ep, data):
response = requests.put(base_url + ep, params=self.params, json=data)
self._assert_response_ok(response)
return response
def _delete(self, ep):
response = requests.delete(base_url + ep, params=self.params)
self._assert_response_ok(response)
def sample_people(self):
response = self._get('/people?limit=10')
return response['results'] # no paging; this is just a sample
def create_person(self, data):
response = self._post('/people', {'person': data})
return response['person']
def update_person(self, data):
response = self._put('/people/{}'.format(data['id']), {'person': data})
return response['person']
def delete_person(self, pid):
self._delete('/people/{}'.format(pid))
def sample_webhooks(self):
response = self._get('/webhooks?limit=10')
return response['results'] # no paging; this is just a sample
def create_webook(self, data):
response = self._post('/webooks', {'webhook': data})
return response['webhook']
def create_contact(self, data):
response = self._post('/people/{}/contacts'.format(data['person_id'], {'contact': data}))
return response['contact']
def sample_contact_types(self):
response = self._get('/settings/contact_types?limit=10')
return response['results'] # no paging; this is just a sample
class NBMock(NBAPI):
'''
Mock API that does not make requests
'''
def __init__(self):
self.people = {}
self.last_people_id = 0
self.webhooks = {}
self.contact_types = [
{'id': 1, 'name': 'Initial outreach'},
{'id': 2, 'name': 'Final outreach'},
]
def sample_people(self):
return self.people.values()
def sample_webhooks(self):
return self.webhooks.values()
def create_person(self, data):
self.last_people_id = self.last_people_id + 1
data['id'] = self.last_people_id
data['contacts'] = {}
data['last_call_id'] = -1
data['last_contacted_at'] = None
self.people[self.last_people_id] = data
return data
def create_webook(self, data):
webhook_id = uuid() # NB id doesn't actually appear to be a UUID
data['id'] = webhook_id
self.webhooks[webhook_id] = data
return data
def update_person(self, data):
pid = data['id']
if pid in self.people:
self.people[pid].update(data)
return self.people[pid]
else:
raise RequestError('No such person id {}. IDs: {}'.format(pid, self.people.keys()))
def delete_person(self, pid):
if pid in self.people:
del self.people[pid]
else:
raise RequestError('No such person id {}'.format(pid))
def create_contact(self, data):
pid = data['person_id']
if pid in self.people:
person_data = self.people[pid]
cid = person_data['last_call_id'] + 1
data['contact_id'] = cid
data['created_at'] = datetime.datetime.utcnow().isoformat(u'T', 'seconds') + 'Z'
person_data['last_call_id'] = cid
person_data['last_contacted_at'] = data['created_at']
person_data['contacts'][cid] = data
return data
else:
raise RequestError('No such person id {}'. format(pid))
def sample_contact_types(self):
return self.contact_types
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment