Merge pull request #3 from KaveTech/feat/entity-attributes

Feat/entity attributes

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)

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
+        )

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,
+        }