api_products.py

"""Get a bunch of products from the API."""
import json
import os
import requests
from requests.auth import HTTPBasicAuth


class APISession:
    """An API session."""

    url = "https://api.amberengine.com"

    @staticmethod
    def get_auth(user, password):
        """Get the http basic auth."""
        return HTTPBasicAuth(user, password)

    def __init__(self, user, password, url=None):
        """Init."""
        self._user = user
        self._password = password

        if url:
            self.url = url

        self.auth = self.get_auth(self.user, self.password)
        self.client = APIClient(self)

    @property
    def user(self):
        """Get user."""
        return self._user

    @user.setter
    def user(self, user):
        """Set user."""
        self._user = user

        if self.password:
            self.auth = self.get_auth(self.user, self.password)

        if self.auth:
            self.client = APIClient(self)

    @property
    def password(self):
        """Get password."""
        return self._password

    @password.setter
    def password(self, password):
        """Set password."""
        self._password = password

        if self.user:
            self.auth = self.get_auth(self.user, self.password)

        if self.auth:
            self.client = APIClient(self)

    def get_response(self, route, **kwargs):
        """Get a response from the api."""
        url = os.path.join(self.url, route)
        query_params = None

        for k, v in kwargs.items():
            if not query_params:
                query_params = f"{k}={v}"
            else:
                query_params = f"{query_params}&{k}={v}"

        if query_params:
            url = f"{url}?{query_params}"

        return requests.get(url, auth=self.auth)


class APIClient:
    """An API client."""

    def __init__(self, session):
        """Init."""
        self.session = session

    def get_brands(self):
        """Get a bunch of brands."""
        response = self.session.get_response("brands")
        brands = json.loads(response.text)

        for brand in brands:
            yield brand

    def get_products(self, brand, limit=100):
        """Get a bunch of products."""
        route = os.path.join("brands", brand, "products")
        offset = 0

        while True:
            response = self.session.get_response(route,
                                                 limit=limit, offset=offset)
            products = json.loads(response.text)
            offset += limit

            if not len(products):
                break

            yield products


if __name__ == "__main__":
    session = APISession("<your email>", "<your password>")
    client = session.client

    for brand in client.get_brands():
        for products in client.get_products(brand):
            print(products)