import logging
from parsons.etl import Table
from parsons.utilities import check_env
from parsons.utilities.api_connector import APIConnector
from parsons.utilities.datetime import convert_unix_to_readable
logger = logging.getLogger(__name__)
EMPOWER_API_ENDPOINT = "https://api.getempower.com/v1/export"
[docs]
class Empower(object):
"""
Instantiate class.
`Args:`
api_key: str
The Empower provided API key.The Empower provided Client UUID. Not
required if ``EMPOWER_API_KEY`` env variable set.
empower_uri: str
The URI to access the Empower API. The default is currently set to
https://api.getempower.com/v1/export. You can set an ``EMPOWER_URI`` env
variable or use this URI parameter if a different endpoint is necessary.
cache: boolean
The Empower API returns all account data after each call. Setting cache
to ``True`` stores the blob and then extracts Parsons tables for each method.
Setting cache to ``False`` will download all account data for each method call.
"""
def __init__(self, api_key=None, empower_uri=None, cache=True):
self.api_key = check_env.check("EMPOWER_API_KEY", api_key)
self.empower_uri = (
check_env.check("EMPOWER_URI", empower_uri, optional=True) or EMPOWER_API_ENDPOINT
)
self.headers = {"accept": "application/json", "secret-token": self.api_key}
self.client = APIConnector(
self.empower_uri,
headers=self.headers,
)
self.data = None
self.data = self._get_data(cache)
def _get_data(self, cache):
"""
Gets fresh data from Empower API based on cache setting.
"""
if not cache or self.data is None:
r = self.client.get_request(self.empower_uri)
logger.info("Empower data downloaded.")
return r
else:
return self.data
def _empty_obj(self, obj_name):
"""
Determine if a dict object is empty.
"""
if len(self.data[obj_name]) == 0:
return True
else:
return False
[docs]
def get_profiles(self):
"""
Get Empower profiles.
`Returns:`
Parsons Table
See :ref:`parsons-table` for output options.
"""
tbl = Table(self.data["profiles"])
for col in ["createdMts", "lastUsedEmpowerMts", "updatedMts"]:
tbl.convert_column(col, lambda x: convert_unix_to_readable(x))
tbl.remove_column("activeCtaIds") # Get as a method via get_profiles_active_ctas
return tbl
[docs]
def get_profiles_active_ctas(self):
"""
Get active ctas assigned to Empower profiles.
`Returns:`
Parsons Table
See :ref:`parsons-table` for output options.
"""
tbl = Table(self.data["profiles"]).long_table("eid", "activeCtaIds")
return tbl
[docs]
def get_regions(self):
"""
Get Empower regions.
`Returns:`
Parsons Table
See :ref:`parsons-table` for output options.
"""
tbl = Table(self.data["regions"])
tbl.convert_column("inviteCodeCreatedMts", lambda x: convert_unix_to_readable(x))
return tbl
[docs]
def get_cta_results(self):
"""
Get Empower call to action results.
`Returns:`
Parsons Table
See :ref:`parsons-table` for output options.
"""
# unpacks answerIdsByPromptId into standalone rows
tbl = Table(self.data["ctaResults"])
tbl.convert_column("contactedMts", lambda x: convert_unix_to_readable(x))
tbl = tbl.unpack_nested_columns_as_rows(
"answerIdsByPromptId", key="profileEid", expand_original=True
)
tbl.unpack_list("answerIdsByPromptId_value", replace=True)
col_list = [v for v in tbl.columns if v.find("value") != -1]
tbl.coalesce_columns("answer_id", col_list, remove_source_columns=True)
tbl.remove_column("uid")
tbl.remove_column("answers") # Per docs, this is deprecated.
return tbl
def _split_ctas(self):
"""
Internal method to split CTA objects into tables.
"""
ctas = Table(self.data["ctas"])
for col in [
"createdMts",
"scheduledLaunchTimeMts",
"updatedMts",
"activeUntilMts",
]:
ctas.convert_column(col, lambda x: convert_unix_to_readable(x))
# Get following data as their own tables via their own methods
ctas.remove_column("regionIds") # get_cta_regions()
ctas.remove_column("shareables") # get_cta_shareables()
ctas.remove_column("prioritizations") # get_cta_prioritizations()
ctas.remove_column("questions") # This column has been deprecated.
cta_prompts = ctas.long_table("id", "prompts", prepend=False, retain_original=False)
cta_prompts.remove_column("ctaId")
cta_prompt_answers = cta_prompts.long_table("id", "answers", prepend=False)
return {
"ctas": ctas,
"cta_prompts": cta_prompts,
"cta_prompt_answers": cta_prompt_answers,
}
[docs]
def get_ctas(self):
"""
Get Empower calls to action.
`Returns:`
Parsons Table
See :ref:`parsons-table` for output options.
"""
return self._split_ctas()["ctas"]
[docs]
def get_cta_prompts(self):
"""
Get Empower calls to action prompts.
`Returns:`
Parsons Table
See :ref:`parsons-table` for output options.
"""
return self._split_ctas()["cta_prompts"]
[docs]
def get_cta_prompt_answers(self):
"""
Get Empower calls to action prompt answers.
`Returns:`
Parsons Table
See :ref:`parsons-table` for output options.
"""
return self._split_ctas()["cta_prompt_answers"]
[docs]
def get_cta_regions(self):
"""
Get a list of regions that each call to active is active in.
`Returns:`
Parsons Table
See :ref:`parsons-table` for output options.
"""
tbl = Table(self.data["ctas"]).long_table("id", "regionIds")
return tbl
[docs]
def get_cta_shareables(self):
"""
Get a list of shareables associated with calls to action.
`Returns:`
Parsons Table
See :ref:`parsons-table` for output options.
"""
tbl = Table(self.data["ctas"]).long_table("id", "shareables")
return tbl
[docs]
def get_cta_prioritizations(self):
"""
Get a list prioritizations associated with calls to action.
`Returns:`
Parsons Table
See :ref:`parsons-table` for output options.
"""
tbl = Table(self.data["ctas"]).long_table("id", "prioritizations")
return tbl
[docs]
def get_outreach_entries(self):
"""
Get outreach entries.
`Returns:`
Parsons Table
See :ref:`parsons-table` for output options.
"""
if self._empty_obj("outreachEntries"):
logger.info("No Outreach Entries found.")
return Table([])
tbl = Table(self.data["outreachEntries"])
for col in [
"outreachCreatedMts",
"outreachSnoozeUntilMts",
"outreachScheduledFollowUpMts",
]:
tbl.convert_column(col, lambda x: convert_unix_to_readable(x))
return tbl
[docs]
def get_full_export(self):
"""
Get a table of the complete, raw data as returned by the API.
Meant to facilitate pure ELT pipelines
`Returns:`
Parsons Table
See :ref:`parsons-table` for output options.
"""
tbl = Table([self.data])
return tbl