Source code for parsons.phone2action.p2a

import requests
from requests.auth import HTTPBasicAuth
from parsons.etl import Table
from parsons.utilities import check_env
import logging

logger = logging.getLogger(__name__)


[docs]class Phone2Action(object): """ Instantiate Phone2Action Class `Args:` app_id: str The Phone2Action provided application id. Not required if ``PHONE2ACTION_APP_ID`` env variable set. app_key: str The Phone2Action provided application key. Not required if ``PHONE2ACTION_APP_KEY`` env variable set. `Returns:` Phone2Action Class """ def __init__(self, app_id=None, app_key=None): self.app_id = check_env.check('PHONE2ACTION_APP_ID', app_id) self.app_key = check_env.check('PHONE2ACTION_APP_KEY', app_key) self.auth = HTTPBasicAuth(self.app_id, self.app_key) self.uri = PHONE2ACTION_URI def _request(self, url, args=None): # Internal request method r = requests.get(url, auth=self.auth, params=args) r.raise_for_status() return r def _paginate_request(self, url, args=None, page=None): # Internal pagination method if page is not None: args['page'] = page r = self._request(url, args=args) json = r.json()['data'] if page is not None: return json # If count of items is less than the total allowed per page, paginate while r.json()['pagination']['count'] == r.json()['pagination']['per_page']: r = self._request(r.json()['pagination']['next_url'], args) json.extend(r.json()['data']) return json
[docs] def get_advocates(self, state=None, campaign_id=None, updated_since=None, page=None): """ Return advocates (person records). If no page is specified, the method will automatically paginate through the available advocates. `Args:` state: str Filter by US postal abbreviation for a state or territory e.g., "CA" "NY" or "DC" campaign_id: int Filter to specific campaign updated_since: str Fetch all advocates updated since UTC date time provided using (ex. '2014-01-05 23:59:43') page: int Page number of data to fetch; if this is specified, call will only return one page. `Returns:` A dict of parsons tables: * emails * phones * memberships * tags * ids * fields * advocates """ url = self.uri + 'advocates' args = {'state': state, 'campaignid': campaign_id, 'updatedSince': updated_since}'Retrieving advocates...') json = self._paginate_request(url, args=args, page=page) return self._advocates_tables(Table(json))
def _advocates_tables(self, tbl): # Convert the advocates nested table into multiple tables tbls = { 'advocates': tbl, 'emails': Table(), 'phones': Table(), 'memberships': Table(), 'tags': Table(), 'ids': Table(), 'fields': Table(), } if not tbl: return tbls'Retrieved {tbl.num_rows} advocates...') # Unpack all of the single objects # The Phone2Action API docs says that created_at and updated_at are dictionaries, but # the data returned from the server is a ISO8601 timestamp. - EHS, 05/21/2020 for c in ['address', 'districts']: tbl.unpack_dict(c) # Unpack all of the arrays child_tables = [child for child in tbls.keys() if child != 'advocates'] for c in child_tables: tbls[c] = tbl.long_table(['id'], c, key_rename={'id': 'advocate_id'}) return tbls
[docs] def get_campaigns(self, state=None, zip=None, include_generic=False, include_private=False, include_content=True): """ Returns a list of campaigns `Args:` state: str Filter by US postal abbreviation for a state or territory e.g., "CA" "NY" or "DC" zip: int Filter by 5 digit zip code include_generic: boolean When filtering by state or ZIP code, include unrestricted campaigns include_private: boolean If true, will include private campaigns in results include_content: boolean If true, include campaign content fields, which may vary. This may cause sync errors. `Returns:` Parsons Table See :ref:`parsons-table` for output options. """ url = self.uri + 'campaigns' args = {'state': state, 'zip': zip, 'includeGeneric': str(include_generic), 'includePrivate': str(include_private) } tbl = Table(self._request(url, args=args).json()) tbl.unpack_dict('updated_at') if include_content: tbl.unpack_dict('content') return tbl