from Newmode import Client
from parsons.utilities.oauth_api_connector import OAuth2APIConnector
from parsons.utilities import check_env
from parsons.etl import Table
import logging
logger = logging.getLogger(__name__)
V2_API_URL = "https://base.newmode.net/api/"
V2_API_AUTH_URL = "https://base.newmode.net/oauth/token/"
V2_API_CAMPAIGNS_URL = "https://base.newmode.net/"
V2_API_CAMPAIGNS_VERSION = "jsonapi"
V2_API_CAMPAIGNS_HEADERS = {
"content-type": "application/vnd.api+json",
"accept": "application/vnd.api+json",
"authorization": "Bearer 1234567890",
}
class NewmodeV1:
def __init__(self, api_user=None, api_password=None, api_version=None):
"""
Args:
api_user: str
The Newmode api user. Not required if ``NEWMODE_API_USER`` env variable is
passed.
api_password: str
The Newmode api password. Not required if ``NEWMODE_API_PASSWORD`` env variable is
passed.
api_version: str
The Newmode api version. Defaults to "v1.0" or the value of ``NEWMODE_API_VERSION``
env variable.
Returns:
Newmode class
"""
logger.warning(
"Newmode V1 API will be sunset in Feburary 28th, 2025. To use V2, set api_version=v2.1"
)
self.api_user = check_env.check("NEWMODE_API_USER", api_user)
self.api_password = check_env.check("NEWMODE_API_PASSWORD", api_password)
self.api_version = api_version
self.client = Client(api_user, api_password, api_version)
def convert_to_table(self, data):
# Internal method to create a Parsons table from a data element.
table = None
if isinstance(data, list):
table = Table(data)
else:
table = Table([data])
return table
def get_tools(self, params={}):
"""
Get existing tools.
Args:
params:
Extra parameters sent to New/Mode library.
Returns:
Tools information as table.
"""
tools = self.client.getTools(params=params)
if tools:
return self.convert_to_table(tools)
else:
logging.warning("Empty tools returned")
return self.convert_to_table([])
def get_tool(self, tool_id, params={}):
"""
Get specific tool.
Args:
tool_id:
The id of the tool to return.
params:
Extra parameters sent to New/Mode library.
Returns:
Tool information.
"""
tool = self.client.getTool(tool_id, params=params)
if tool:
return tool
else:
logging.warning("Empty tool returned")
return None
def lookup_targets(self, tool_id, search=None, params={}):
"""
Lookup targets for a given tool
Args:
tool_id:
The tool to lookup targets.
search:
The search criteria. It could be:
- Empty: If empty, return custom targets associated to the tool.
- Postal code: Return targets matched by postal code.
- Lat/Long: Latitude and Longitude pair separated by '::'.
Ex. 45.451596::-73.59912099999997. It will return targets
matched for those coordinates.
- Search term: For your csv tools, this will return targets
matched by given valid search term.
Returns:
Targets information as table.
"""
targets = self.client.lookupTargets(tool_id, search, params=params)
if targets:
data = []
for key in targets:
if key != "_links":
data.append(targets[key])
return self.convert_to_table(data)
else:
logging.warning("Empty targets returned")
return self.convert_to_table([])
def get_action(self, tool_id, params={}):
"""
Get the action information for a given tool.
Args:
tool_id:
The id of the tool to return.
params:
Extra parameters sent to New/Mode library.
Returns:
Tool action information.
"""
action = self.client.getAction(tool_id, params=params)
if action:
return action
else:
logging.warning("Empty action returned")
return None
def run_action(self, tool_id, payload, params={}):
"""
Run specific action with given payload.
Args:
tool_id:
The id of the tool to run.
payload:
Payload data used to run the action. Structure will depend
on the stuff returned by get_action.
params:
Extra parameters sent to New/Mode library.
Returns:
Action link (if otl) or sid.
"""
action = self.client.runAction(tool_id, payload, params=params)
if action:
if "link" in action:
return action["link"]
else:
return action["sid"]
else:
logging.warning("Error in response")
return None
def get_target(self, target_id, params={}):
"""
Get specific target.
Args:
target_id:
The id of the target to return.
params:
Extra parameters sent to New/Mode library.
Returns:
Target information.
"""
target = self.client.getTarget(target_id, params=params)
if target:
return target
else:
logging.warning("Empty target returned")
return None
def get_targets(self, params={}):
"""
Get all targets
Args:
params dict:
Extra paramaters sent to New/Mode library
Returns:
Target information
"""
targets = self.client.getTargets(params=params)
if targets:
return self.convert_to_table(targets)
else:
logging.warning("No targets returned")
return None
def get_campaigns(self, params={}):
"""
Get existing campaigns.
Args:
params:
Extra parameters sent to New/Mode library.
Returns:
Campaigns information as table.
"""
campaigns = self.client.getCampaigns(params=params)
if campaigns:
return self.convert_to_table(campaigns)
else:
logging.warning("Empty campaigns returned")
return self.convert_to_table([])
def get_campaign(self, campaign_id, params={}):
"""
Get specific campaign.
Args:
campaign_id:
The id of the campaign to return.
params:
Extra parameters sent to New/Mode library.
Returns:
Campaign information.
"""
campaign = self.client.getCampaign(campaign_id, params=params)
if campaign:
return campaign
else:
logging.warning("Empty campaign returned")
return None
def get_organizations(self, params={}):
"""
Get existing organizations.
Args:
params:
Extra parameters sent to New/Mode library.
Returns:
Organizations information as table.
"""
organizations = self.client.getOrganizations(params=params)
if organizations:
return self.convert_to_table(organizations)
else:
logging.warning("Empty organizations returned")
return self.convert_to_table([])
def get_organization(self, organization_id, params={}):
"""
Get specific organization.
Args:
organization_id:
The id of the organization to return.
params:
Extra parameters sent to New/Mode library.
Returns:
Organization information.
"""
organization = self.client.getOrganization(organization_id, params=params)
if organization:
return organization
else:
logging.warning("Empty organization returned")
return None
def get_services(self, params={}):
"""
Get existing services.
Args:
params:
Extra parameters sent to New/Mode library.
Returns:
Services information as table.
"""
services = self.client.getServices(params=params)
if services:
return self.convert_to_table(services)
else:
logging.warning("Empty services returned")
return self.convert_to_table([])
def get_service(self, service_id, params={}):
"""
Get specific service.
Args:
service_id:
The id of the service to return.
params:
Extra parameters sent to New/Mode library.
Returns:
Service information.
"""
service = self.client.getService(service_id, params=params)
if service:
return service
else:
logging.warning("Empty service returned")
return None
def get_outreaches(self, tool_id, params={}):
"""
Get existing outreaches for a given tool.
Args:
tool_id:
Tool to return outreaches.
params:
Extra parameters sent to New/Mode library.
Returns:
Outreaches information as table.
"""
outreaches = self.client.getOutreaches(tool_id, params=params)
if outreaches:
return self.convert_to_table(outreaches)
else:
logging.warning("Empty outreaches returned")
return self.convert_to_table([])
def get_outreach(self, outreach_id, params={}):
"""
Get specific outreach.
Args:
outreach_id:
The id of the outreach to return.
params:
Extra parameters sent to New/Mode library.
Returns:
Outreach information.
"""
outreach = self.client.getOutreach(outreach_id, params=params)
if outreach:
return outreach
else:
logging.warning("Empty outreach returned")
return None
class NewmodeV2:
# TODO: Add param definition and requirements once official Newmode docs are published
def __init__(
self,
client_id=None,
client_secret=None,
api_version="v2.1",
):
"""
Instantiate Class
`Args`:
client_id: str
The client id to use for the API requests. Not required if ``NEWMODE_API_CLIENT_ID``
env variable set.
client_secret: str
The client secret to use for the API requests. Not required if ``NEWMODE_API_CLIENT_SECRET``
env variable set.
api_version: str
The api version to use. Defaults to v2.1
Returns:
NewMode Class
"""
self.api_version = api_version
self.base_url = V2_API_URL
self.client_id = check_env.check("NEWMODE_API_CLIENT_ID", client_id)
self.client_secret = check_env.check("NEWMODE_API_CLIENT_SECRET", client_secret)
self.headers = {"content-type": "application/json"}
self.default_client = OAuth2APIConnector(
uri=self.base_url,
auto_refresh_url=V2_API_AUTH_URL,
client_id=self.client_id,
client_secret=self.client_secret,
headers=self.headers,
token_url=V2_API_AUTH_URL,
grant_type="client_credentials",
)
def base_request(
self,
method,
endpoint,
client,
data=None,
json=None,
data_key=None,
params={},
supports_version=True,
override_api_version=None,
):
"""
Internal method to instantiate OAuth2APIConnector class,
make a call to Newmode API, and validate the response.
"""
api_version = override_api_version if override_api_version else self.api_version
url = f"{api_version}/{endpoint}" if supports_version else endpoint
response = client.request(url=url, req_type=method, json=json, data=data, params=params)
response.raise_for_status()
success_codes = [200, 201, 202, 204]
client.validate_response(response)
if response.status_code in success_codes:
response_json = response.json() if client.json_check(response) else None
return response_json[data_key] if data_key and response_json else response_json
raise Exception(f"API request encountered an error. Response: {response}")
def converted_request(
self,
endpoint,
method,
supports_version=True,
data=None,
json=None,
params={},
convert_to_table=True,
data_key=None,
client=None,
override_api_version=None,
):
"""Internal method to make a call to the Newmode API and convert the result to a Parsons table."""
client = client if client else self.default_client
response = self.base_request(
method=method,
json=json,
data=data,
params=params,
data_key=data_key,
supports_version=supports_version,
endpoint=endpoint,
client=client,
override_api_version=override_api_version,
)
if convert_to_table:
return client.convert_to_table(data=response)
else:
return response
def get_campaign(self, campaign_id, params={}):
"""
Retrieve a specific campaign by ID.
In v2, a campaign is equivalent to Tools or Actions in V1.
`Args:`
campaign_id: str
The ID of the campaign to retrieve.
params: dict
Query parameters to include in the request.
`Returns:`
Parsons Table containing campaign data.
"""
endpoint = f"/campaign/{campaign_id}/form"
data = self.converted_request(
endpoint=endpoint,
method="GET",
params=params,
)
return data
def get_campaign_ids(self, params={}):
"""
Retrieve all campaigns
In v2, a campaign is equivalent to Tools or Actions in V1.
`Args:`
organization_id: str
ID of organization
params: dict
Query parameters to include in the request.
`Returns:`
List containing all campaign ids.
"""
endpoint = "node/action"
campaigns_client = OAuth2APIConnector(
uri=V2_API_CAMPAIGNS_URL,
auto_refresh_url=V2_API_AUTH_URL,
client_id=self.client_id,
client_secret=self.client_secret,
headers=V2_API_CAMPAIGNS_HEADERS,
token_url=V2_API_AUTH_URL,
grant_type="client_credentials",
)
data = self.converted_request(
endpoint=endpoint,
method="GET",
params=params,
data_key="data",
client=campaigns_client,
override_api_version=V2_API_CAMPAIGNS_VERSION,
)
return data["id"]
def get_recipient(
self,
campaign_id,
street_address=None,
city=None,
postal_code=None,
region=None,
params={},
):
"""
Retrieve a specific recipient by ID
`Args:`
campaign_id: str
The ID of the campaign to retrieve.
street_address: str
Street address of recipient
city: str
City of recipient
postal_code: str
Postal code of recipient
region: str
Region (i.e. state/province abbreviation) of recipient
params: dict
Query parameters to include in the request.
`Returns:`
Parsons Table containing recipient data.
"""
address_params = {
"street_address": street_address,
"city": city,
"postal_code": postal_code,
"region": region,
}
all_address_params_are_missing = all(x is None for x in address_params.values())
if all_address_params_are_missing:
raise ValueError(
"Incomplete Request. Please specify a street address, city, postal code, and/or region."
)
params = {f"address[value][{key}]": value for key, value in address_params.items() if value}
response = self.converted_request(
endpoint=f"campaign/{campaign_id}/target",
method="GET",
params=params,
)
return response
def run_submit(self, campaign_id, json=None, data=None, params={}):
"""
Pass a submission from a supporter to a campaign
that ultimately fills in a petition,
sends an email or triggers a phone call
depending on your campaign type
`Args:`
campaign_id: str
The ID of the campaign to retrieve.
params: dict
Query parameters to include in the request.
`Returns:`
Parsons Table containing submit data.
"""
response = self.converted_request(
endpoint=f"campaign/{campaign_id}/submit",
method="POST",
data=data,
json=json,
params=params,
convert_to_table=False,
)
return response
def get_submissions(self, campaign_id, params={}):
"""
Retrieve and sort submissions and contact data
for a specified campaign using a range of filters
that include campaign id, data range and submission status
`Args:`
params: dict
Query parameters to include in the request.
`Returns:`
Parsons Table containing submit data.
"""
params = {"action": campaign_id}
response = self.converted_request(endpoint="submission", method="GET", params=params)
return response
[docs]
class Newmode:
def __new__(
cls,
client_id=None,
client_secret=None,
api_user=None,
api_password=None,
api_version="v1.0",
):
"""
Create and return Newmode instance based on chosen version (V1 or V2)
`Args`:
api_user: str
The Newmode api user. Not required if ``NEWMODE_API_USER`` env variable is
passed. Needed for V1.
api_password: str
The Newmode api password. Not required if ``NEWMODE_API_PASSWORD`` env variable is
passed. Needed for V1.
client_id: str
The client id to use for the API requests. Not required if ``NEWMODE_API_CLIENT_ID``
env variable set. Needed for V2.
client_secret: str
The client secret to use for the API requests. Not required if ``NEWMODE_API_CLIENT_SECRET``
env variable set. Needed for V2.
api_version: str
The api version to use. Defaults to v1.0.
Returns:
NewMode Class
"""
api_version = check_env.check("NEWMODE_API_VERSION", api_version)
if api_version.startswith("v2"):
return NewmodeV2(
client_id=client_id, client_secret=client_secret, api_version=api_version
)
if api_version.startswith("v1"):
return NewmodeV1(api_user=api_user, api_password=api_password, api_version=api_version)
raise ValueError(f"{api_version} not supported.")