import json
import logging
import re
import warnings
from parsons import Table
from parsons.utilities import check_env
from parsons.utilities.api_connector import APIConnector
logger = logging.getLogger(__name__)
API_URL = "https://actionnetwork.org/api/v2"
[docs]class ActionNetwork(object):
"""
`Args:`
api_token: str
The OSDI API token
"""
def __init__(self, api_token=None):
self.api_token = check_env.check("AN_API_TOKEN", api_token)
self.headers = {
"Content-Type": "application/json",
"OSDI-API-Token": self.api_token,
}
self.api_url = API_URL
self.api = APIConnector(self.api_url, headers=self.headers)
def _get_page(self, object_name, page, per_page=25, filter=None):
# returns data from one page of results
if per_page > 25:
per_page = 25
logger.info(
"Action Network's API will not return more than 25 entries per page. \
Changing per_page parameter to 25."
)
params = {"page": page, "per_page": per_page, "filter": filter}
return self.api.get_request(url=object_name, params=params)
def _get_entry_list(self, object_name, limit=None, per_page=25, filter=None):
# returns a list of entries for a given object, such as people, tags, or actions
# Filter can only be applied to people, petitions, events, forms, fundraising_pages,
# event_campaigns, campaigns, advocacy_campaigns, signatures, attendances, submissions,
# donations and outreaches.
# See Action Network API docs for more info: https://actionnetwork.org/docs/v2/
count = 0
page = 1
return_list = []
while True:
response = self._get_page(object_name, page, per_page, filter=filter)
page = page + 1
response_list = response["_embedded"][list(response["_embedded"])[0]]
if not response_list:
return Table(return_list)
return_list.extend(response_list)
count = count + len(response_list)
if limit:
if count >= limit:
return Table(return_list[0:limit])
# Advocacy Campaigns
[docs] def get_advocacy_campaigns(self, limit=None, per_page=25, page=None, filter=None):
"""
`Args:`
limit:
The number of entries to return. When None, returns all entries.
per_page:
The number of entries per page to return. 25 maximum.
page:
Which page of results to return
filter:
The OData query for filtering results. E.g. "modified_date gt '2014-03-25'".
When None, no filter is applied.
`Returns:`
A JSON with all of the advocacy_campaigns (letters) entries
"""
if page:
return self._get_page("advocacy_campaigns", page, per_page, filter)
return self._get_entry_list("advocacy_campaigns", limit, per_page, filter)
[docs] def get_advocacy_campaign(self, advocacy_campaign_id):
"""
`Args:`
advocacy_campaign_id:
The unique id of the advocacy_campaign
`Returns:`
A JSON with advocacy_campaign entry
"""
return self.api.get_request(f"advocacy_campaigns/{advocacy_campaign_id}")
# Attendances
[docs] def get_person_attendances(
self, person_id, limit=None, per_page=25, page=None, filter=None
):
"""
`Args:`
person_id:
The unique id of the person
limit:
The number of entries to return. When None, returns all entries.
per_page:
The number of entries per page to return. 25 maximum.
page:
Which page of results to return
filter:
The OData query for filtering results. E.g. "modified_date gt '2014-03-25'".
When None, no filter is applied.
`Returns:`
A JSON with all the attendances entries
"""
if page:
return self._get_page(
f"people/{person_id}/attendances", page, per_page, filter
)
return self._get_entry_list(
f"people/{person_id}/attendances", limit, per_page, filter
)
[docs] def get_event_attendances(
self, event_id, limit=None, per_page=25, page=None, filter=None
):
"""
`Args:`
event_id: the unique id of the event
limit:
The number of entries to return. When None, returns all entries.
per_page
The number of entries per page to return. 25 maximum.
page
Which page of results to return
filter
The OData query for filtering results. E.g. "modified_date gt '2014-03-25'".
When None, no filter is applied.
`Returns:`
A JSON with the attendances entries related to the event
"""
if page:
return self._get_page(
f"events/{event_id}/attendances", page, per_page, filter
)
return self._get_entry_list(
f"events/{event_id}/attendances", limit, per_page, filter
)
[docs] def get_event_attendance(self, event_id, attendance_id):
"""
`Args:`
event_id:
The unique id of the event
attendance_id:
The unique id of the attendance
`Returns:`
A JSON with the attendance entry
"""
return self.api.get_request(f"events/{event_id}/attendances/{attendance_id}")
[docs] def get_person_attendance(self, person_id, attendance_id):
"""
`Args:`
person_id:
The unique id of the person
attendance_id:
The unique id of the attendance
`Returns:`
A JSON with the attendance entry
"""
return self.api.get_request(f"people/{person_id}/attendances/{attendance_id}")
# Campaigns
[docs] def get_campaigns(self, limit=None, per_page=25, page=None, filter=None):
"""
`Args:`
limit:
The number of entries to return. When None, returns all entries.
per_page:
The number of entries per page to return. 25 maximum.
page:
Which page of results to return
filter:
The OData query for filtering results. E.g. "modified_date gt '2014-03-25'".
When None, no filter is applied.
`Returns:`
A JSON with all of the campaigns entries
"""
if page:
return self._get_page("campaigns", page, per_page, filter)
return self._get_entry_list("campaigns", limit, per_page, filter)
[docs] def get_campaign(self, campaign_id):
"""
`Args:`
campaign_id:
The unique id of the campaign
`Returns:`
A JSON with the campaign entry
"""
return self.api.get_request(f"campaigns/{campaign_id}")
# Custom Fields
[docs] def get_custom_fields(self):
"""
`Args:`
None
`Returns:`
A JSON with the custom_fields associated with your API key.
"""
return self.api.get_request("metadata/custom_fields")
# Donations
[docs] def get_donation(self, donation_id):
"""
`Args:`
donation_id: The unique id of the donation
`Returns:`
A JSON with donation data
"""
return self.api.get_request(url=f"donations/{donation_id}")
[docs] def get_donations(self, limit=None, per_page=25, page=None, filter=None):
"""
`Args:`
limit:
The number of entries to return. When None, returns all entries.
per_page:
The number of entries per page to return. 25 maximum.
page:
Which page of results to return
filter:
The OData query for filtering results. E.g. "modified_date gt '2014-03-25'".
When None, no filter is applied.
`Returns:`
A JSON with all the donations entries
"""
if page:
return self._get_page("donations", page, per_page, filter)
return self._get_entry_list("donations", limit, per_page, filter)
[docs] def get_fundraising_page_donations(
self, fundraising_page_id, limit=None, per_page=25, page=None, filter=None
):
"""
`Args:`
fundraising_page_id: The id of the fundraiser
limit:
The number of entries to return. When None, returns all entries.
per_page
The number of entries per page to return. 25 maximum.
page
Which page of results to return
filter
The OData query for filtering results. E.g. "modified_date gt '2014-03-25'".
When None, no filter is applied.
`Returns:`
A JSON with fundraising_page entry
"""
if page:
return self._get_page(
f"fundraising_pages/{fundraising_page_id}/donations",
page,
per_page,
filter,
)
return self._get_entry_list(
f"fundraising_pages/{fundraising_page_id}/donations",
limit,
per_page,
filter,
)
[docs] def get_person_donations(
self, person_id, limit=None, per_page=25, page=None, filter=None
):
"""
`Args:`
person_id: The id of the person
limit:
The number of entries to return. When None, returns all entries.
per_page
The number of entries per page to return. 25 maximum.
page
Which page of results to return
filter
The OData query for filtering results. E.g. "modified_date gt '2014-03-25'".
When None, no filter is applied.
`Returns:`
A JSON with all donations related to person
"""
if page:
return self._get_page(
f"people/{person_id}/donations",
page,
per_page,
filter,
)
return self._get_entry_list(
f"people/{person_id}/donations",
limit,
per_page,
filter,
)
# Embeds
[docs] def get_embeds(self, action_type, action_id):
"""
`Args:`
action_type:
The action type (petition, events, etc.)
action_id:
The unique id of the action
`Returns:`
A JSON with the embeds (for you to be able to embed action outside of ActionNetwork).
"""
return self.api.get_request(f"{action_type}/{action_id}/embed")
# Event Campaigns
[docs] def get_event_campaigns(self, limit=None, per_page=25, page=None, filter=None):
"""
`Args:`
limit:
The number of entries to return. When None, returns all entries.
per_page:
The number of entries per page to return. 25 maximum.
page:
Which page of results to return
filter:
The OData query for filtering results. E.g. "modified_date gt '2014-03-25'".
When None, no filter is applied.
`Returns:`
A JSON with all the event_campaigns entries
"""
if page:
return self._get_page("event_campaigns", page, per_page, filter)
return self._get_entry_list("event_campaigns", limit, per_page, filter)
[docs] def get_event_campaign(self, event_campaign_id):
"""
`Args:`
event_campaign_id:
The unique id of the event_campaign
`Returns:`
A JSON with event_campaign entry
"""
return self.api.get_request(f"event_campaigns/{event_campaign_id}")
# Events
[docs] def get_events(self, limit=None, per_page=25, page=None, filter=None):
"""
`Args:`
limit:
The number of entries to return. When None, returns all entries.
per_page
The number of entries per page to return. 25 maximum.
page
Which page of results to return
filter
The OData query for filtering results. E.g. "modified_date gt '2014-03-25'".
When None, no filter is applied.
`Returns:`
A JSON with all the events entries
"""
if page:
return self._get_page("events", page, per_page, filter)
return self._get_entry_list("events", limit, per_page, filter)
[docs] def get_event(self, event_id):
"""
`Args:`
event_id: the unique id of the event
`Returns:`
A JSON with event entry
"""
return self.api.get_request(f"events/{event_id}")
[docs] def get_event_campaign_events(
self, event_campaign_id, limit=None, per_page=25, page=None, filter=None
):
"""
`Args:`
event_campaign_id:
The unique id of the event_campaign
limit:
The number of entries to return. When None, returns all entries.
per_page
The number of entries per page to return. 25 maximum.
page
Which page of results to return
filter
The OData query for filtering results. E.g. "modified_date gt '2014-03-25'".
When None, no filter is applied.
`Returns:`
A JSON with all the eventes related to the event_campaign entry
"""
if page:
return self._get_page(
f"event_campaigns/{event_campaign_id}/events", page, per_page, filter
)
return self._get_entry_list(
f"event_campaigns/{event_campaign_id}/events", limit, per_page, filter
)
[docs] def create_event(self, title, start_date=None, location=None):
"""
Create an event in Action Network
`Args:`
title: str
The public title of the event
start_date: str OR datetime
OPTIONAL: The starting date & time. If a string, use format "YYYY-MM-DD HH:MM:SS"
(hint: the default format you get when you use `str()` on a datetime)
location: dict
OPTIONAL: A dict of location details. Can include any combination of the types of
values in the following example:
.. code-block:: python
my_location = {
"venue": "White House",
"address_lines": [
"1600 Pennsylvania Ave"
],
"locality": "Washington",
"region": "DC",
"postal_code": "20009",
"country": "US"
}
`Returns:`
Dict of Action Network Event data.
"""
data = {"title": title}
if start_date:
start_date = str(start_date)
data["start_date"] = start_date
if isinstance(location, dict):
data["location"] = location
event_dict = self.api.post_request(
url=f"{self.api_url}/events", data=json.dumps(data)
)
an_event_id = event_dict["_links"]["self"]["href"].split("/")[-1]
event_dict["event_id"] = an_event_id
return event_dict
# Forms
# Fundraising Pages
[docs] def get_fundraising_page(self, fundraising_page_id):
"""
`Args:`
fundraising_page_id: The id of the fundraiser
`Returns:`
A JSON with fundraising_page entry
"""
return self.api.get_request(url=f"fundraising_pages/{fundraising_page_id}")
[docs] def get_fundraising_pages(self, limit=None, per_page=25, page=None, filter=None):
"""
`Args:`
limit:
The number of entries to return. When None, returns all entries.
per_page
The number of entries per page to return. 25 maximum.
page
Which page of results to return
filter
The OData query for filtering results. E.g. "modified_date gt '2014-03-25'".
When None, no filter is applied.
`Returns:`
A JSON with all the fundraising_pages entries
"""
if page:
return self._get_page("fundraising_pages", page, per_page, filter)
return self._get_entry_list(
"fundraising_pages",
limit,
)
# Items
[docs] def get_items(self, list_id, limit=None, per_page=25, page=None, filter=None):
"""
`Args:`
list_id:
The unique id of the list
limit:
The number of entries to return. When None, returns all entries.
per_page:
The number of entries per page to return. 25 maximum.
page:
Which page of results to return
filter:
The OData query for filtering results. E.g. "modified_date gt '2014-03-25'".
When None, no filter is applied.
`Returns:`
A JSON with all the list item entries
"""
if page:
return self._get_page(f"lists/{list_id}/items", page, per_page, filter)
return self._get_entry_list(f"lists/{list_id}/items", limit, per_page, filter)
[docs] def get_item(self, list_id, item_id):
"""
`Args:`
list_id:
The unique id of the list
item_id:
The unique id of the item
`Returns:`
A JSON with the item entry
"""
return self.api.get_request(f"lists/{list_id}/items/{item_id}")
# Lists
[docs] def get_lists(self, limit=None, per_page=25, page=None, filter=None):
"""
`Args:`
limit:
The number of entries to return. When None, returns all entries.
per_page:
The number of entries per page to return. 25 maximum.
page:
Which page of results to return
filter:
The OData query for filtering results. E.g. "modified_date gt '2014-03-25'".
When None, no filter is applied.
`Returns:`
A JSON with all the list entries
"""
if page:
return self._get_page("lists", page, per_page, filter)
return self._get_entry_list("lists", limit, per_page, filter)
[docs] def get_list(self, list_id):
"""
`Args:`
list_id:
The unique id of the list
`Returns:`
A JSON with the list entry
"""
return self.api.get_request(f"lists/{list_id}")
# Messages
[docs] def get_messages(
self, limit=None, per_page=25, page=None, filter=None, unpack_statistics=False
):
"""
`Args:`
limit:
The number of entries to return. When None, returns all entries.
per_page:
The number of entries per page to return. 25 maximum.
page:
Which page of results to return
filter:
The OData query for filtering results. E.g. "modified_date gt '2014-03-25'".
When None, no filter is applied.
unpack_statistics:
Whether to unpack the statistics dictionary into the table. Default to False.
`Returns:`
A Parsons Table with all the messages related entries
"""
# Fetch messages
if page:
return self._get_page("messages", page, per_page, filter)
tbl = self._get_entry_list("messages", limit, per_page, filter)
# Unpack statistics
if unpack_statistics:
tbl.unpack_dict("statistics", prepend=False, include_original=True)
return tbl
[docs] def get_message(self, message_id):
"""
`Args:`
message_id:
The unique id of the message
`Returns:`
A JSON with the signature entry.
"""
return self.api.get_request(f"messages/{message_id}")
# Metadata
# Outreaches
[docs] def get_advocacy_campaign_outreaches(
self, advocacy_campaign_id, limit=None, per_page=25, page=None, filter=None
):
"""
`Args:`
advocacy_campaign_id:
The unique id of the advocacy_campaign
limit:
The number of entries to return. When None, returns all entries.
per_page:
The number of entries per page to return. 25 maximum.
page:
Which page of results to return
filter:
The OData query for filtering results. E.g. "modified_date gt '2014-03-25'".
When None, no filter is applied.
`Returns:`
A JSON with all the outreaches entries related to the advocacy_campaign_id
"""
if page:
return self._get_page(
f"advocacy_campaigns/{advocacy_campaign_id}/outreaches",
page,
per_page,
filter,
)
return self._get_entry_list(
f"advocacy_campaigns/{advocacy_campaign_id}/outreaches",
limit,
per_page,
filter,
)
[docs] def get_person_outreaches(
self, person_id, limit=None, per_page=25, page=None, filter=None
):
"""
`Args:`
person_id:
The unique id of the person
limit:
The number of entries to return. When None, returns all entries.
per_page:
The number of entries per page to return. 25 maximum.
page:
Which page of results to return
filter:
The OData query for filtering results. E.g. "modified_date gt '2014-03-25'".
When None, no filter is applied.
`Returns:`
A JSON with all the outreaches entries related to our group
"""
if page:
return self._get_page(
f"people/{person_id}/outreaches", page, per_page, filter
)
return self._get_entry_list(
f"people/{person_id}/outreaches", limit, per_page, filter
)
[docs] def get_advocacy_campaign_outreach(self, advocacy_campaign_id, outreach_id):
"""
`Args:`
advocacy_campaign_id:
The unique id of the campaign
outreach_id:
The unique id of the outreach
`Returns:`
A JSON with the outreach entry
"""
return self.api.get_request(
f"advocacy_campaigns/{advocacy_campaign_id}/outreaches/{outreach_id}"
)
[docs] def get_person_outreach(self, person_id, outreach_id):
"""
`Args:`
person_id:
The unique id of the campaign
outreach_id:
The unique id of the outreach
`Returns:`
A JSON with the outreach entry
"""
return self.api.get_request(f"people/{person_id}/outreaches/{outreach_id}")
# People
[docs] def get_people(self, limit=None, per_page=25, page=None, filter=None):
"""
`Args:`
limit:
The number of entries to return. When None, returns all entries.
per_page
The number of entries per page to return. 25 maximum.
page
Which page of results to return
filter
The OData query for filtering results. E.g. "modified_date gt '2014-03-25'".
When None, no filter is applied.
`Returns:`
A list of JSONs of people stored in Action Network.
"""
if page:
return self._get_page("people", page, per_page, filter=filter)
return self._get_entry_list("people", limit, per_page, filter=filter)
[docs] def get_person(self, person_id):
"""
`Args:`
person_id:
Id of the person.
`Returns:`
A JSON of the entry. If the entry doesn't exist, Action Network returns
``{'error': 'Couldn't find person with id = <id>'}``.
"""
return self.api.get_request(url=f"people/{person_id}")
[docs] def upsert_person(
self,
email_address=None,
given_name=None,
family_name=None,
tags=None,
languages_spoken=None,
postal_addresses=None,
mobile_number=None,
mobile_status="subscribed",
background_processing=False,
**kwargs,
):
"""
Creates or updates a person record. In order to update an existing record instead of
creating a new one, you must supply an email or mobile number which matches a record
in the database.
Identifiers are intentionally not included as an option on
this method, because their use can cause buggy behavior if
they are not globally unique. ActionNetwork support strongly
encourages developers not to use custom identifiers.
`Args:`
email_address:
Either email_address or mobile_number are required. Can be any of the following
- a string with the person's email
- a list of strings with a person's emails
- a dictionary with the following fields
- email_address (REQUIRED)
- primary (OPTIONAL): Boolean indicating the user's primary email address
- status (OPTIONAL): can taken on any of these values
- "subscribed"
- "unsubscribed"
- "bouncing"
- "previous bounce"
- "spam complaint"
- "previous spam complaint"
given_name:
The person's given name
family_name:
The person's family name
tags:
Optional field. A list of strings of pre-existing tags to be applied to the person.
languages_spoken:
Optional field. A list of strings of the languages spoken by the person
postal_addresses:
Optional field. A list of dictionaries.
For details, see Action Network's documentation:
https://actionnetwork.org/docs/v2/person_signup_helper
mobile_number:
Either email_address or mobile_number are required. Can be any of the following
- a string with the person's cell phone number
- an integer with the person's cell phone number
- a list of strings with the person's cell phone numbers
- a list of integers with the person's cell phone numbers
- a dictionary with the following fields
- number (REQUIRED)
- primary (OPTIONAL): Boolean indicating the user's primary mobile number
- status (OPTIONAL): can taken on any of these values
- "subscribed"
- "unsubscribed"
mobile_status:
'subscribed' or 'unsubscribed'
background_request: bool
If set `true`, utilize ActionNetwork's "background processing". This will return
an immediate success, with an empty JSON body, and send your request to the
background queue for eventual processing.
https://actionnetwork.org/docs/v2/#background-processing
**kwargs:
Any additional fields to store about the person. Action Network allows
any custom field.
Adds a person to Action Network
"""
email_addresses_field = None
if isinstance(email_address, str):
email_addresses_field = [{"address": email_address}]
elif isinstance(email_address, list):
if isinstance(email_address[0], str):
email_addresses_field = [{"address": email} for email in email_address]
email_addresses_field[0]["primary"] = True
if isinstance(email_address[0], dict):
email_addresses_field = email_address
mobile_numbers_field = None
if isinstance(mobile_number, str):
mobile_numbers_field = [
{"number": re.sub("[^0-9]", "", mobile_number), "status": mobile_status}
]
elif isinstance(mobile_number, int):
mobile_numbers_field = [
{"number": str(mobile_number), "status": mobile_status}
]
elif isinstance(mobile_number, list):
if len(mobile_number) > 1:
raise ("Action Network allows only 1 phone number per activist")
if isinstance(mobile_number[0], list):
mobile_numbers_field = [
{"number": re.sub("[^0-9]", "", cell), "status": mobile_status}
for cell in mobile_number
]
mobile_numbers_field[0]["primary"] = True
if isinstance(mobile_number[0], int):
mobile_numbers_field = [
{"number": cell, "status": mobile_status} for cell in mobile_number
]
mobile_numbers_field[0]["primary"] = True
if isinstance(mobile_number[0], dict):
mobile_numbers_field = mobile_number
if not email_addresses_field and not mobile_numbers_field:
raise (
"Either email_address or mobile_number is required and can be formatted "
"as a string, list of strings, a dictionary, a list of dictionaries, or "
"(for mobile_number only) an integer or list of integers"
)
data = {"person": {}}
if email_addresses_field is not None:
data["person"]["email_addresses"] = email_addresses_field
if mobile_numbers_field is not None:
data["person"]["phone_numbers"] = mobile_numbers_field
if given_name is not None:
data["person"]["given_name"] = given_name
if family_name is not None:
data["person"]["family_name"] = family_name
if languages_spoken is not None:
data["person"]["languages_spoken"] = languages_spoken
if postal_addresses is not None:
data["person"]["postal_addresses"] = postal_addresses
if tags is not None:
data["add_tags"] = tags
data["person"]["custom_fields"] = {**kwargs}
url = f"{self.api_url}/people"
if background_processing:
url = f"{url}?background_processing=true"
response = self.api.post_request(url, data=json.dumps(data))
identifiers = response["identifiers"]
person_id = [
entry_id.split(":")[1]
for entry_id in identifiers
if "action_network:" in entry_id
]
if not person_id:
logger.error(f"Response gave no valid person_id: {identifiers}")
else:
person_id = person_id[0]
if response["created_date"] == response["modified_date"]:
logger.info(f"Entry {person_id} successfully added.")
else:
logger.info(f"Entry {person_id} successfully updated.")
return response
[docs] def add_person(
self,
email_address=None,
given_name=None,
family_name=None,
tags=None,
languages_spoken=None,
postal_addresses=None,
mobile_number=None,
mobile_status="subscribed",
**kwargs,
):
"""
Creates a person in the database. WARNING: this endpoint has been deprecated in favor of
upsert_person.
"""
logger.warning(
"Method 'add_person' has been deprecated. Please use 'upsert_person'."
)
# Pass inputs to preferred method:
self.upsert_person(
email_address=email_address,
given_name=given_name,
family_name=family_name,
languages_spoken=languages_spoken,
postal_addresses=postal_addresses,
mobile_number=mobile_number,
mobile_status=mobile_status,
**kwargs,
)
[docs] def update_person(self, entry_id, background_processing=False, **kwargs):
"""
Updates a person's data in Action Network, given their Action Network ID. Note that you
can't alter a person's tags with this method. Instead, use upsert_person.
`Args:`
entry_id:
The person's Action Network id
background_processing: bool
If set `true`, utilize ActionNetwork's "background processing". This will return
an immediate success, with an empty JSON body, and send your request to the
background queue for eventual processing.
https://actionnetwork.org/docs/v2/#background-processing
**kwargs:
Fields to be updated. The possible fields are
email_address:
Can be any of the following
- a string with the person's email
- a dictionary with the following fields
- email_address (REQUIRED)
- primary (OPTIONAL): Boolean indicating the user's
primary email address
- status (OPTIONAL): can taken on any of these values
- "subscribed"
- "unsubscribed"
- "bouncing"
- "previous bounce"
- "spam complaint"
- "previous spam complaint"
given_name:
The person's given name
family_name:
The person's family name
languages_spoken:
Optional field. A list of strings of the languages spoken by the person
postal_addresses:
Optional field. A list of dictionaries.
For details, see Action Network's documentation:
https://actionnetwork.org/docs/v2/people#put
custom_fields:
A dictionary of any other fields to store about the person.
"""
data = {**kwargs}
url = f"{self.api_url}/people/{entry_id}"
if background_processing:
url = f"{url}?background_processing=true"
response = self.api.put_request(
url=url,
data=json.dumps(data),
success_codes=[204, 201, 200],
)
logger.info(f"Person {entry_id} successfully updated")
return response
# Petitions
[docs] def get_petitions(self, limit=None, per_page=25, page=None, filter=None):
"""
`Args:`
limit:
The number of entries to return. When None, returns all entries.
per_page:
The number of entries per page to return. 25 maximum.
page:
Which page of results to return
filter:
The OData query for filtering results. E.g. "modified_date gt '2014-03-25'".
When None, no filter is applied.
`Returns:`
A JSON with all of the petitions entries
"""
if page:
return self._get_page("petitions", page, per_page, filter)
return self._get_entry_list("petitions", limit, per_page, filter)
[docs] def get_petition(self, petition_id):
"""
`Args:`
petition_id:
The unique id of the petition
`Returns:`
A JSON with the petition entry
"""
return self.api.get_request(f"petitions/{petition_id}")
# Queries
[docs] def get_queries(self, limit=None, per_page=25, page=None, filter=None):
"""
`Args:`
limit:
The number of entries to return. When None, returns all entries.
per_page:
The number of entries per page to return. 25 maximum.
page:
Which page of results to return
filter:
The OData query for filtering results. E.g. "modified_date gt '2014-03-25'".
When None, no filter is applied.
`Returns:`
A JSON with all the query entries
"""
if page:
return self._get_page("queries", page, per_page, filter)
return self._get_entry_list("queries", limit, per_page, filter)
[docs] def get_query(self, query_id):
"""
`Args:`
query_id:
The unique id of the query
`Returns:`
A JSON with the query entry
"""
return self.api.get_request(f"queries/{query_id}")
# Signatures
[docs] def get_petition_signatures(
self, petition_id, limit=None, per_page=25, page=None, filter=None
):
"""
`Args:`
petition_id:
The unique id of the petition
limit:
The number of entries to return. When None, returns all entries.
per_page:
The number of entries per page to return. 25 maximum.
page:
Which page of results to return
filter:
The OData query for filtering results. E.g. "modified_date gt '2014-03-25'".
When None, no filter is applied.
`Returns:`
A JSON with all the signatures related to the petition entry
"""
if page:
return self._get_page(
f"petitions/{petition_id}/signatures", page, per_page, filter
)
return self._get_entry_list(
f"petitions/{petition_id}/signatures", limit, per_page, filter
)
[docs] def get_person_signatures(
self, person_id, limit=None, per_page=25, page=None, filter=None
):
"""
`Args:`
person_id:
The unique id of the person
limit:
The number of entries to return. When None, returns all entries.
per_page:
The number of entries per page to return. 25 maximum.
page:
Which page of results to return
filter:
The OData query for filtering results. E.g. "modified_date gt '2014-03-25'".
When None, no filter is applied.
`Returns:`
A JSON with all the signatures related to the petition entry
"""
if page:
return self._get_page(
f"people/{person_id}/signatures", page, per_page, filter
)
return self._get_entry_list(
f"people/{person_id}/signatures", limit, per_page, filter
)
[docs] def get_petition_signature(self, petition_id, signature_id):
"""
`Args:`
petition_id:
The unique id of the petition
signature_id:
The unique id of the signature
`Returns:`
A JSON with the signature entry
"""
return self.api.get_request(
f"petitions/{petition_id}/signatures/{signature_id}"
)
[docs] def get_person_signature(self, person_id, signature_id):
"""
`Args:`
person_id:
The unique id of the person
signature_id:
The unique id of the signature
`Returns:`
A JSON with the signature entry
"""
return self.api.get_request(f"people/{person_id}/signatures/{signature_id}")
# Submissions
[docs] def get_person_submissions(
self, person_id, limit=None, per_page=25, page=None, filter=None
):
"""
`Args:`
person_id:
The unique id of the person
limit:
The number of entries to return. When None, returns all entries.
per_page:
The number of entries per page to return. 25 maximum.
page:
Which page of results to return
filter:
The OData query for filtering results. E.g. "modified_date gt '2014-03-25'".
When None, no filter is applied.
`Returns:`
A JSON with all the submissions entries related with our group
"""
if page:
return self._get_page(
f"people/{person_id}/submissions", page, per_page, filter
)
return self._get_entry_list(
f"people/{person_id}/submissions", limit, per_page, filter
)
[docs] def get_person_submission(self, person_id, submission_id):
"""
`Args:`
person_id:
The unique id of the submission
submission_id:
The unique id of the submission
`Returns:`
A JSON with the submission entry
"""
return self.api.get_request(f"people/{person_id}/submissions/{submission_id}")
# Tags
[docs] def get_tag(self, tag_id):
"""
`Args:`
tag_id:
Id of the tag.
`Returns:`
A JSON of the entry. If the entry doesn't exist, Action Network returns
"{'error': 'Couldn't find tag with id = <id>'}"
"""
return self.api.get_request(url=f"tags/{tag_id}")
[docs] def add_tag(self, name):
"""
`Args:`
name:
The tag's name. This is the ONLY editable field
Adds a tag to Action Network. Once created, tags CANNOT be edited or deleted.
"""
data = {"name": name}
response = self.api.post_request(
url=f"{self.api_url}/tags", data=json.dumps(data)
)
identifiers = response["identifiers"]
person_id = [
entry_id.split(":")[1]
for entry_id in identifiers
if "action_network:" in entry_id
][0]
logger.info(f"Tag {person_id} successfully added to tags.")
return response
# Taggings
[docs] def get_taggings(self, tag_id, limit=None, per_page=25, page=None, filter=None):
"""
`Args:`
tag_id:
The unique id of the tag
limit:
The number of entries to return. When None, returns all entries.
per_page:
The number of entries per page to return. 25 maximum.
page:
Which page of results to return
filter:
The OData query for filtering results. E.g. "modified_date gt '2014-03-25'".
When None, no filter is applied.
`Returns:`
A JSON with all the tagging entries associated with the tag_id
"""
if page:
return self._get_page(f"tags/{tag_id}/taggings", page, per_page, filter)
return self._get_entry_list(f"tags/{tag_id}/taggings", limit, per_page, filter)
[docs] def get_tagging(self, tag_id, tagging_id):
"""
`Args:`
tag_id:
The unique id of the tag
tagging_id:
The unique id of the tagging
`Returns:`
A JSON with the tagging entry
"""
return self.api.get_request(f"tags/{tag_id}/taggings/{tagging_id}")
# Wrappers
[docs] def get_wrappers(self, limit=None, per_page=25, page=None, filter=None):
"""
`Args:`
limit:
The number of entries to return. When None, returns all entries.
per_page:
The number of entries per page to return. 25 maximum.
page:
Which page of results to return
filter:
The OData query for filtering results. E.g. "modified_date gt '2014-03-25'".
When None, no filter is applied.
`Returns:`
A JSON with all the wrapper entries
"""
if page:
return self._get_page("wrappers", page, per_page, filter)
return self._get_entry_list("wrappers", limit, per_page, filter)
[docs] def get_wrapper(self, wrapper_id):
"""
`Args:`
wrapper_id:
The unique id of the wrapper
tagging_id:
The unique id of the tagging
`Returns:`
A JSON with the wrapper entry
"""
return self.api.get_request(f"wrappers/{wrapper_id}")