Source code for parsons.nation_builder.nation_builder

import json
import logging
import time
from typing import Any, Dict, Optional, Tuple, cast
from urllib.parse import parse_qs, urlparse

from parsons import Table
from parsons.utilities import check_env
from parsons.utilities.api_connector import APIConnector

logger = logging.getLogger(__name__)

[docs]class NationBuilder: """ Instantiate the NationBuilder class `Args:` slug: str The Nation Builder slug Not required if ``NB_SLUG`` env variable set. The slug is the nation slug of the nation from which your application is requesting approval to retrieve data via the NationBuilder API. For example, your application's user could provide this slug via a text field in your application. access_token: str The Nation Builder access_token Not required if ``NB_ACCESS_TOKEN`` env variable set. """ def __init__(self, slug: Optional[str] = None, access_token: Optional[str] = None) -> None: slug = check_env.check("NB_SLUG", slug) token = check_env.check("NB_ACCESS_TOKEN", access_token) headers = {"Content-Type": "application/json", "Accept": "application/json"} headers.update(NationBuilder.get_auth_headers(token)) self.client = APIConnector(NationBuilder.get_uri(slug), headers=headers) @classmethod def get_uri(cls, slug: Optional[str]) -> str: if slug is None: raise ValueError("slug can't None") if not isinstance(slug, str): raise ValueError("slug must be an str") if len(slug.strip()) == 0: raise ValueError("slug can't be an empty str") return f"https://{slug}" @classmethod def get_auth_headers(cls, access_token: Optional[str]) -> Dict[str, str]: if access_token is None: raise ValueError("access_token can't None") if not isinstance(access_token, str): raise ValueError("access_token must be an str") if len(access_token.strip()) == 0: raise ValueError("access_token can't be an empty str") return {"authorization": f"Bearer {access_token}"} @classmethod def parse_next_params(cls, next_value: str) -> Tuple[str, str]: next_params = parse_qs(urlparse(next_value).query) if "__nonce" not in next_params: raise ValueError("__nonce param not found") if "__token" not in next_params: raise ValueError("__token param not found") nonce = next_params["__nonce"][0] token = next_params["__token"][0] return nonce, token @classmethod def make_next_url(cls, original_url: str, nonce: str, token: str) -> str: return f"{original_url}?limit=100&__nonce={nonce}&__token={token}"
[docs] def get_people(self) -> Table: """ `Returns:` A Table of all people stored in Nation Builder. """ data = [] original_url = "people" url = f"{original_url}" while True: try: logging.debug("sending request %s" % url) response = self.client.get_request(url) res = response.get("results", None) if res is None: break logging.debug("response got %s records" % len(res)) data.extend(res) if response.get("next", None): nonce, token = NationBuilder.parse_next_params(response["next"]) url = NationBuilder.make_next_url(original_url, nonce, token) else: break except Exception as error: logging.error("error requesting data from Nation Builder: %s" % error) wait_time = 30"waiting %d seconds before retrying" % wait_time) time.sleep(wait_time) return Table(data)
[docs] def update_person(self, person_id: str, person: Dict[str, Any]) -> Dict[str, Any]: """ This method updates a person with the provided id to have the provided data. It returns a full representation of the updated person. `Args:` person_id: str Nation Builder person id. data: dict Nation builder person object. For example {"email": "", "tags": ["foo", "bar"]} Docs: `Returns:` A person object with the updated data. """ if person_id is None: raise ValueError("person_id can't None") if not isinstance(person_id, str): raise ValueError("person_id must be a str") if len(person_id.strip()) == 0: raise ValueError("person_id can't be an empty str") if not isinstance(person, dict): raise ValueError("person must be a dict") url = f"people/{person_id}" response = self.client.put_request(url, data=json.dumps({"person": person})) response = cast(Dict[str, Any], response) return response
[docs] def upsert_person(self, person: Dict[str, Any]) -> Tuple[bool, Optional[Dict[str, Any]]]: """ Updates a matched person or creates a new one if the person doesn't exist. This method attempts to match the input person resource to a person already in the nation. If a match is found, the matched person is updated. If a match is not found, a new person is created. Matches are found by including one of the following IDs in the request: - civicrm_id - county_file_id - dw_id - external_id - email - facebook_username - ngp_id - salesforce_id - twitter_login - van_id `Args:` data: dict Nation builder person object. For example {"email": "", "tags": ["foo", "bar"]} Docs: `Returns:` A tuple of `created` and `person` object with the updated data. If the request fails the method will return a tuple of `False` and `None`. """ _required_keys = [ "civicrm_id", "county_file_id", "dw_id", "external_id", "email", "facebook_username", "ngp_id", "salesforce_id", "twitter_login", "van_id", ] if not isinstance(person, dict): raise ValueError("person must be a dict") has_required_key = any(x in person for x in _required_keys) if not has_required_key: _keys = ", ".join(_required_keys) raise ValueError(f"person dict must contain at least one key of {_keys}") url = "people/push" response = self.client.request(url, "PUT", data=json.dumps({"person": person})) self.client.validate_response(response) if response.status_code == 200: if self.client.json_check(response): return (False, response.json()) if response.status_code == 201: if self.client.json_check(response): return (True, response.json()) return (False, None)