Config reader (#11)

* Add config reader module and update inventree and digikey packages

* No need to run CI on every push, just PRs are okay

* Remove pypy support

.github/workflows/python-test.yaml

@@ -1,6 +1,6 @@
 name: Pytest
 
-on: [push, pull_request]
+on: [pull_request]
 
 jobs:
   build:
@@ -10,7 +10,7 @@ jobs:
         DIGIKEY_INVENTREE_TEST_CONFIG_PATH: "tests/test_data/test_config.ini"
       strategy:
         matrix:
-          python-version: ["pypy3.8", "3.8", "pypy3.9", "pypy3.10", "3.9", "3.10", "3.11"]
+          python-version: ["3.8", "3.9", "3.10", "3.11"]
 
       steps:
       - uses: actions/checkout@v4

inventree_digikey/ConfigReader.py

@@ -0,0 +1,94 @@
+import configparser
+
+from pathlib import Path
+from inventree.api import InvenTreeAPI
+
+class ConfigReader:
+    """
+    Manage global configuration settings and Inventree instance
+
+    Optionally read configuration from a file
+    """
+
+    DEFAULT_DIGIKEY_STORAGE_PATH = "."
+    DEFAULT_DIGIKEY_CLIENT_SANDBOX = "False"
+
+    def __init__(self, config_file=None):
+        self.config = configparser.ConfigParser()
+
+        self.digikey_client_id = None
+        self.digikey_client_secret = None
+        self.digikey_client_sandbox = self.DEFAULT_DIGIKEY_CLIENT_SANDBOX
+        self.digikey_storage_path = self.DEFAULT_DIGIKEY_STORAGE_PATH
+
+        self.inventree_url = None
+        self.inventree_username = None
+        self.inventree_password = None
+        self._inventree_api = None
+        self._reinit_api = False
+
+        if config_file:
+            self.read_config(config_file)
+
+    def read_config(self, config_file: Path) -> None:
+        """
+        Read configuration from a file
+        """
+        self.config.read(config_file)
+        self.digikey_client_id = self.config['DIGIKEY_API']['CLIENT_ID']
+        self.digikey_client_secret = self.config['DIGIKEY_API']['CLIENT_SECRET']
+        if 'SANDBOX' in self.config['DIGIKEY_API']:
+            self.digikey_client_sandbox = self.config['DIGIKEY_API'].getboolean('SANDBOX')
+        if 'STORAGE_PATH' in self.config['DIGIKEY_API']:
+            self.digikey_storage_path = self.config['DIGIKEY_API']['STORAGE_PATH']
+        self._inventree_url = self.config['INVENTREE_API']['URL']
+        self._inventree_username = self.config['INVENTREE_API']['USER']
+        self._inventree_password = self.config['INVENTREE_API']['PASSWORD']
+
+    @property
+    def inventree_api(self):
+        if self._inventree_api is None or self._reinit_api: # Allows us to reinit the api if the config changes
+            if self.inventree_url and self.inventree_username and self.inventree_password:
+                try: 
+                    api = InvenTreeAPI(self.inventree_url, username=self.inventree_username, password=self.inventree_password)
+                except:
+                    print("Error: Could not connect to Inventree API") #FIXME
+
+                self._reinit_api = False
+                self._inventree_api = api
+                return api
+            else:
+                raise AttributeError("Cannot init inventree_api without inventree_[url|username|password] set")
+        else:
+            return self._inventree_api
+
+    @inventree_api.setter
+    def inventree_api(self, value):
+        raise AttributeError("Cannot set inventree_api directly. Use inventree_[url|username|password] instead")
+
+    @property
+    def inventree_url(self):
+        return self._inventree_url
+
+    @property
+    def inventree_username(self):
+        return self._inventree_username
+
+    @property
+    def inventree_password(self):
+        return self._inventree_password
+
+    @inventree_url.setter
+    def inventree_url(self, value):
+        self._reinit_api = True
+        self._inventree_url = value
+
+    @inventree_username.setter
+    def inventree_username(self, value):
+        self._reinit_api = True
+        self._inventree_username = value
+
+    @inventree_password.setter
+    def inventree_password(self, value):
+        self._reinit_api = True
+        self._inventree_password = value

inventree_digikey/Digikey.py

@@ -1,14 +1,9 @@
-from digikey import product_details
+from digikey import status_salesorder_id
+import digikey # have to use digikey.product_details for monkeypatch testing
+import os
+from pathlib import Path
 
-from .ImageManager import ImageManager
-
-
-def get_part_from_part_number(partnum: str):
-    raw = product_details(partnum)
-    print(raw)
-    part = DigiPart(raw)
-    part.injest_api()
-    return part
+from .ConfigReader import ConfigReader
 
 class DigiPart:
     def __init__(self, api_value):
@@ -40,27 +35,53 @@ def injest_api(self, prompt=True):
         if prompt:
             self.prompt_part_name()
         else:
-            self.set_part_name(self.raw_value.manufacturer_part_number)
-
-
-    def set_part_name(self, name: str):
-        self.name = name
+            self.name = self.raw_value.manufacturer_part_number
 
     def prompt_part_name(self):
         found_name = self.raw_value.manufacturer_part_number
         print(f"Found {found_name} - Would you like to use this name (y/n)")
         ans = input("> ")
         if ans == "y":
-            self.set_part_name(found_name)
+            self.name = found_name
         else:
             print("Type a new name")
             name = input("> ")
-            self.set_part_name(name)
+            self.name = name
 
 
     def _extract_picture(self):
         for media in self.raw_value.media_links:
             print(media.media_type)
             if "Product Photos" in media.media_type:
                 self.picture = "%s" % media.url
-
+
+    @staticmethod
+    def _set_environment(config):
+        if config.digikey_client_id and config.digikey_client_secret and config.digikey_client_sandbox and config.digikey_storage_path:
+            os.environ['DIGIKEY_CLIENT_ID'] = config.digikey_client_id
+            os.environ['DIGIKEY_CLIENT_SECRET'] = config.digikey_client_secret
+            os.environ['DIGIKEY_CLIENT_SANDBOX'] = config.digikey_client_sandbox
+            os.environ['DIGIKEY_STORAGE_PATH'] = config.digikey_storage_path
+        else:
+            errmsg = "Cannot set environment variables for digikey module. Please set "
+            if not config.digikey_client_id:
+                errmsg += "DIGIKEY_CLIENT_ID "
+            if not config.digikey_client_secret:
+                errmsg += "DIGIKEY_CLIENT_SECRET "
+            if not config.digikey_client_sandbox:
+                errmsg += "DIGIKEY_CLIENT_SANDBOX "
+            if not config.digikey_storage_path:
+                errmsg += "DIGIKEY_STORAGE_PATH "
+
+            raise AttributeError(errmsg)
+
+    @classmethod
+    def from_digikey_part_number(cls, partnum: str, config: ConfigReader, injest_api_automatically=True, prompt=False) -> 'DigiPart':
+        cls._set_environment(config)
+        raw = digikey.product_details(partnum)
+        if injest_api_automatically:
+            part = cls(raw)
+            part.injest_api(prompt)
+            return part
+        else:
+            return cls(raw)

inventree_digikey/Inventree.py

@@ -1,53 +1,30 @@
-import configparser
-import os
-
-from inventree.api import InvenTreeAPI
 from inventree.company import SupplierPart, Company, ManufacturerPart
-from inventree.part import Part, PartCategory, Parameter, ParameterTemplate
-from pathlib import Path
+from inventree.part import Part, PartCategory
 
 from .Digikey import DigiPart
 from .ImageManager import ImageManager
+from .ConfigReader import ConfigReader
 
-if "DIGIKEY_INVENTREE_TEST_MODE" in os.environ:
-    CONFIG_FILE_PATH = os.environ["DIGIKEY_INVENTREE_TEST_CONFIG_PATH"]
-else:
-    CONFIG_FILE_PATH = Path(__file__).resolve().parent / "config.ini"
-
-API_URL = None
-USERNAME = None
-PASSWORD = None
-API = None
-
-def load_config():
-    config = configparser.ConfigParser()
-    config.read(CONFIG_FILE_PATH)
-    global API_URL, USERNAME, PASSWORD, API
-
-    API_URL = config['INVENTREE_API']['URL']
-    USERNAME = config['INVENTREE_API']['USER']
-    PASSWORD = config['INVENTREE_API']['PASSWORD']
-    API = InvenTreeAPI(API_URL, username=USERNAME, password=PASSWORD)
+def import_digikey_part(partnum: str, prompt=False):
+    dkpart = DigiPart.from_digikey_part_number(partnum, injest_api_automatically=True, prompt=prompt)
+    return add_digikey_part(dkpart)
 
-
-def add_digikey_part(dkpart: DigiPart):
-    if API is None:
-        load_config()
-    dk = get_digikey_supplier()
-    inv_part = create_inventree_part(dkpart)
+def add_digikey_part(dkpart: DigiPart, config: ConfigReader):
+    dk = get_digikey_supplier(config)
+    inv_part = create_inventree_part(dkpart, config)
     if inv_part == -1:
         return
     base_pk = int(inv_part.pk)
-    mfg = find_manufacturer(dkpart)
+    mfg = find_manufacturer(dkpart, config)
 
-    ManufacturerPart.create(API, {
+    ManufacturerPart.create(config.inventree_api, {
         'part': base_pk,
         'supplier': dk.pk,
         'MPN': dkpart.mfg_part_num,
         'manufacturer': mfg.pk
         })
 
-    SupplierPart.create(API, {
+    return SupplierPart.create(config.inventree_api, {
             "part":base_pk,
             "supplier": dk.pk,
             "SKU": dkpart.digi_part_num,
@@ -56,15 +33,12 @@ def add_digikey_part(dkpart: DigiPart):
             "link": dkpart.link
             })
 
-    return
 
 
-def get_digikey_supplier():
-    if API is None:
-        load_config()
-    dk = Company.list(API, name="Digikey")
+def get_digikey_supplier(config: ConfigReader):
+    dk = Company.list(config.inventree_api, name="Digikey")
     if len(dk) == 0:
-        dk = Company.create(API, {
+        dk = Company.create(config.inventree_api, {
             'name': 'Digikey',
             'is_supplier': True,
             'description': 'Electronics Supply Store'
@@ -73,20 +47,19 @@ def get_digikey_supplier():
     else:
         return dk[0]
 
-def create_inventree_part(dkpart: DigiPart):
-    if API is None:
-        load_config()
-    category = find_category()
-    possible_parts = Part.list(API, name=dkpart.name, description=dkpart.description)
+
+def create_inventree_part(dkpart: DigiPart, config: ConfigReader):
+    category = find_category(config)
+    possible_parts = Part.list(config.inventree_api, name=dkpart.name, description=dkpart.description)
     if len(possible_parts) > 0:
         part_names = [p.name.lower() for p in possible_parts]
         if dkpart.name.lower() in part_names:
             print("Part already exists")
             return -1
-    part = Part.create(API, {
+    part = Part.create(config.inventree_api, {
         'name': dkpart.name,
         'description': dkpart.description,
-        'category': category.pk,
+        'category': category,
         'active': True,
         'virtual': False,
         'component': True,
@@ -96,24 +69,21 @@ def create_inventree_part(dkpart: DigiPart):
     return part
 
 
-def find_category():
-    if API is None:
-        load_config()
-    categories = PartCategory.list(API)
+def find_category(config):
+    categories = PartCategory.list(config.inventree_api)
     print("="*20)
-    print("Choose a category")
+    print(f"Choose a category")
     for idx, category in enumerate(categories):
         print("\t%d %s" %(idx, category.name))
     print("="*20)
     idx = int(input("> "))
-    return categories[idx]
+    return categories[idx].pk
+
 
-def find_manufacturer(dkpart: DigiPart):
-    if API is None:
-        load_config()
-    possible_manufacturers = Company.list(API, name=dkpart.manufacturer)
+def find_manufacturer(dkpart: DigiPart, config: ConfigReader):
+    possible_manufacturers = Company.list(config.inventree_api, name=dkpart.manufacturer)
     if len(possible_manufacturers) == 0:
-        mfg = create_manufacturer(dkpart.manufacturer)
+        mfg = create_manufacturer(dkpart.manufacturer, config)
         return mfg
     else:
         print("="*20)
@@ -124,10 +94,8 @@ def find_manufacturer(dkpart: DigiPart):
         idx = int(input("> "))
         return possible_manufacturers[idx]
 
-def create_manufacturer(name: str, is_supplier: bool=False):
-    if API is None:
-        load_config()
-    mfg = Company.create(API, {
+def create_manufacturer(name: str, config: ConfigReader, is_supplier: bool=False):
+    mfg = Company.create(config.inventree_api, {
         'name': name,
         'is_manufacturer': True,
         'is_supplier': is_supplier,
@@ -139,4 +107,4 @@ def upload_picture(dkpart: DigiPart, invPart):
     if dkpart.picture is not None:
         img_file = ImageManager.get_image(dkpart.picture)
         invPart.uploadImage(img_file)
-        ImageManager.clean_cache()
+        ImageManager.clean_cache()