From 4fe45f942d5ff1ab5a5e00ecb93b2ad234d13faa Mon Sep 17 00:00:00 2001 From: Lance Edgar Date: Mon, 8 May 2023 14:54:36 -0500 Subject: [PATCH] Add basic web API client for NationBuilder --- .../nationbuilder/__init__.py | 0 rattail_nationbuilder/nationbuilder/webapi.py | 118 ++++++++++++++++++ 2 files changed, 118 insertions(+) create mode 100644 rattail_nationbuilder/nationbuilder/__init__.py create mode 100644 rattail_nationbuilder/nationbuilder/webapi.py diff --git a/rattail_nationbuilder/nationbuilder/__init__.py b/rattail_nationbuilder/nationbuilder/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/rattail_nationbuilder/nationbuilder/webapi.py b/rattail_nationbuilder/nationbuilder/webapi.py new file mode 100644 index 0000000..a972c6b --- /dev/null +++ b/rattail_nationbuilder/nationbuilder/webapi.py @@ -0,0 +1,118 @@ +# -*- coding: utf-8; -*- +################################################################################ +# +# Rattail -- Retail Software Framework +# Copyright © 2010-2023 Lance Edgar +# +# This file is part of Rattail. +# +# Rattail is free software: you can redistribute it and/or modify it under the +# terms of the GNU General Public License as published by the Free Software +# Foundation, either version 3 of the License, or (at your option) any later +# version. +# +# Rattail is distributed in the hope that it will be useful, but WITHOUT ANY +# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS +# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more +# details. +# +# You should have received a copy of the GNU General Public License along with +# Rattail. If not, see . +# +################################################################################ +""" +NationBuilder Web API +""" + +import requests + + +class NationBuilderWebAPI(object): + """ + Simple web API for NationBuilder. + """ + + def __init__(self, config, base_url=None, access_token=None, **kwargs): + self.config = config + self.app = self.config.get_app() + + self.base_url = base_url or self.config.require( + 'nationbuilder', 'api.base_url') + self.base_url = self.base_url.rstrip('/') + + self.access_token = access_token or self.config.require( + 'nationbuilder', 'api.access_token') + + def _request(self, request_method, api_method, params=None): + """ + Perform a request for the given API method, and return the response. + """ + api_method = api_method.lstrip('/') + + params = params or {} + params['access_token'] = self.access_token + + if request_method == 'GET': + response = requests.get('{}/{}'.format(self.base_url, api_method), + params=params) + + else: + raise NotImplementedError("unknown request method: {}".format( + request_method)) + + response.raise_for_status() + return response + + def get(self, api_method, params=None): + """ + Perform a GET request for the given API method, and return the response. + """ + return self._request('GET', api_method, params=params) + + def get_people(self, page_size=100, progress=None, **kwargs): + """ + Retrieve all Person records. + + https://apiexplorer.nationbuilder.com/nationbuilder#People + """ + response = self.get('/api/v1/people/count') + count = response.json()['people_count'] + pages = count // page_size + if count % page_size: + pages += 1 + + url = {'next': '/api/v1/people?limit={}'.format(page_size)} + people = [] + + def fetch(page, i): + response = self.get(url['next']) + data = response.json() + people.extend(data['results']) + url['next'] = data['next'] + + self.app.progress_loop(fetch, range(pages), progress, + message="Fetching Person data from NationBuilder") + return people + + def get_people_with_tag(self, tag, page_size=100, **kwargs): + """ + Retrieve all Person records with the given tag. + + https://apiexplorer.nationbuilder.com/nationbuilder#People + """ + people = [] + + # get first page + api_method = '/api/v1/tags/{}/people'.format(tag) + api_method = '{}?limit={}'.format(api_method, page_size) + response = self.get(api_method) + data = response.json() + people.extend(data['results']) + + # get more pages, until complete + while data['next']: + response = self.get(data['next']) + data = response.json() + people.extend(data['results']) + + return people