Skip to content
Snippets Groups Projects
Select Git revision
  • f8cda1bf69bfacb203679812530bf20b7aad6f04
  • master default protected
  • add_to_basket
3 results

shopping_client.py

Blame
  • Forked from ASTRON SDC / ESCAPE WP5 / esap-userprofile-python-client
    Source project has a limited visibility.
    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:")