diff --git a/inventree/base.py b/inventree/base.py index e0a37148..c6d3cc3b 100644 --- a/inventree/base.py +++ b/inventree/base.py @@ -21,8 +21,13 @@ class InventreeObject(object): URL = "" @classmethod - def get_url(cls): - """Helper method to get the URL associated with this model.""" + def get_url(cls, api=None): + """Helper method to get the URL associated with this model. + + Arguments: + api: Optional InvenTreeAPI instance. Subclasses may override this method + to return a different URL based on the server API version. + """ url = cls.URL @@ -97,7 +102,7 @@ def __init__(self, api, pk=None, data=None): if pk <= 0: raise ValueError(f"Supplier value ({pk}) for {self.__class__} must be positive.") - url = self.get_url() + url = self.get_url(api) self._url = f"{url}/{pk}/" self._api = api @@ -147,18 +152,18 @@ def options(cls, api): cls.checkApiVersion(api) response = api.request( - cls.get_url(), + cls.get_url(api), method='OPTIONS', ) if not response.status_code == 200: - logger.error(f"OPTIONS for '{cls.URL}' returned code {response.status_code}") + logger.error(f"OPTIONS for '{cls.get_url(api)}' returned code {response.status_code}") return {} try: data = json.loads(response.text) except json.decoder.JSONDecodeError: - logger.error(f"Error decoding OPTIONS response for '{cls.URL}'") + logger.error(f"Error decoding OPTIONS response for '{cls.get_url(api)}'") return {} return data @@ -187,7 +192,7 @@ def fieldInfo(cls, field_name, api): if field_name in fields: return fields[field_name] else: - logger.warning(f"Field '{field_name}' not found in OPTIONS request for {cls.URL}") + logger.warning(f"Field '{field_name}' not found in OPTIONS request for {cls.get_url(api)}") return {} @classmethod @@ -208,7 +213,7 @@ def create(cls, api, data, **kwargs): if cls.getPkField() in data.keys(): data.pop(cls.getPkField()) - response = api.post(cls.URL, data, **kwargs) + response = api.post(cls.get_url(api), data, **kwargs) if response is None: logger.error("Error creating new object") @@ -225,7 +230,7 @@ def count(cls, api, **kwargs): # By limiting to a single result, we perform a fast query, but get a total number of results params['limit'] = 1 - response = api.get(url=cls.URL, params=params) + response = api.get(url=cls.get_url(api), params=params) return response['count'] @@ -243,7 +248,7 @@ def list(cls, api, **kwargs): if 'url' in kwargs: url = kwargs.pop('url') else: - url = cls.URL + url = cls.get_url(api) try: response = api.get(url=url, params=kwargs) @@ -423,7 +428,7 @@ def bulkDelete(cls, api: inventree_api.InvenTreeAPI, items=None, filters=None): data['filters'] = filters return api.delete( - cls.URL, + cls.get_url(api), json=data, ) diff --git a/inventree/part.py b/inventree/part.py index a2de68fe..03a5f7e8 100644 --- a/inventree/part.py +++ b/inventree/part.py @@ -10,6 +10,9 @@ import inventree.report import inventree.stock +# Pricing models were moved to inventree.pricing - re-exported here for backwards compatibility +from inventree.pricing import InternalPrice, PartPricing, SalePrice # noqa: F401 + logger = logging.getLogger('inventree') @@ -133,6 +136,8 @@ def getInternalPriceList(self): Returns the InternalPrice list for this part """ + + return InternalPrice.list(self._api, part=self.pk) def setInternalPrice(self, quantity: int, price: float): @@ -148,6 +153,12 @@ def getSalePrice(self): """ return SalePrice.list(self._api, part=self.pk)[0].price + def getPricing(self): + """ + Get the cached pricing data for this part + """ + return PartPricing(self._api, part=self.pk) + def getRequirements(self): """ Get required amounts from requirements API endpoint for this part @@ -210,49 +221,6 @@ class BomItemSubstitute( URL = "bom/substitute/" -class InternalPrice(inventree.base.InventreeObject): - """ Class representing the InternalPrice model """ - - URL = 'part/internal-price/' - - @classmethod - def setInternalPrice(cls, api, part, quantity: int, price: float): - """ - Set the internal price for this part - """ - - data = { - 'part': part, - 'quantity': quantity, - 'price': price, - } - - # Send the data to the server - return api.post(cls.URL, data) - - -class SalePrice(inventree.base.InventreeObject): - """ Class representing the SalePrice model """ - - URL = 'part/sale-price/' - - @classmethod - def setSalePrice(cls, api, part, quantity: int, price: float, price_currency: str): - """ - Set the sale price for this part - """ - - data = { - 'part': part, - 'quantity': quantity, - 'price': price, - 'price_currency': price_currency, - } - - # Send the data to the server - return api.post(cls.URL, data) - - class PartRelated(inventree.base.InventreeObject): """ Class representing a relationship between parts""" diff --git a/inventree/pricing.py b/inventree/pricing.py new file mode 100644 index 00000000..94e0aabc --- /dev/null +++ b/inventree/pricing.py @@ -0,0 +1,191 @@ +# -*- coding: utf-8 -*- +"""Pricing models. + +Ref: https://github.com/inventree/InvenTree/pull/13034 +""" + +import inventree.base + + +class PricingModelMixin: + """Mixin class for models whose API endpoints moved to the 'pricing' app. + + - For servers with API version >= PRICING_API_VERSION, the new URL is used + - For older servers, the LEGACY_URL is used + """ + + # API version at which pricing endpoints moved from /api/part/ to /api/pricing/ + PRICING_API_VERSION = 561 + + # URL for the endpoint on older servers + LEGACY_URL = "" + + @classmethod + def useLegacyUrl(cls, api=None): + """Return True if the server requires the legacy (pre-pricing app) URL. + + If the API version is unknown, the new URL is assumed. + """ + + api_version = getattr(api, 'api_version', None) + + return api_version is not None and api_version < cls.PRICING_API_VERSION + + @classmethod + def get_url(cls, api=None): + """Return the URL for this model, based on the server API version.""" + + if cls.useLegacyUrl(api): + url = cls.LEGACY_URL + + if not url.endswith('/'): + url += '/' + + return url + + return super().get_url(api) + + +class InternalPrice(PricingModelMixin, inventree.base.InventreeObject): + """ Class representing the InternalPrice model """ + + URL = 'pricing/internal-price/' + LEGACY_URL = 'part/internal-price/' + + @classmethod + def setInternalPrice(cls, api, part, quantity: int, price: float): + """ + Set the internal price for this part + """ + + data = { + 'part': part, + 'quantity': quantity, + 'price': price, + } + + # Send the data to the server + return api.post(cls.get_url(api), data) + + +class SalePrice(PricingModelMixin, inventree.base.InventreeObject): + """ Class representing the SalePrice model """ + + URL = 'pricing/sale-price/' + LEGACY_URL = 'part/sale-price/' + + @classmethod + def setSalePrice(cls, api, part, quantity: int, price: float, price_currency: str): + """ + Set the sale price for this part + """ + + data = { + 'part': part, + 'quantity': quantity, + 'price': price, + 'price_currency': price_currency, + } + + # Send the data to the server + return api.post(cls.get_url(api), data) + + +class PartPricing(PricingModelMixin, inventree.base.InventreeObject): + """Class representing the cached pricing data for a particular Part. + + Unlike other models, this is accessed via the primary key of the linked Part: + + - API version >= 561: pricing/part// + - Older API versions: part//pricing/ + + There is no list endpoint, and pricing data cannot be created or deleted. + Writable fields (e.g. override_min, override_max) can be changed via save(). + """ + + URL = 'pricing/part/' + LEGACY_URL = 'part/' + + def __init__(self, api, part, data=None): + """Instantiate the PartPricing object for a given Part. + + Args: + api - The request manager object + part - The Part instance (or primary key value) to which this pricing data belongs + data - JSON representation of the pricing data + """ + + self.checkApiVersion(api) + + part = getattr(part, 'pk', part) + + try: + part = int(str(part).strip()) + except Exception: + raise TypeError(f"Invalid part value '{part}' for {self.__class__}") + + if part <= 0: + raise ValueError(f"Supplied part value ({part}) for {self.__class__} must be positive.") + + self._part = part + self._url = self.get_detail_url(api, part) + self._api = api + + if data is None: + data = {} + + self._data = data + + # If the data are not populated, fetch from server + if len(self._data) == 0: + self.reload() + + @classmethod + def get_detail_url(cls, api, part: int): + """Return the pricing URL for the given Part, based on the server API version.""" + + if cls.useLegacyUrl(api): + return f"{cls.LEGACY_URL}{part}/pricing/" + + return f"{cls.URL}{part}/" + + @classmethod + def getPkField(cls): + """Pricing data is identified by the linked Part.""" + return 'part' + + def getPkValue(self): + """Return the primary key of the linked Part.""" + return self._part + + def updatePricing(self): + """Request that the server recalculate pricing data for this part. + + Note: The server may perform the update in the background, + in which case the returned data may not yet reflect the new pricing. + """ + + return self.save(data={'update': True}) + + @classmethod + def _unsupported(cls, action): + raise NotImplementedError(f"'{action}' is not supported for {cls.__name__} - use Part.getPricing() instead") + + @classmethod + def options(cls, api): + cls._unsupported('options') + + @classmethod + def create(cls, api, data, **kwargs): + cls._unsupported('create') + + @classmethod + def count(cls, api, **kwargs): + cls._unsupported('count') + + @classmethod + def list(cls, api, **kwargs): + cls._unsupported('list') + + def delete(self): + self._unsupported('delete') diff --git a/test/test_internal_price.py b/test/test_internal_price.py deleted file mode 100644 index e4d3af7d..00000000 --- a/test/test_internal_price.py +++ /dev/null @@ -1,51 +0,0 @@ -# -*- coding: utf-8 -*- - -import os -import sys - -sys.path.append(os.path.abspath(os.path.dirname(__file__))) - -from test_api import InvenTreeTestCase # noqa: E402 - -from inventree.part import InternalPrice, Part # noqa: E402 - - -class InternalPriceTest(InvenTreeTestCase): - """ Test that the InternalPrice related objects can be managed via the API """ - - def test_fields(self): - """ - Test field names via OPTIONS request - """ - - field_names = InternalPrice.fieldNames(self.api) - - for field in [ - 'part', - 'quantity', - 'price' - ]: - self.assertIn(field, field_names) - - def test_internal_price_create(self): - """ - Tests the ability to create an internal price - """ - p = Part.create(self.api, { - 'name': 'Test Part', - 'description': 'Test Part', - 'category': 1, - 'revision': 1, - 'active': True, - }) - - self.assertIsNotNone(p) - self.assertIsNotNone(p.pk) - - ip = InternalPrice.create(self.api, { - 'part': p.pk, - 'quantity': 1, - 'price': '1.00' - }) - - self.assertIsNotNone(ip) diff --git a/test/test_pricing.py b/test/test_pricing.py new file mode 100644 index 00000000..74ac0d20 --- /dev/null +++ b/test/test_pricing.py @@ -0,0 +1,239 @@ +# -*- coding: utf-8 -*- + +import os +import sys +import unittest + +sys.path.append(os.path.abspath(os.path.dirname(__file__))) + +from test_api import InvenTreeTestCase # noqa: E402 + +import inventree.part # noqa: E402 +from inventree.part import Part # noqa: E402 +from inventree.pricing import InternalPrice, PartPricing, SalePrice # noqa: E402 + + +class DummyAPI: + """Minimal stand-in for InvenTreeAPI, providing only the API version""" + + def __init__(self, api_version): + self.api_version = api_version + + +class PricingURLTests(unittest.TestCase): + """Test that pricing model URLs are determined by the server API version""" + + def test_backwards_compatible_imports(self): + """Pricing models can still be imported from inventree.part""" + + self.assertIs(inventree.part.InternalPrice, InternalPrice) + self.assertIs(inventree.part.SalePrice, SalePrice) + self.assertIs(inventree.part.PartPricing, PartPricing) + + def test_urls(self): + """Legacy URLs are used for older servers, new URLs otherwise""" + + expected = { + InternalPrice: ('part/internal-price/', 'pricing/internal-price/'), + SalePrice: ('part/sale-price/', 'pricing/sale-price/'), + } + + for cls, (legacy_url, new_url) in expected.items(): + version = cls.PRICING_API_VERSION + + self.assertEqual(cls.get_url(DummyAPI(version - 1)), legacy_url) + self.assertEqual(cls.get_url(DummyAPI(version)), new_url) + self.assertEqual(cls.get_url(DummyAPI(version + 1)), new_url) + + # No API (or unknown version) defaults to the new URL + self.assertEqual(cls.get_url(), new_url) + self.assertEqual(cls.get_url(None), new_url) + + def test_instance_url(self): + """Instance URL is constructed from the version-specific base URL""" + + data = {'pk': 3, 'part': 1, 'quantity': 1, 'price': '1.00'} + + legacy = InternalPrice(DummyAPI(InternalPrice.PRICING_API_VERSION - 1), data=data) + self.assertTrue(legacy._url.startswith('part/internal-price/')) + + new = InternalPrice(DummyAPI(InternalPrice.PRICING_API_VERSION), data=data) + self.assertTrue(new._url.startswith('pricing/internal-price/')) + + +class PartPricingOfflineTests(unittest.TestCase): + """Unit tests for the PartPricing class which do not require a server""" + + DATA = {'overall_min': '1.00', 'overall_max': '2.00'} + + def test_detail_url(self): + """Pricing URL is determined by the server API version""" + + version = PartPricing.PRICING_API_VERSION + + self.assertEqual(PartPricing.get_detail_url(DummyAPI(version - 1), 7), 'part/7/pricing/') + self.assertEqual(PartPricing.get_detail_url(DummyAPI(version), 7), 'pricing/part/7/') + self.assertEqual(PartPricing.get_detail_url(None, 7), 'pricing/part/7/') + + legacy = PartPricing(DummyAPI(version - 1), part=7, data=self.DATA) + self.assertEqual(legacy._url, 'part/7/pricing/') + + new = PartPricing(DummyAPI(version), part=7, data=self.DATA) + self.assertEqual(new._url, 'pricing/part/7/') + + def test_part_reference(self): + """PartPricing is identified by the linked part""" + + class DummyPart: + pk = 12 + + for part in [12, '12', DummyPart()]: + pricing = PartPricing(None, part=part, data=self.DATA) + self.assertEqual(pricing.pk, 12) + self.assertTrue(pricing.is_valid()) + self.assertEqual(pricing.overall_min, '1.00') + + with self.assertRaises(TypeError): + PartPricing(None, part='abc', data=self.DATA) + + for part in [0, -1]: + with self.assertRaises(ValueError): + PartPricing(None, part=part, data=self.DATA) + + def test_unsupported(self): + """Operations not supported by the pricing endpoint raise an error""" + + api = DummyAPI(PartPricing.PRICING_API_VERSION) + + with self.assertRaises(NotImplementedError): + PartPricing.list(api) + + with self.assertRaises(NotImplementedError): + PartPricing.count(api) + + with self.assertRaises(NotImplementedError): + PartPricing.create(api, {}) + + with self.assertRaises(NotImplementedError): + PartPricing.options(api) + + with self.assertRaises(NotImplementedError): + PartPricing(api, part=1, data=self.DATA).delete() + + +class InternalPriceTest(InvenTreeTestCase): + """ Test that the InternalPrice related objects can be managed via the API """ + + def test_fields(self): + """ + Test field names via OPTIONS request + """ + + field_names = InternalPrice.fieldNames(self.api) + + for field in [ + 'part', + 'quantity', + 'price' + ]: + self.assertIn(field, field_names) + + def test_internal_price_create(self): + """ + Tests the ability to create an internal price + """ + p = Part.create(self.api, { + 'name': 'Test Part', + 'description': 'Test Part', + 'category': 1, + 'revision': 1, + 'active': True, + }) + + self.assertIsNotNone(p) + self.assertIsNotNone(p.pk) + + ip = InternalPrice.create(self.api, { + 'part': p.pk, + 'quantity': 1, + 'price': '1.00' + }) + + self.assertIsNotNone(ip) + + +class SalePriceTest(InvenTreeTestCase): + """Test that the SalePrice related objects can be managed via the API""" + + def test_fields(self): + """Test field names via OPTIONS request""" + + field_names = SalePrice.fieldNames(self.api) + + for field in [ + 'part', + 'quantity', + 'price', + 'price_currency', + ]: + self.assertIn(field, field_names) + + def test_set_sale_price(self): + """Test that a sale price can be set and retrieved for a part""" + + p = Part.list(self.api, salable=True)[0] + + # Delete any existing sale prices + for price in SalePrice.list(self.api, part=p.pk): + self.assertEqual(type(price), SalePrice) + price.delete() + + self.assertEqual(len(SalePrice.list(self.api, part=p.pk)), 0) + + SalePrice.setSalePrice(self.api, p.pk, 1, 12.5, 'USD') + + prices = SalePrice.list(self.api, part=p.pk) + self.assertEqual(len(prices), 1) + self.assertEqual(prices[0].part, p.pk) + self.assertEqual(float(prices[0].price), 12.5) + self.assertEqual(float(p.getSalePrice()), 12.5) + + +class PartPricingTest(InvenTreeTestCase): + """Test that PartPricing data can be accessed via the API""" + + def test_get_pricing(self): + """Pricing data can be retrieved for a part""" + + p = Part.list(self.api)[0] + + pricing = p.getPricing() + + self.assertEqual(type(pricing), PartPricing) + self.assertEqual(pricing.pk, p.pk) + self.assertTrue(pricing.is_valid()) + + for field in [ + 'overall_min', + 'overall_max', + 'internal_cost_min', + 'internal_cost_max', + 'override_min', + 'override_max', + ]: + self.assertIn(field, pricing) + + def test_update_pricing(self): + """A pricing update can be requested for a part""" + + p = Part.list(self.api)[0] + + pricing = PartPricing(self.api, part=p) + + response = pricing.updatePricing() + + self.assertIsNotNone(response) + self.assertTrue(pricing.is_valid()) + + pricing.reload() + self.assertTrue(pricing.is_valid())