Source code for parsons.rockthevote.rtv

import datetime
import logging
import petl
import re
import requests
import time

from dateutil.parser import parse as parse_date

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

logger = logging.getLogger(__name__)


VALID_REPORT_TYPES = ["extended"]
TESTING_URI = "https://staging.rocky.rockthevote.com/api/v4"
PRODUCTION_URI = "https://register.rockthevote.com/api/v4"

DATETIME_FORMAT = "%Y-%m-%d %H:%M:%S UTC"
"""Datetime format for sending date's to the API."""

REQUEST_HEADERS = {
    # For some reason, RTV's firewall REALLY doesn't like the
    # user-agent string that Python's request library gives by default,
    # though it seems fine with the curl user agent
    # For more info on user agents, see:
    # https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/User-Agent
    "user-agent": "curl/7.54.0"
}
"""Standard request header for sending requests to the API."""

STATUS_URL_PARSE_REGEX = re.compile(r"(\d+)$")
"""Regex for parsing the report ID from the status URL."""


class RTVFailure(Exception):
    """Exception raised when there is an error with the connector."""


[docs]class RockTheVote: """ Instantiate the RockTheVote class `Args:` partner_id: str The RockTheVote partner ID for the RTV account. Not required if the ``RTV_PARTNER_ID`` environmental variable is set. partner_api_key: str The API Key for the partner. Not required if the ``RTV_PARTNER_API_KEY`` environmental variable is set. testing: bool Whether or not to use the staging instance. Defaults to False. `Returns`: RockTheVote class """ def __init__(self, partner_id=None, partner_api_key=None, testing=False): self.partner_id = check_env.check("RTV_PARTNER_ID", partner_id) self.partner_api_key = check_env.check("RTV_PARTNER_API_KEY", partner_api_key) if testing: self.client = APIConnector(TESTING_URI, headers=REQUEST_HEADERS) else: self.client = APIConnector(PRODUCTION_URI, headers=REQUEST_HEADERS)
[docs] def create_registration_report(self, before=None, since=None, report_type=None): """ Create a new registration report. `Args:` before: str Limit to registrations that were started before this date, in ISO format (e.g. 2020-01-01) since: str Limit to registrations that were started since this date, in ISO format (e.g. 2020-01-01) report_type: str The type of report to create. If left as None, it creates the default report. The ``extended`` report includes additional fields. Currently only accepts ``extended``. `Returns:` int The ID of the created report. """ report_url = "registrant_reports.json" # Create the report for the new data report_parameters = { "partner_id": self.partner_id, "partner_API_key": self.partner_api_key, } # Declare these here so the logging doesn't error out since_date = before_date = None if report_type: if report_type not in VALID_REPORT_TYPES: raise RTVFailure( f"Invalid report type. Must be one of {VALID_REPORT_TYPES}" ) report_parameters["report_type"] = report_type if since: since_date = parse_date(since).strftime(DATETIME_FORMAT) report_parameters["since"] = since_date if before: before_date = parse_date(before).strftime(DATETIME_FORMAT) report_parameters["before"] = before_date # The report parameters get passed into the request as JSON in the body # of the request. report_str = f"{report_type} report" if report_type else "report" logger.info( f"Creating {report_str} for {self.partner_id} " f"for dates: {since_date} to {before_date}..." ) response = self.client.request(report_url, "post", json=report_parameters) if response.status_code != requests.codes.ok: raise RTVFailure("Couldn't create RTV registrations report") response_json = response.json() # The RTV API says the response should include the report_id, but I have not found # that to be the case report_id = response_json.get("report_id") if report_id: logger.info(f"Created report with id {report_id}.") return report_id # If the response didn't include the report_id, then we will parse it out of the URL. status_url = response_json.get("status_url") url_match = STATUS_URL_PARSE_REGEX.search(status_url) if url_match: report_id = url_match.group(1) logger.info(f"Created report with id {report_id}.") return report_id
[docs] def get_registration_report( self, report_id, block=False, poll_interval_seconds=60, report_timeout_seconds=3600, ): """ Get data from an existing registration report. `Args:` report_id: int The ID of the report to get data from block: bool Whether or not to block execution until the report is complete poll_interval_seconds: int If blocking, how long to pause between attempts to check if the report is done report_timeout_seconds: int If blocking, how long to wait for the report before timing out `Returns:` Parsons Table Parsons table with the report data. """ logger.info(f"Getting report with id {report_id}...") credentials = { "partner_id": self.partner_id, "partner_API_key": self.partner_api_key, } status_url = f"registrant_reports/{report_id}" download_url = None # Let's figure out at what time should we just give up because we waited # too long end_time = datetime.datetime.now() + datetime.timedelta( seconds=report_timeout_seconds ) # If we have a download URL, we can move on and just download the # report. Otherwise, as long as we haven't run out of time, we will # check the status. while not download_url and datetime.datetime.now() < end_time: logger.debug( f"Registrations report not ready yet, sleeping {poll_interval_seconds} seconds" ) # Check the status again via the status endpoint status_response = self.client.request(status_url, "get", params=credentials) # Check to make sure the call got a valid response if status_response.status_code == requests.codes.ok: status_json = status_response.json() # Grab the download_url from the response. download_url = status_json.get("download_url") if not download_url and not block: return None else: raise RTVFailure("Couldn't get report status") if not download_url: # We just got the status, so we should wait a minute before # we check it again. time.sleep(poll_interval_seconds) # If we never got a valid download_url, then we timed out waiting for # the report to generate. We will log an error and exit. if not download_url: raise RTVFailure("Timed out waiting for report") # Download the report data download_response = self.client.request(download_url, "get", params=credentials) # Check to make sure the call got a valid response if download_response.status_code == requests.codes.ok: report_data = download_response.text # Load the report data into a Parsons Table table = Table.from_csv_string(report_data) # Transform the data from the report's CSV format to something more # Pythonic (snake case) normalized_column_names = [ re.sub(r"\s", "_", name).lower() for name in table.columns ] normalized_column_names = [ re.sub(r"[^A-Za-z\d_]", "", name) for name in normalized_column_names ] table.table = petl.setheader(table.table, normalized_column_names) return table else: raise RTVFailure("Unable to download report data")
[docs] def run_registration_report( self, before=None, since=None, report_type=None, poll_interval_seconds=60, report_timeout_seconds=3600, ): """ Run a new registration report. This method will block until the report has finished generating, or until the specified timeout is reached. `Args:` before: str Limit to registrations that were started before this date, in ISO format (e.g. 2020-01-01) since: str Limit to registrations that were started since this date, in ISO format (e.g. 2020-01-01) report_type: str The type of report to run. If left as None, it runs the default report. The ``extended`` report includes additional fields. Currently only accepts ``extended``. poll_interval_seconds: int If blocking, how long to pause between attempts to check if the report is done report_timeout_seconds: int If blocking, how long to wait for the report before timing out `Returns:` Parsons.Table The table with the report data. """ report_str = f"{report_type} report" if report_type else "report" logger.info( f"Running {report_str} for {self.partner_id} " f"for dates: {since} to {before}..." ) report_id = self.create_registration_report( before=before, since=since, report_type=report_type ) return self.get_registration_report( report_id, block=True, poll_interval_seconds=poll_interval_seconds, report_timeout_seconds=report_timeout_seconds, )
[docs] def get_state_requirements( self, lang, home_state_id, home_zip_code, date_of_birth=None, callback=None ): """ Checks state eligibility and provides state specific fields information. Args: lang: str Required. Language. Represented by an abbreviation. 'en', 'es', etc home_state_id: str Required. 2-character state abbreviation home_zip_code: str Required. 'zzzzz' 5 digit zip codes date_of_birth: str Optional. 'mm-dd-yyyy' callback: str Optional. If used, will change the return value from JSON format to jsonp Returns: Parsons.Table A single row table with the response json """ requirements_url = "state_requirements.json" logger.info(f"Getting the requirements for {home_state_id}...") params = { "lang": lang, "home_state_id": home_state_id, "home_zip_code": home_zip_code, } if date_of_birth: params["date_of_birth"] = date_of_birth if callback: params["callback"] = callback requirements_response = self.client.request( requirements_url, "get", params=params ) if requirements_response.status_code == requests.codes.ok: response_json = requirements_response.json() table = Table([response_json]) return table else: error_json = requirements_response.json() logger.info(f"{error_json}") raise RTVFailure("Could not retrieve state requirements")