Select Git revision
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
shopping_client.py 5.99 KiB
import requests
import json
import urllib.parse
import getpass
import pandas as pd
from typing import Union, Optional
from warnings import warn
class shopping_client:
endpoint = "esap-api/accounts/user-profiles/"
def __init__(
self,
token: str = None,
host: str = "http://localhost:5555/",
connectors: list = [],
):
"""Constructor.
Parameters
----------
token : str
OAuth access token as a string.
host : str
Hostname of the EASP Gateway backend.
connectors : list
List of connector classes that can handle specific types of shopping
item.
"""
self.token = token
self.host = host
self.connectors = connectors
self.basket = None
self.username = None
def get_basket(
self,
convert_to_pandas: bool = False,
reload: bool = False,
filter_archives: bool = False,
) -> Union[list, pd.DataFrame, None]:
"""Retrieve the shopping basket for a user.
Prompts for access token if one was not supplied to constructor.
Parameters
----------
convert_to_pandas : bool
If `True`, attempt to convert items from the basket to pandas Series
and concatenate items from the same archive into pandas DataFrames.
Note that items that cannot be converted by any of the connector
classes passed to the constructor will be ignored and lost. The
default `convert_to_pandas = False` returns all items as a `list`
of `dict`s
reload : bool
If `True` a fresh query is issued to the ESAP API to refresh the
basket contents.
filter_archives : bool
If `True` then the items are checked for an 'archive' value.
If this archive matches the 'archive' property of the provided connector
then the item is handled further, otherwise it is ignored.
Returns
-------
Union[list, pd.DataFrame, None]
Description of returned object.
"""
if self.basket is None or reload:
url = urllib.parse.urljoin(self.host, shopping_client.endpoint)
response = requests.get(url, headers=self._request_header())
if response.ok:
self.basket = json.loads(response.content)["results"][0][
"shopping_cart"
]
else:
warn(f"Unable to load data from {self.host}; is your key valid?")
if filter_archives:
self.basket = self._filter_on_archive()
if convert_to_pandas:
return self._basket_to_pandas()
return self.basket
def add_to_basket(self, items: list):
"""Add items to user's shopping basket.
Parameters
----------
items : list
List of objects, each containing the data for a single
shopping item. Objects should be JSON-serializable. Examples
include Python `dict`s or `list`s.
Returns
-------
bool
`True` if items were successfully added otherwise `False`.
"""
url = urllib.parse.urljoin(self.host, shopping_client.endpoint)
if self.username is None:
# Retrieve username
response = requests.get(url, headers=self._request_header())
if response.ok:
self.username = json.loads(response.content)["results"][0]["user_name"]
else:
warn(
f"Unable to retrieve username from {self.host}; is your key valid?"
)
return False, response
# PATCH not working properly need to add existing basket items
payload = {
"shopping_cart": [
json.loads(item["item_data"]) for item in self.get_basket()
]
+ items
}
# trailing "/" required for PATCH
url = urllib.parse.urljoin(url, self.username) + "/"
response = requests.patch(url, json=payload, headers=self._request_header())
if response.ok:
print(
f"{len(items)} item{'s' if len(items) > 1 else ''} successfully added."
)
return True
else:
warn(f"Unable to add data to basket at {self.host}; is your key valid?")
return False, response
def _request_header(self):
while self.token is None:
self._get_token()
return dict(Accept="application/json", Authorization=f"Bearer {self.token}")
# filter on items belonging to the provided connectors
def _filter_on_archive(self):
filtered_items = []
if len(self.connectors):
for item in self.basket:
item_data = json.loads(item["item_data"])
for connector in self.connectors:
if (
"archive" in item_data
and item_data["archive"] == connector.archive
):
filtered_items.append(item)
return filtered_items
def _basket_to_pandas(self):
if len(self.connectors):
converted_basket = {
connector.name: pd.concat(
[
connector.basket_item_to_pandas(item)
for item in self.basket
if connector.validate_basket_item(item)
],
axis=1,
)
for connector in self.connectors
}
return {
name: data.to_frame().T if data.ndim < 2 else data.T
for name, data in converted_basket.items()
}
warn(
"No archive connectors specified - could not convert any basket items to Pandas DataFrame"
)
return self.basket
def _get_token(self):
self.token = getpass.getpass("Enter your ESAP access token:")