feat: add Nation Builder Connector

docs/index.rst

@@ -205,6 +205,7 @@ Indices and tables
    hustle
    mailchimp
    mobilize_america
+   nation_builder
    newmode
    ngpvan
    p2a

docs/nation_builder.rst

@@ -0,0 +1,52 @@
+NationBuilder
+==============
+
+********
+Overview
+********
+
+The NationBuilder class allows you to interact with the NationBuilder API. Users of this Parsons integration can download a full list of people, update and upsert people.
+
+.. note::
+  Authentication
+    In order to use this class you need your nation slug and access token. To get your access token login to your nation and navigate to ``Settings > Developer > API Token`` and create a new token. You can get more info in the `NationBuilder API docs <https://nationbuilder.com/api_quickstart>`_.
+
+==========
+Quickstart
+==========
+
+To instantiate the NationBuilder class, you can either store your ``NB_SLUG`` and ``NB_ACCESS_TOKEN`` keys as environment
+variables or pass them in as arguments:
+
+.. code-block:: python
+
+   from parsons import NationBuilder
+
+   # First approach: Use API key environment variables
+
+   # In bash, set your environment variables like so:
+   # export NB_SLUG='my-nation-slug'
+   # export NB_ACCESS_TOKEN='MY_ACCESS_TOKEN'
+   nb = NationBuilder()
+
+   # Second approach: Pass API keys as arguments
+   nb = NationBuilder(slug='my-nation-slug', access_token='MY_ACCESS_TOKEN')
+
+You can then make a request to get all people and save its data to a Parsons table using the method, ``get_people()``:
+
+.. code-block:: python
+
+    # Create Parsons table with people data from API
+    parsons_table = nb.get_people()
+
+    # Save people as CSV
+    parsons_table.to_csv('people.csv')
+
+The above example shows how to create a Parsons table with all people registered in your NationBuilder nation.
+
+***
+API
+***
+
+.. autoclass :: parsons.NationBuilder
+   :inherited-members:

parsons/__init__.py

@@ -1,7 +1,7 @@
 # Provide shortcuts to importing Parsons submodules and set up logging
 import importlib
-import os
 import logging
+import os
 
 from parsons.etl.table import Table
 
@@ -63,6 +63,7 @@
     ("parsons.hustle.hustle", "Hustle"),
     ("parsons.mailchimp.mailchimp", "Mailchimp"),
     ("parsons.mobilize_america.ma", "MobilizeAmerica"),
+    ("parsons.nation_builder.nation_builder", "NationBuilder"),
     ("parsons.newmode.newmode", "Newmode"),
     ("parsons.ngpvan.van", "VAN"),
     ("parsons.notifications.gmail", "Gmail"),

parsons/nation_builder/__init__.py

@@ -0,0 +1,3 @@
+from parsons.nation_builder.nation_builder import NationBuilder
+
+__all__ = ["NationBuilder"]

parsons/nation_builder/nation_builder.py

@@ -0,0 +1,221 @@
+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__)
+
+
+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}.nationbuilder.com/api/v1"
+
+    @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}"
+
+    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
+                logging.info("waiting %d seconds before retrying" % wait_time)
+                time.sleep(wait_time)
+
+        return Table(data)
+
+    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": "[email protected]", "tags": ["foo", "bar"]}
+                Docs: https://nationbuilder.com/people_api
+        `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
+
+    def upsert_person(
+        self, person: Dict[str, Any]
+    ) -> Tuple[bool, dict[str, Any] | None]:
+        """
+        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": "[email protected]", "tags": ["foo", "bar"]}
+                Docs: https://nationbuilder.com/people_api
+        `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)