import json
import requests
import logging
from parsons.utilities import check_env
from parsons.utilities.api_connector import APIConnector
from parsons.etl import Table
logger = logging.getLogger(__name__)
URI = "https://api.bloomerang.co/v2/"
URI_AUTH = "https://crm.bloomerang.co/authorize/"
[docs]class Bloomerang(object):
"""
Instantiate Bloomerang class
`Args:`
api_key: str
The Bloomerang API key. Not required if the ``BLOOMERANG_API_KEY`` environmental
variable is set or if the OAuth2 authentication parameters ``client_id`` and
``client_secret`` are set.
client_id: str
The Bloomerang client ID for OAuth2 authentication. Not required if
the ``BLOOMERANG_CLIENT_ID`` env variable is set or if the ``api_key``
parameter is set. Note that the ``client_secret`` parameter must
also be set in order to use OAuth2 authentication.
client_secret: str
The Bloomerang client secret for OAuth2 authetication. Not required if
the ``BLOOMERANG_CLIENT_SECRET`` env variable is set or if the ``api_key``
parameter is set. Note that the ``client_id`` parameter must
also be set in order to use OAuth2 authentication.
"""
def __init__(self, api_key=None, client_id=None, client_secret=None):
self.api_key = check_env.check('BLOOMERANG_API_KEY', api_key, optional=True)
self.client_id = check_env.check('BLOOMERANG_CLIENT_ID', client_id, optional=True)
self.client_secret = check_env.check('BLOOMERANG_CLIENT_SECRET', client_secret,
optional=True)
self.uri = URI
self.uri_auth = URI_AUTH
self.conn = self._conn()
def _conn(self):
# Instantiate APIConnector with authentication credentials
headers = {"accept": "application/json",
"Content-Type": "application/json"}
if self.api_key is not None:
logger.info("Using API key authentication.")
headers['X-API-KEY'] = f"{self.api_key}"
elif (self.client_id is not None) & (self.client_secret is not None):
logger.info('Using OAuth2 authentication.')
self._generate_authorization_code()
self._generate_access_token()
headers['Authorization'] = f"Bearer {self.access_token}"
else:
raise Exception('Missing authorization credentials.')
return APIConnector(uri=self.uri, headers=headers)
def _generate_authorization_code(self):
data = {'client_id': self.client_id,
'response_type': 'code'}
r = requests.post(url=self.uri_auth, json=data)
self.authorization_code = r.json().get('code', None)
def _generate_access_token(self):
data = {'client_id': self.client_id,
'client_secret': self.client_secret,
'grant_type': 'authorization_code',
'code': self.authorization_code}
r = requests.post(url=self.uri + 'oauth/token', json=data)
self.access_token = r.json().get('access_token', None)
def _base_endpoint(self, endpoint, entity_id=None):
url = f'{self.uri}{endpoint}/'
if entity_id:
url = url + f'{entity_id}/'
return url
@staticmethod
def _base_pagination_params(page_number=1, page_size=50):
return {'skip': page_size * (page_number - 1), 'take': min(page_size, 50)}
@staticmethod
def _base_ordering_params(order_by=None, order_direction=None):
params = {}
if order_by:
params["orderBy"] = order_by
if order_direction:
params["orderDirection"] = order_direction
return params
def _base_create(self, endpoint, entity_id=None, **kwargs):
return self.conn.post_request(url=self._base_endpoint(endpoint, entity_id),
json=json.dumps({**kwargs}))
def _base_update(self, endpoint, entity_id=None, **kwargs):
return self.conn.put_request(url=self._base_endpoint(endpoint, entity_id),
json=json.dumps({**kwargs}))
def _base_get(self, endpoint, entity_id=None, params=None):
return self.conn.get_request(url=self._base_endpoint(endpoint, entity_id), params=params)
def _base_delete(self, endpoint, entity_id=None):
return self.conn.delete_request(url=self._base_endpoint(endpoint, entity_id))
[docs] def create_constituent(self, **kwargs):
"""
`Args:`
**kwargs:`
Fields to include, e.g., FirstName = 'Rachel'.
See the Bloomerang API docs for a full list of `fields <https://bloomerang.co/features/integrations/api/rest-api#/Constituents/post_constituent>`_. # noqa
"""
return self._base_create('constituent', **kwargs)
[docs] def update_constituent(self, constituent_id, **kwargs):
"""
`Args:`
constituent_id: str or int
Constituent ID to update
**kwargs:`
Fields to update, e.g., FirstName = 'RJ'.
See the Bloomerang API docs for a full list of `fields <https://bloomerang.co/features/integrations/api/rest-api#/Constituents/put_constituent__id_>`_. # noqa
"""
return self._base_update('constituent', entity_id=constituent_id, **kwargs)
[docs] def get_constituent(self, constituent_id):
"""
`Args:`
constituent_id: str or int
Constituent ID to get fields for
`Returns:`
A JSON of the entry or an error.
"""
return self._base_get('constituent', entity_id=constituent_id)
[docs] def delete_constituent(self, constituent_id):
"""
`Args:`
constituent_id: str or int
Constituent ID to delete
"""
return self._base_delete('constituent', entity_id=constituent_id)
[docs] def get_constituents(self, page_number=1, page_size=50, order_by=None, order_direction=None,
last_modified=None):
"""
`Args:`
page_number: int
Number of the page to fetch
page_size: int
Number of records per page (maximum allowed is 50)
order_by: str
Sorts by ``Id``, ``CreatedDate``, or ``LastModifiedDate`` (default ``Id``).
order_direction: str
Sorts the order_by in ``Asc`` or ``Desc`` order.
last_modified: str
Filters to constituents last modified after the specified date (ISO-8601 format).
`Returns:`
A Table of the entries.
"""
params = self._base_pagination_params(page_number, page_size)
params.update(self._base_ordering_params(order_by, order_direction))
if last_modified:
params["lastModified"] = last_modified
response = self._base_get('constituents', params=params)
return Table(response['Results'])
[docs] def create_transaction(self, **kwargs):
"""
`Args:`
**kwargs:`
Fields to include, e.g., CreditCardType = 'Visa'.
See the Bloomerang API docs for a full list of `fields <https://bloomerang.co/features/integrations/api/rest-api#/Transactions/post_transaction>`_. # noqa
"""
return self._base_create('transaction', **kwargs)
[docs] def update_transaction(self, transaction_id, **kwargs):
"""
`Args:`
transaction_id: str or int
Transaction ID to update
**kwargs:`
Fields to update, e.g., CreditCardType = 'Visa'.
See the Bloomerang API docs for a full list of `fields <https://bloomerang.co/features/integrations/api/rest-api#/Transactions/put_transaction__id_>`_. # noqa
"""
return self._base_update('transaction', entity_id=transaction_id, **kwargs)
[docs] def get_transaction(self, transaction_id):
"""
`Args:`
transaction_id: str or int
Transaction ID to get fields for
`Returns:`
A JSON of the entry or an error.
"""
return self._base_get('transaction', entity_id=transaction_id)
[docs] def delete_transaction(self, transaction_id):
"""
`Args:`
transaction_id: str or int
Transaction ID to delete
"""
return self._base_delete('transaction', entity_id=transaction_id)
[docs] def get_transactions(self, page_number=1, page_size=50, order_by=None, order_direction=None):
"""
`Args:`
page_number: int
Number of the page to fetch
page_size: int
Number of records per page (maximum allowed is 50)
order_by: str
Sorts by ``Date``, ``CreatedDate``, or ``LastModifiedDate`` (default ``Date``).
order_direction: str
Sorts the order_by in ``Asc`` or ``Desc`` order (default ``Desc``).
`Returns:`
A JSON of the entry or an error.
"""
params = self._base_pagination_params(page_number, page_size)
params.update(self._base_ordering_params(order_by, order_direction))
response = self._base_get('transactions', params=params)
return Table(response['Results'])
[docs] def get_transaction_designation(self, designation_id):
"""
`Args:`
designation_id: str or int
Transaction Designation ID to get fields for
`Returns:`
A JSON of the entry or an error.
"""
return self._base_get('transaction/designation', entity_id=designation_id)
[docs] def get_transaction_designations(self, page_number=1, page_size=50, order_by=None,
order_direction=None):
"""
`Args:`
page_number: int
Number of the page to fetch
page_size: int
Number of records per page (maximum allowed is 50)
order_by: str
Sorts by ``Date``, ``CreatedDate``, or ``LastModifiedDate`` (default ``Date``).
order_direction: str
Sorts the order_by in ``Asc`` or ``Desc`` order (default ``Desc``).
`Returns:`
A JSON of the entry or an error.
"""
params = self._base_pagination_params(page_number, page_size)
params.update(self._base_ordering_params(order_by, order_direction))
response = self._base_get('transactions/designations', params=params)
return Table(response['Results'])
[docs] def create_interaction(self, **kwargs):
"""
`Args:`
**kwargs:`
Fields to include, e.g., Channel = "Email".
See the Bloomerang API docs for a full list of `fields <https://bloomerang.co/features/integrations/api/rest-api#/Interactions/post_interaction>`_. # noqa
"""
return self._base_create('interaction', **kwargs)
[docs] def update_interaction(self, interaction_id, **kwargs):
"""
`Args:`
interaction_id: str or int
Interaction ID to update
**kwargs:`
Fields to update, e.g., EmailAddress = "user@example.com".
See the Bloomerang API docs for a full list of `fields <https://bloomerang.co/features/integrations/api/rest-api#/Interactions/put_interaction__id_>`_. # noqa
"""
return self._base_update('interaction', entity_id=interaction_id, **kwargs)
[docs] def get_interaction(self, interaction_id):
"""
`Args:`
interaction_id: str or int
Interaction ID to get fields for
`Returns:`
A JSON of the entry or an error.
"""
return self._base_get('interaction', entity_id=interaction_id)
[docs] def delete_interaction(self, interaction_id):
"""
`Args:`
interaction_id: str or int
Interaction ID to delete
"""
return self._base_delete('interaction', entity_id=interaction_id)
[docs] def get_interactions(self, page_number=1, page_size=50):
"""
`Args:`
page_number: int
Number of the page to fetch
page_size: int
Number of records per page (maximum allowed is 50)
`Returns:`
A JSON of the entry or an error.
"""
params = self._base_pagination_params(page_number, page_size)
response = self._base_get('interactions', params=params)
return Table(response['Results'])