"""
A client for working with the Heap server side API
see https://docs.heap.io/reference#track-1
"""
import collections
import datetime
import functools
import operator
import posixpath
import pathlib
import typing
import urllib.parse
import marshmallow
import requests
[docs]class HeapTrackRequestEventSchema(marshmallow.Schema):
"""
schema for `POST`\\ing to :code:`track`
"""
identity = marshmallow.fields.String()
timestamp = marshmallow.fields.DateTime(default=datetime.datetime.now)
event = marshmallow.fields.String()
idempotency_key = marshmallow.fields.String()
properties = marshmallow.fields.Mapping(
keys=marshmallow.fields.String,
values=marshmallow.fields.String,
)
[docs]class HeapAddUserPropertiesUserSchema(marshmallow.Schema):
"""
schema for `POST`\\ing to :code:`add_user_properties`
"""
identity = marshmallow.fields.String()
properties = marshmallow.fields.Mapping(
keys=marshmallow.fields.String,
values=marshmallow.fields.String,
)
[docs]class HeapAddAccountPropertiesAccountSchema(marshmallow.Schema):
"""
schema for `POST`\\ing to :code:`add_account_properties`
"""
account_id = marshmallow.fields.String()
properties = marshmallow.fields.Mapping(
keys=marshmallow.fields.String,
values=marshmallow.fields.String,
)
[docs]class HeapClient(object):
"""
A client for working with the Heap server side API
"""
BASE_URL = 'https://heapanalytics.com/api/'
ENDPOINTS = {
'track': HeapTrackRequestEventSchema,
'add_user_properties': HeapAddUserPropertiesUserSchema,
'add_account_properties': HeapAddAccountPropertiesAccountSchema,
}
[docs] @classmethod
def api_url(cls, *endpoint_path_segments):
"""
combine path segments with the API base url
"""
base_url = urllib.parse.urlsplit(cls.BASE_URL)
segments = (base_url.path,) + endpoint_path_segments
new_path = pathlib.PosixPath(
*map(operator.methodcaller('strip', posixpath.sep), segments)).as_posix()
return base_url._replace(path=new_path).geturl()
[docs] def __init__(self, app_id):
self.app_id = app_id
def __getattr__(self, name):
if name in self.ENDPOINTS:
return functools.partial(self.post, name)
raise AttributeError(name)
[docs] def post(
self,
endpoint: str,
endpoint_data: typing.Union[typing.Mapping, typing.Iterable],
app_id: str = None):
"""
`POST` `endpoint_data` to `endpoint` as `application/json`
"""
schema_class = self.ENDPOINTS[endpoint]
if isinstance(endpoint_data, collections.abc.Mapping):
schema = schema_class()
json_data = schema.dump(endpoint_data)
else:
schema = schema_class(many=True)
json_data = dict(events=schema.dump(endpoint_data))
json_data.setdefault('app_id', app_id or self.app_id)
url = self.api_url(endpoint)
response = requests.post(url, json=json_data)
response.raise_for_status()
return response