Added def to upload media files to Akeneo

akeneo_connector/akeneo_connector.py

@@ -24,6 +24,7 @@ class AkeneoConnector:
     PRODUCT_URL = 'https://{origin}/api/rest/{version}/products/{identifier}'
     PRODUCTS_URL = 'https://{origin}/api/rest/{version}/products'
     ATTRIBUTE_URL = 'https://{origin}/api/rest/{version}/attributes/{code}'
+    PRODUCTS_MEDIA_URL = 'https://{origin}/api/rest/{version}/media-files'
 
     def __init__(self, origin: str | None = None, username = None, password = None, auth_token = None, auth_url = None, version='v1'):
         """
@@ -54,6 +55,8 @@ def __init__(self, origin: str | None = None, username = None, password = None,
         self.version = version
         self.product_url = self.PRODUCT_URL.format(origin=self.origin, version=self.version, identifier='{identifier}')
         self.products_url = self.PRODUCTS_URL.format(origin=self.origin, version=self.version)
+        self.products_media_url = self.PRODUCTS_MEDIA_URL.format(origin=self.origin, version=self.version)
+
 
     def get_access_token(self):
         """
@@ -189,4 +192,39 @@ def get_attribute(self, attributecode: str):
         except:
             data = response.text
 
+        return data
+
+    def upload_media(self, url: str, product_dict: dict, media_file: str):
+        """
+        Uploads media to Akeneo.
+
+        Args:
+            payload (dict): The payload to send in the request.
+        """
+        # Convert to JSON-string
+        data_str = json.dumps({
+            'product': product_dict,
+            'file': media_file	
+        })
+
+        # Set the payload to the JSON-string
+        print(f"POST {url}")
+        headers = {
+            'Authorization': 'Bearer ' + self.access_token,
+            'Content-type': 'multipart/form-data'
+        }
+
+        response = req.post(url, headers=headers, data=data_str)
+
+        # Check if the request was successful
+        if response.status_code < 200 or response.status_code >= 300:
+            print(f"Request error: {response.status_code} - {response.text}")
+            return None
+
+        # Try to parse the response as JSON
+        try:
+            data = response.json()
+        except:
+            data = response.text
+
         return data

akeneo_connector/akeneo_product.py

@@ -287,4 +287,29 @@ def get_media(self, media_attribute: str, locale: str | None = None, scope: str
         media_url = self.get_href(media_attribute, locale, scope)
         if media_url:
             return self.connector.get_media_file(media_url)
-        return None
+        return None
+
+    def set_media(self, attribute: str, locale: str | None = None, scope: str | None = None, file: str | None = None):
+        """
+        Sets a media file for the given attribute.
+        
+        Args:
+            attribute (str): The attribute to set the media file for.
+            locale (str): The locale of the value.
+            scope (str): The scope of the value.
+            data (str): The data of the value.
+            
+        Returns:
+            bool: JSON response if successful, None otherwise.
+        """
+
+        url = self.connector.upload_media(attribute, locale, scope)
+        if url is None:
+            return None
+
+        return self.connector.upload_media(url, {
+            'identifier': self.identifier,
+            'attribute': attribute,
+            'locale': locale,
+            'scope': scope  
+        }, file)