diff --git a/pyakeneo/auth.py b/pyakeneo/auth.py index 5882dc3..fb8ae18 100644 --- a/pyakeneo/auth.py +++ b/pyakeneo/auth.py @@ -59,8 +59,7 @@ def _request_a_token(self, grant_type="password"): url = urljoin(self._base_url, self.TOKEN_PATH) r = requests.post(url, data=data, headers=headers) - if r.status_code != 200: - raise requests.HTTPError("Status code: {0}".format(r.status_code)) + r.raise_for_status() try: json_data = json.loads(r.text) diff --git a/pyakeneo/resources.py b/pyakeneo/resources.py index 8a3f032..029a638 100644 --- a/pyakeneo/resources.py +++ b/pyakeneo/resources.py @@ -30,13 +30,8 @@ def fetch_list(self, args=None): url = self._endpoint r = self._session.get(url, params=args) + r.raise_for_status() - if r.status_code != 200: - raise requests.HTTPError( - "Status code: {0}. Content: {1}".format(r.status_code, r.text) - ) - - # c = Collection(self._session, json_text=r.text) c = Result.from_json_text(self._session, json_text=r.text) return c @@ -65,11 +60,7 @@ def fetch_item(self, code_or_item): url = urljoin(self._endpoint, code) r = self._session.get(url) - - if r.status_code != 200: - raise requests.HTTPError( - "The item {0} doesn't exit: {1}".format(code, r.status_code) - ) + r.raise_for_status() return json.loads(r.text) # returns item as a dict @@ -130,8 +121,8 @@ def update_create_list(self, items, code=None): num = 100 n = math.ceil(len(items) / num) - itemss = [items[i : i + num] for i in range(0, (n - 1) * num, num)] - itemss.append(items[(n - 1) * num :]) + itemss = [items[i: i + num] for i in range(0, (n - 1) * num, num)] + itemss.append(items[(n - 1) * num:]) return [ item @@ -140,9 +131,7 @@ def update_create_list(self, items, code=None): ] if r.status_code != 200: - raise requests.HTTPError( - "Status code: {0}. Content: {1}".format(r.status_code, r.text) - ) + r.raise_for_status() else: statuses = [] @@ -412,6 +401,29 @@ class ReferenceEntityRecordPool( pass +class ReferenceEntityAttributeOptionsPool( + ResourcePool, + CodeBasedResource, + GettableResource, + ListableResource, + UpdatableResource, +): + pass + + +class ReferenceEntityAttributePool( + ResourcePool, + CodeBasedResource, + GettableResource, + ListableResource, + UpdatableResource, +): + def options(self, code): + return ReferenceEntityAttributeOptionsPool( + urljoin(self._endpoint, code, "options/"), self._session + ) + + class ReferenceEntityPool( ResourcePool, CodeBasedResource, @@ -423,3 +435,8 @@ def records(self, entity_code): return ReferenceEntityRecordPool( urljoin(self._endpoint, entity_code, "records/"), self._session ) + + def attributes(self, entity_code): + return ReferenceEntityAttributePool( + urljoin(self._endpoint, entity_code, "attributes/"), self._session + ) diff --git a/pyakeneo/result.py b/pyakeneo/result.py index 497116b..20eb9a0 100644 --- a/pyakeneo/result.py +++ b/pyakeneo/result.py @@ -1,25 +1,39 @@ -# -*- coding: utf-8 -*- +from __future__ import annotations import json +from typing import Dict, Iterable + +import requests class Result(object): - """Holds the result of a search. It can be iterated through as a list, + """ + Holds the result of a search. It can be iterated through as a list, as an iterator, or as a generator. Note that only one iteration should be used on a given Collection object. Search results are paginated: https://api.akeneo.com/documentation/pagination.html - The next page will be loaded once the user - iterated over the whole current page. The content of the new page will replace - the content of the previous page.""" + The next page will be loaded once the user iterated over the whole current page. + The content of the new page will replace the content of the previous page. + """ - def __init__(self, session, count, current_items, link_first, link_next, link_self): - self._items = current_items + def __init__( + self, + session: requests.Session, + *, + items: list | dict, + count: int, + link_first: str, + link_next: str, + link_self: str, + ): + self._session = session + self._items = items + self._count = count self._link_next = link_next self._link_self = link_self self._link_first = link_first - self._count = count - self._session = session + self._page_iterator = iter(self._items) self._reached_the_end = False @@ -32,7 +46,7 @@ def __iter__(self): def __next__(self): try: return next(self._page_iterator) - except Exception: + except StopIteration: self.fetch_next_page() if not self._reached_the_end: return next(self._page_iterator) @@ -44,16 +58,16 @@ def get_page_items(self): def fetch_next_page(self): """Return True if a next page exists. Returns False otherwise.""" - if self._link_next: - r = self._session.get(self._link_next) - if r.status_code == 200: - ( - self._link_first, - self._link_self, - self._link_next, - self._items, - self._count, - ) = Result.parse_page(json.loads(r.text)) + if self.is_paginated(self._items) and self._link_next: + response = self._session.get(self._link_next) + if response.ok: + next_page = Result.parse_page(json.loads(response.text)) + self._items = next_page["items"] + self._count = next_page["count"] + self._link_next = next_page["link_next"] + self._link_self = next_page["link_self"] + self._link_first = next_page["link_first"] + self._page_iterator = iter(self._items) self._reached_the_end = False else: @@ -74,23 +88,46 @@ def get_self_link(self): def get_first_link(self): return self._link_first - @staticmethod - def parse_page(json_data): + @classmethod + def parse_page(cls, json_data: dict) -> Dict: """Returns (next link, retrieved items, count of items)""" final_next_link = None next_link = json_data["_links"].get("next") if next_link: final_next_link = next_link["href"] - return ( - json_data["_links"]["first"]["href"], - json_data["_links"]["self"]["href"], - final_next_link, - json_data["_embedded"]["items"], - json_data.get("items_count"), - ) + return { + "items": json_data["_embedded"]["items"], + "count": json_data.get("items_count"), + "link_first": json_data["_links"]["first"]["href"], + "link_next": final_next_link, + "link_self": json_data["_links"]["self"]["href"], + } + + @classmethod + def parse_result(cls, session: requests.Session, json_data: dict | list): + if cls.is_paginated(json_data): + return cls(session, **cls.parse_page(json_data)) + else: + return cls(session, **cls.parse_non_paginated(json_data)) @staticmethod - def from_json_text(session, json_text): + def from_json_text(session: requests.Session, json_text: str) -> "Result": json_data = json.loads(json_text) - parsed = Result.parse_page(json_data) - return Result(session, parsed[4], parsed[3], parsed[0], parsed[2], parsed[1]) + return Result.parse_result(session, json_data) + + @classmethod + def is_paginated(cls, json_data: dict | list): + return isinstance(json_data, dict) and "_links" in json_data + + @classmethod + def parse_non_paginated(cls, json_data: list) -> Dict: + link_self = "" + if json_data and "_links" in json_data[0]: + link_self = json_data["_links"]["self"]["href"] + return { + "items": json_data, + "count": len(json_data), + "link_first": "", + "link_next": "", + "link_self": link_self, + }