Source code for parsons.action_network.action_network

import json
from parsons import Table
import re
from parsons.utilities import check_env
from parsons.utilities.api_connector import APIConnector
import logging

logger = logging.getLogger(__name__)

API_URL = "https://actionnetwork.org/api/v2"


[docs]class ActionNetwork(object): """ `Args:` api_token: str The OSDI API token """ def __init__(self, api_token=None): self.api_token = check_env.check("AN_API_TOKEN", api_token) self.headers = { "Content-Type": "application/json", "OSDI-API-Token": self.api_token, } self.api_url = API_URL self.api = APIConnector(self.api_url, headers=self.headers) def _get_page(self, object_name, page, per_page=25, filter=None): # returns data from one page of results if per_page > 25: per_page = 25 logger.info( "Action Network's API will not return more than 25 entries per page. \ Changing per_page parameter to 25." ) params = {"page": page, "per_page": per_page, "filter": filter} return self.api.get_request(url=object_name, params=params) def _get_entry_list(self, object_name, limit=None, per_page=25, filter=None): # returns a list of entries for a given object, such as people, tags, or actions # Filter can only be applied to people, petitions, events, forms, fundraising_pages, # event_campaigns, campaigns, advocacy_campaigns, signatures, attendances, submissions, # donations and outreaches. # See Action Network API docs for more info: https://actionnetwork.org/docs/v2/ count = 0 page = 1 return_list = [] while True: response = self._get_page(object_name, page, per_page, filter=filter) page = page + 1 response_list = response["_embedded"][f"osdi:{object_name}"] if not response_list: return Table(return_list) return_list.extend(response_list) count = count + len(response_list) if limit: if count >= limit: return Table(return_list[0:limit])
[docs] def get_people(self, limit=None, per_page=25, page=None, filter=None): """ `Args:` limit: The number of entries to return. When None, returns all entries. per_page The number of entries per page to return. 25 maximum. page Which page of results to return filter The OData query for filtering results. E.g. "modified_date gt '2014-03-25'". When None, no filter is applied. `Returns:` A list of JSONs of people stored in Action Network. """ if page: return self._get_page("people", page, per_page, filter=filter) return self._get_entry_list("people", limit, per_page, filter=filter)
[docs] def get_person(self, person_id): """ `Args:` person_id: Id of the person. `Returns:` A JSON of the entry. If the entry doesn't exist, Action Network returns ``{'error': 'Couldn't find person with id = <id>'}``. """ return self.api.get_request(url=f"people/{person_id}")
[docs] def upsert_person( self, email_address=None, given_name=None, family_name=None, tags=None, languages_spoken=None, postal_addresses=None, mobile_number=None, mobile_status="subscribed", **kwargs, ): """ Creates or updates a person record. In order to update an existing record instead of creating a new one, you must supply an email or mobile number which matches a record in the database. `Args:` email_address: Either email_address or mobile_number are required. Can be any of the following - a string with the person's email - a list of strings with a person's emails - a dictionary with the following fields - email_address (REQUIRED) - primary (OPTIONAL): Boolean indicating the user's primary email address - status (OPTIONAL): can taken on any of these values - "subscribed" - "unsubscribed" - "bouncing" - "previous bounce" - "spam complaint" - "previous spam complaint" given_name: The person's given name family_name: The person's family name tags: Optional field. A list of strings of pre-existing tags to be applied to the person. languages_spoken: Optional field. A list of strings of the languages spoken by the person postal_addresses: Optional field. A list of dictionaries. For details, see Action Network's documentation: https://actionnetwork.org/docs/v2/person_signup_helper mobile_number: Either email_address or mobile_number are required. Can be any of the following - a string with the person's cell phone number - an integer with the person's cell phone number - a list of strings with the person's cell phone numbers - a list of integers with the person's cell phone numbers - a dictionary with the following fields - number (REQUIRED) - primary (OPTIONAL): Boolean indicating the user's primary mobile number - status (OPTIONAL): can taken on any of these values - "subscribed" - "unsubscribed" mobile_status: 'subscribed' or 'unsubscribed' **kwargs: Any additional fields to store about the person. Action Network allows any custom field. Adds a person to Action Network """ email_addresses_field = None if type(email_address) == str: email_addresses_field = [{"address": email_address}] elif type(email_address) == list: if type(email_address[0]) == str: email_addresses_field = [{"address": email} for email in email_address] email_addresses_field[0]["primary"] = True if type(email_address[0]) == dict: email_addresses_field = email_address mobile_numbers_field = None if type(mobile_number) == str: mobile_numbers_field = [ {"number": re.sub("[^0-9]", "", mobile_number), "status": mobile_status} ] elif type(mobile_number) == int: mobile_numbers_field = [ {"number": str(mobile_number), "status": mobile_status} ] elif type(mobile_number) == list: if len(mobile_number) > 1: raise ("Action Network allows only 1 phone number per activist") if type(mobile_number[0]) == str: mobile_numbers_field = [ {"number": re.sub("[^0-9]", "", cell), "status": mobile_status} for cell in mobile_number ] mobile_numbers_field[0]["primary"] = True if type(mobile_number[0]) == int: mobile_numbers_field = [ {"number": cell, "status": mobile_status} for cell in mobile_number ] mobile_numbers_field[0]["primary"] = True if type(mobile_number[0]) == dict: mobile_numbers_field = mobile_number if not email_addresses_field and not mobile_numbers_field: raise ( "Either email_address or mobile_number is required and can be formatted " "as a string, list of strings, a dictionary, a list of dictionaries, or " "(for mobile_number only) an integer or list of integers" ) data = {"person": {}} if email_addresses_field is not None: data["person"]["email_addresses"] = email_addresses_field if mobile_numbers_field is not None: data["person"]["phone_numbers"] = mobile_numbers_field if given_name is not None: data["person"]["given_name"] = given_name if family_name is not None: data["person"]["family_name"] = family_name if languages_spoken is not None: data["person"]["languages_spoken"] = languages_spoken if postal_addresses is not None: data["person"]["postal_addresses"] = postal_addresses if tags is not None: data["add_tags"] = tags data["person"]["custom_fields"] = {**kwargs} response = self.api.post_request( url=f"{self.api_url}/people", data=json.dumps(data) ) identifiers = response["identifiers"] person_id = [ entry_id.split(":")[1] for entry_id in identifiers if "action_network:" in entry_id ][0] if response["created_date"] == response["modified_date"]: logger.info(f"Entry {person_id} successfully added.") else: logger.info(f"Entry {person_id} successfully updated.") return response
[docs] def add_person( self, email_address=None, given_name=None, family_name=None, tags=None, languages_spoken=None, postal_addresses=None, mobile_number=None, mobile_status="subscribed", **kwargs, ): """ Creates a person in the database. WARNING: this endpoint has been deprecated in favor of upsert_person. """ logger.warning( "Method 'add_person' has been deprecated. Please use 'upsert_person'." ) # Pass inputs to preferred method: self.upsert_person( email_address=email_address, given_name=given_name, family_name=family_name, languages_spoken=languages_spoken, postal_addresses=postal_addresses, mobile_number=mobile_number, mobile_status=mobile_status, **kwargs, )
[docs] def update_person(self, entry_id, **kwargs): """ Updates a person's data in Action Network, given their Action Network ID. Note that you can't alter a person's tags with this method. Instead, use upsert_person. `Args:` entry_id: The person's Action Network id **kwargs: Fields to be updated. The possible fields are email_address: Can be any of the following - a string with the person's email - a dictionary with the following fields - email_address (REQUIRED) - primary (OPTIONAL): Boolean indicating the user's primary email address - status (OPTIONAL): can taken on any of these values - "subscribed" - "unsubscribed" - "bouncing" - "previous bounce" - "spam complaint" - "previous spam complaint" given_name: The person's given name family_name: The person's family name languages_spoken: Optional field. A list of strings of the languages spoken by the person postal_addresses: Optional field. A list of dictionaries. For details, see Action Network's documentation: https://actionnetwork.org/docs/v2/people#put custom_fields: A dictionary of any other fields to store about the person. """ data = {**kwargs} response = self.api.put_request( url=f"{self.api_url}/people/{entry_id}", data=json.dumps(data), success_codes=[204, 201, 200], ) logger.info(f"Person {entry_id} successfully updated") return response
[docs] def get_tags(self, limit=None, per_page=25, page=None): """ `Args:` limit: The number of entries to return. When None, returns all entries. per_page The number of entries per page to return. 25 maximum. page Which page of results to return `Returns:` A list of JSONs of tags in Action Network. """ if page: self.get_page("tags", page, per_page) return self._get_entry_list("tags", limit, per_page)
[docs] def get_tag(self, tag_id): """ `Args:` tag_id: Id of the tag. `Returns:` A JSON of the entry. If the entry doesn't exist, Action Network returns "{'error': 'Couldn't find tag with id = <id>'}" """ return self.api.get_request(url=f"tags/{tag_id}")
[docs] def add_tag(self, name): """ `Args:` name: The tag's name. This is the ONLY editable field Adds a tag to Action Network. Once created, tags CANNOT be edited or deleted. """ data = {"name": name} response = self.api.post_request( url=f"{self.api_url}/tags", data=json.dumps(data) ) identifiers = response["identifiers"] person_id = [ entry_id.split(":")[1] for entry_id in identifiers if "action_network:" in entry_id ][0] logger.info(f"Tag {person_id} successfully added to tags.") return response
[docs] def create_event(self, title, start_date=None, location=None): """ Create an event in Action Network `Args:` title: str The public title of the event start_date: str OR datetime OPTIONAL: The starting date & time. If a string, use format "YYYY-MM-DD HH:MM:SS" (hint: the default format you get when you use `str()` on a datetime) location: dict OPTIONAL: A dict of location details. Can include any combination of the types of values in the following example: .. code-block:: python my_location = { "venue": "White House", "address_lines": [ "1600 Pennsylvania Ave" ], "locality": "Washington", "region": "DC", "postal_code": "20009", "country": "US" } `Returns:` Dict of Action Network Event data. """ data = {"title": title} if start_date: start_date = str(start_date) data["start_date"] = start_date if isinstance(location, dict): data["location"] = location event_dict = self.api.post_request( url=f"{self.api_url}/events", data=json.dumps(data) ) an_event_id = event_dict["_links"]["self"]["href"].split("/")[-1] event_dict["event_id"] = an_event_id return event_dict