Pricing role (#13034)

* add 'pricing' role

* Create new 'pricing' app

- migrate existing pricing models

* bug fixes

* Move API endpoints to /api/pricing/

* Adjust target branch for playwright tests

* Fix permission checks for nested OptionalField entries

* Migrate existing background tasks

* Point to master branch demo data

---------

Co-authored-by: Matthias Mair <code@mjmair.com>
This commit is contained in:
OliverandMatthias Mair authored and GitHub committed 2026-10-21 15:00:50 +11:00
1 parent a91f079530
commit 4c948cda1b
43 files changed
+3175 -1388

No files matched your search

+1
View File
@@ -33,6 +33,7 @@ API schema documentation is split into the following categories:
| [External Order Management](./schema/order.md) | External Order Management |
| [Parts and Part Categories](./schema/part.md) | Parts and Part Categories |
| [Plugin Functionality](./schema/plugins.md) | Plugin Functionality |
| [Pricing Information](./schema/pricing.md) | Part Pricing |
| [Report Generation](./schema/report.md) | Report Generation |
| [Settings Management](./schema/settings.md) | Settings Management |
| [Stock and Stock Locations](./schema/stock.md) | Stock and Stock Locations |
+7
View File
@@ -0,0 +1,7 @@
---
title: Part Pricing
---
The *Part Pricing* section of the InvenTree API schema is documented below.
[OAD(./docs/docs/api/schema/pricing.yml)]
+27
View File
@@ -29,6 +29,33 @@ Additionally, the following information is stored for each part, in relation to
| Sale Price | How much a salable item is sold for (with price-breaks) | [Part](../part/index.md) |
| Sale Cost | How much an item was sold for | [Sales Order](../sales/sales_order.md) |
## Pricing Permissions
Access to part pricing data is controlled by the dedicated *Pricing* [user role](../settings/permissions.md#roles). This role is separate from the *Parts* role, allowing pricing information to be restricted (or granted) independently of general part data. For example, a group may be allowed to view and edit parts, without being able to see what those parts cost.
The *Pricing* role covers the following data:
| Data | Description |
| --- | --- |
| Cached Pricing | The calculated [pricing overview](#pricing-overview) data for each part, including any [pricing overrides](#override-pricing) |
| Sale Price Breaks | The [sale pricing](#sale-pricing) defined for a salable part |
| Internal Price Breaks | The [internal pricing](#internal-pricing) defined for a part |
The permissions within the *Pricing* role apply as follows:
| Permission | Effect |
| --- | --- |
| **View** | View the *Pricing* tab on the part detail page, and see pricing information (e.g. minimum / maximum price) displayed against parts and BOM items |
| **Change** | Edit pricing overrides, manually [refresh](#manual-updates) pricing data, and edit existing price breaks |
| **Add** | Create new sale and internal price breaks |
| **Delete** | Delete sale and internal price breaks |
If a user does not have *view* permission for the *Pricing* role, the *Pricing* tab is hidden on the part detail page, and cached pricing fields are omitted from part and BOM data returned via the API.
!!! info "Other Pricing Sources"
Some of the data used to calculate part pricing is covered by other roles. [Supplier price breaks](../purchasing/supplier.md#supplier-part-pricing) are covered by the *Purchase Orders* role. Within the *Pricing* tab, the *Purchase History* and *Supplier Pricing* sections are only displayed if the user also has *view* permission for the *Purchase Orders* role, and the *Sale Pricing* and *Sale History* sections are only displayed if the user also has *view* permission for the *Sales Orders* role.
## Pricing Tab
The pricing tab for a given Part provides all available pricing information for that part. It shows all price ranges and provides tools to calculate them.
+3
View File
@@ -27,6 +27,9 @@ InvenTree functionality is split into a number of distinct roles. A group will h
{{ roles() }}
!!! info "Pricing Role"
Access to part pricing data is controlled separately from general part data, via the *Pricing* role. Refer to the [pricing permissions](../part/pricing.md#pricing-permissions) documentation for more information.
{{ image("admin/roles.png", "Roles") }}
### Role Permissions
+1
View File
@@ -25,6 +25,7 @@ SPECIAL_PATHS = {
'order': 'External Order Management',
'part': 'Parts and Part Categories',
'plugins': 'Plugin Functionality',
'pricing': 'Part Pricing',
'report': 'Report Generation',
'settings': 'Settings Management',
'stock': 'Stock and Stock Locations',
+49 -1
View File
@@ -8,9 +8,11 @@ from django.conf import settings
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ObjectDoesNotExist
from django.db import transaction
from django.http import HttpRequest, JsonResponse
from django.http import HttpRequest, HttpResponsePermanentRedirect, JsonResponse
from django.urls import path, reverse
from django.utils.decorators import method_decorator
from django.utils.translation import gettext_lazy as _
from django.views.decorators.csrf import csrf_exempt
from django.views.generic.base import RedirectView
import structlog
@@ -1009,6 +1011,52 @@ class MetadataRedirectView(RedirectView):
return reverse('api-generic-metadata', args=args, kwargs=_kwargs)
class HttpResponsePermanentRedirectPreserveMethod(HttpResponsePermanentRedirect):
"""Permanent redirect which preserves the request method and body (HTTP 308).
Unlike a 301 redirect, clients must not change a POST / PUT / PATCH / DELETE
request into a GET request when following a 308 redirect.
"""
status_code = 308
@method_decorator(csrf_exempt, name='dispatch')
class MovedEndpointRedirectView(RedirectView):
"""Redirect requests for an API endpoint which has moved to a new URL.
- The redirect target is resolved from the named URL pattern of the new endpoint
- The request method and body are preserved (HTTP 308), so that write operations
against the old endpoint continue to work
- Any query parameters are passed through to the new endpoint
CSRF protection is not applied here, as this view does not perform any action itself.
The target endpoint applies the usual authentication, permission and CSRF checks.
"""
query_string = True
def get(self, request, *args, **kwargs):
"""Return a method-preserving redirect to the new endpoint."""
url = self.get_redirect_url(*args, **kwargs)
return HttpResponsePermanentRedirectPreserveMethod(url)
# All request methods are redirected in the same way
post = put = patch = delete = options = head = get
def moved_path(route: str, pattern_name: str):
"""Construct a path which redirects from a legacy API endpoint to its new location.
Arguments:
route: The legacy URL route (relative to the including URL pattern)
pattern_name: The name of the URL pattern for the new endpoint
Note: Any URL kwargs captured by the legacy route are passed to the new URL pattern.
"""
return path(route, MovedEndpointRedirectView.as_view(pattern_name=pattern_name))
def meta_path(model, lookup_field: str = 'pk', lookup_field_ref: str = 'pk'):
"""Helper function for constructing metadata path for a given model.
+12 -1
View File
@@ -1,11 +1,22 @@
"""InvenTree API version information."""
# InvenTree API version
INVENTREE_API_VERSION = 561
INVENTREE_API_VERSION = 562
"""Increment this API version number whenever there is a significant change to the API that any clients need to know about."""
INVENTREE_API_TEXT = """
v562 -> 2026-10-21 : https://github.com/inventree/InvenTree/pull/13034
- Adds new "pricing" user role, covering the PartPricing, PartSellPriceBreak and PartInternalPriceBreak models
- The "part" role no longer grants access to part pricing data
- Cached pricing fields (pricing_min, pricing_max, etc) are omitted from Part and BomItem API responses for users without the "pricing" view permission
- This also applies to nested (embedded) part data, e.g. "part_detail" and "sub_part_detail" fields
- Moves part pricing endpoints to the new /api/pricing/ namespace:
- /api/part/sale-price/ -> /api/pricing/sale-price/
- /api/part/internal-price/ -> /api/pricing/internal-price/
- /api/part/<id>/pricing/ -> /api/pricing/part/<id>/
- The legacy endpoints redirect (HTTP 308, preserving the request method and body) to the new endpoints
v561 -> 2026-10-21 : https://github.com/inventree/InvenTree/pull/13046
- Fix ordering of social applications in API schema
+32 -2
View File
@@ -91,8 +91,17 @@ class FilterableSerializerMixin:
optional_fields: set = None
filter_on_query: bool = True
# Keyword argument used to pass the requesting user to nested serializers
PERMISSION_USER_KWARG = 'permission_user'
def __init__(self, *args, **kwargs):
"""Initialization routine for the serializer. This gathers and applies filters through kwargs."""
# The requesting user, as provided by a parent serializer (if this serializer is nested)
# Note: This is used *only* for permission checks against optional fields.
# A nested serializer does not receive the request itself,
# so that query parameters are only applied to the top-level serializer.
self.permission_user = kwargs.pop(self.PERMISSION_USER_KWARG, None)
# Extract some useful context information for later use
context = kwargs.get('context', {})
self.request = context.get('request', None) or getattr(self, 'request', None)
@@ -237,7 +246,7 @@ class FilterableSerializerMixin:
if model in (get_user_model(), Group, Owner, RuleSet):
return included
user = getattr(self.request, 'user', None)
user = self.get_permission_user()
if user is None:
return included
@@ -319,12 +328,33 @@ class FilterableSerializerMixin:
return field_names
def get_permission_user(self):
"""Return the user against which optional field permissions are checked.
This is the user associated with the request (for a top-level serializer),
or the user provided by the parent serializer (for a nested serializer).
"""
if user := getattr(self.request, 'user', None):
return user
return self.permission_user
def build_optional_field(self, field_name: str):
"""Build an optional field, based on the provided field name."""
field = getattr(self, field_name, None)
if field and isinstance(field, OptionalField):
serializer_kwargs = {**field.serializer_kwargs} or {}
serializer_kwargs = {**(field.serializer_kwargs or {})}
# Pass the requesting user through to any nested filterable serializer,
# so that it can check permissions for its own optional fields
if isinstance(field.serializer_class, type) and issubclass(
field.serializer_class, FilterableSerializerMixin
):
serializer_kwargs[self.PERMISSION_USER_KWARG] = (
self.get_permission_user()
)
return field.serializer_class, serializer_kwargs
def build_relational_field(self, field_name, relation_info):
@@ -324,6 +324,7 @@ INSTALLED_APPS = [
'machine.apps.MachineConfig',
'data_exporter.apps.DataExporterConfig',
'importer.apps.ImporterConfig',
'pricing.apps.PricingConfig',
'web',
'generic',
'InvenTree.apps.InvenTreeConfig', # InvenTree app runs last
+2
View File
@@ -25,6 +25,7 @@ import machine.api
import order.api
import part.api
import plugin.api
import pricing.api
import report.api
import scim.api
import stock.api
@@ -62,6 +63,7 @@ apipatterns = [
path('machine/', include(machine.api.machine_api_urls)),
path('order/', include(order.api.order_api_urls)),
path('part/', include(part.api.part_api_urls)),
path('pricing/', include(pricing.api.pricing_api_urls)),
path('report/', include(report.api.report_api_urls)),
path('search/', APISearchView.as_view(), name='api-search'),
path('settings/', include(common.api.settings_api_urls)),
+3 -1
View File
@@ -780,7 +780,9 @@ class BuildOutputCreate(BuildOrderContextMixin, CreateAPI):
outputs = serializer.save()
queryset = stock.serializers.StockItemSerializer.annotate_queryset(outputs)
response = stock.serializers.StockItemSerializer(queryset, many=True)
response = stock.serializers.StockItemSerializer(
queryset, many=True, permission_user=request.user
)
# Return the created outputs
return Response(response.data, status=status.HTTP_201_CREATED)
+6 -2
View File
@@ -498,7 +498,9 @@ class PurchaseOrderViewSet(
items = serializer.save()
queryset = stock_serializers.StockItemSerializer.annotate_queryset(items)
response = stock_serializers.StockItemSerializer(queryset, many=True)
response = stock_serializers.StockItemSerializer(
queryset, many=True, permission_user=request.user
)
return Response(response.data, status=status.HTTP_201_CREATED)
@@ -681,7 +683,9 @@ class PurchaseOrderLineItemViewSet(
):
line_item.update_pricing()
serializer = serializers.PurchaseOrderLineItemSerializer(line_item)
serializer = serializers.PurchaseOrderLineItemSerializer(
line_item, permission_user=request.user
)
headers = self.get_success_headers(serializer.data)
return Response(
serializer.data, status=status.HTTP_201_CREATED, headers=headers
-43
View File
@@ -32,17 +32,6 @@ class PartAdmin(admin.ModelAdmin):
inlines = []
@admin.register(models.PartPricing)
class PartPricingAdmin(admin.ModelAdmin):
"""Admin class for PartPricing model."""
list_display = ('part', 'overall_min', 'overall_max')
search_fields = ['part__name', 'part__IPN', 'part__description']
autocomplete_fields = ['part']
@admin.register(models.PartStocktake)
class PartStocktakeAdmin(admin.ModelAdmin):
"""Admin class for PartStocktake model."""
@@ -98,35 +87,3 @@ class BomItemAdmin(admin.ModelAdmin):
)
autocomplete_fields = ('part', 'sub_part')
@admin.register(models.PartSellPriceBreak)
class PartSellPriceBreakAdmin(admin.ModelAdmin):
"""Admin class for the PartSellPriceBreak model."""
class Meta:
"""Metaclass options."""
model = models.PartSellPriceBreak
list_display = ('part', 'quantity', 'price')
search_fields = ['part__name', 'part__IPN']
autocomplete_fields = ('part',)
@admin.register(models.PartInternalPriceBreak)
class PartInternalPriceBreakAdmin(admin.ModelAdmin):
"""Admin class for the PartInternalPriceBreak model."""
class Meta:
"""Metaclass options."""
model = models.PartInternalPriceBreak
list_display = ('part', 'quantity', 'price')
search_fields = ['part__name', 'part__IPN']
autocomplete_fields = ('part',)
+8 -86
View File
@@ -23,6 +23,7 @@ from InvenTree.api import (
ParameterListMixin,
TreeMixin,
meta_path,
moved_path,
)
from InvenTree.fields import InvenTreeOutputOption, OutputConfiguration
from InvenTree.filters import (
@@ -56,9 +57,7 @@ from .models import (
Part,
PartCategory,
PartCategoryParameterTemplate,
PartInternalPriceBreak,
PartRelated,
PartSellPriceBreak,
PartStocktake,
PartTestTemplate,
)
@@ -368,44 +367,6 @@ class CategoryParameterDetail(RetrieveUpdateDestroyAPI):
serializer_class = part_serializers.CategoryParameterTemplateSerializer
class PartSalePriceDetail(RetrieveUpdateDestroyAPI):
"""Detail endpoint for PartSellPriceBreak model."""
queryset = PartSellPriceBreak.objects.all()
serializer_class = part_serializers.PartSalePriceSerializer
class PartSalePriceList(DataExportViewMixin, ListCreateAPI):
"""API endpoint for list view of PartSalePriceBreak model."""
queryset = PartSellPriceBreak.objects.all()
serializer_class = part_serializers.PartSalePriceSerializer
filter_backends = SEARCH_ORDER_FILTER
filterset_fields = ['part']
ordering_fields = ['quantity', 'price']
ordering = 'quantity'
class PartInternalPriceDetail(RetrieveUpdateDestroyAPI):
"""Detail endpoint for PartInternalPriceBreak model."""
queryset = PartInternalPriceBreak.objects.all()
serializer_class = part_serializers.PartInternalPriceSerializer
class PartInternalPriceList(DataExportViewMixin, ListCreateAPI):
"""API endpoint for list view of PartInternalPriceBreak model."""
queryset = PartInternalPriceBreak.objects.all()
serializer_class = part_serializers.PartInternalPriceSerializer
filter_backends = SEARCH_ORDER_FILTER
filterset_fields = ['part']
ordering_fields = ['quantity', 'price']
ordering = 'quantity'
class PartTestTemplateFilter(FilterSet):
"""Custom filterset class for the PartTestTemplateList endpoint."""
@@ -567,25 +528,6 @@ class PartRequirements(RetrieveAPI):
serializer_class = part_serializers.PartRequirementsSerializer
class PartPricingDetail(RetrieveUpdateAPI):
"""API endpoint for viewing part pricing data."""
serializer_class = part_serializers.PartPricingSerializer
queryset = Part.objects.all().select_related('pricing_data')
def get_object(self):
"""Return the PartPricing object associated with the linked Part."""
part = super().get_object()
return part.pricing
def _get_serializer(self, *args, **kwargs):
"""Return a part pricing serializer object."""
part = self.get_object()
kwargs['instance'] = part.pricing
return self.serializer_class(**kwargs)
class PartSerialNumberDetail(RetrieveAPI):
"""API endpoint for returning extra serial number information about a particular part."""
@@ -1637,32 +1579,11 @@ part_api_urls = [
),
]),
),
# Base URL for part sale pricing
path(
'sale-price/',
include([
path(
'<int:pk>/',
PartSalePriceDetail.as_view(),
name='api-part-sale-price-detail',
),
path('', PartSalePriceList.as_view(), name='api-part-sale-price-list'),
]),
),
# Base URL for part internal pricing
path(
'internal-price/',
include([
path(
'<int:pk>/',
PartInternalPriceDetail.as_view(),
name='api-part-internal-price-detail',
),
path(
'', PartInternalPriceList.as_view(), name='api-part-internal-price-list'
),
]),
),
# Legacy URLs for part pricing (moved to the 'pricing' app)
moved_path('sale-price/<int:pk>/', 'api-part-sale-price-detail'),
moved_path('sale-price/', 'api-part-sale-price-list'),
moved_path('internal-price/<int:pk>/', 'api-part-internal-price-detail'),
moved_path('internal-price/', 'api-part-internal-price-list'),
# Base URL for PartRelated API endpoints
path(
'related/',
@@ -1728,7 +1649,8 @@ part_api_urls = [
# Part metadata
meta_path(Part),
# Part pricing
path('pricing/', PartPricingDetail.as_view(), name='api-part-pricing'),
# Legacy URL for part pricing (moved to the 'pricing' app)
moved_path('pricing/', 'api-part-pricing'),
# Part detail endpoint
path('', PartDetail.as_view(), name='api-part-detail'),
]),
+1 -1
View File
@@ -61,7 +61,7 @@ class PartConfig(AppConfig):
Prevents issues with state machine if the server is restarted mid-update
"""
from .models import PartPricing
from pricing.models import PartPricing
if InvenTree.ready.isImportingData():
return
@@ -0,0 +1,27 @@
# Generated by Django 5.2.17 on 2026-10-17 02:26
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [('part', '0154_remove_part_notes')]
# The PartPricing, PartSellPriceBreak and PartInternalPriceBreak models
# have been moved to the 'pricing' app.
# The underlying database tables are retained (see pricing.0001_initial),
# so these operations only affect the migration state, not the database.
operations = [
migrations.SeparateDatabaseAndState(
state_operations=[
migrations.RemoveField(model_name='partpricing', name='part'),
migrations.AlterUniqueTogether(
name='partsellpricebreak', unique_together=None
),
migrations.RemoveField(model_name='partsellpricebreak', name='part'),
migrations.DeleteModel(name='PartInternalPriceBreak'),
migrations.DeleteModel(name='PartPricing'),
migrations.DeleteModel(name='PartSellPriceBreak'),
],
database_operations=[],
)
]
+4 -869
View File
@@ -7,7 +7,6 @@ import inspect
import math
import os
import re
from datetime import timedelta
from decimal import ROUND_HALF_UP, Decimal, InvalidOperation
from typing import Optional, TypedDict, cast
@@ -27,13 +26,9 @@ from django.utils.translation import gettext_lazy as _
import structlog
from django_cleanup import cleanup
from djmoney.contrib.exchange.exceptions import MissingRate
from djmoney.contrib.exchange.models import convert_money
from djmoney.money import Money
from mptt.managers import TreeManager
from mptt.models import TreeForeignKey
import common.currency
import common.models
import company.models
import InvenTree.conversion
@@ -48,17 +43,14 @@ import report.mixins
import users.models
from build import models as BuildModels
from build.status_codes import BuildStatusGroups
from common.currency import currency_code_default
from common.icons import validate_icon
from common.settings import get_global_setting
from InvenTree import helpers, validators
from InvenTree.cache import get_session_cache, set_session_cache
from InvenTree.exceptions import log_error
from InvenTree.fields import InvenTreeURLField
from InvenTree.helpers import decimal2string, normalize
from order import models as OrderModels
from order.status_codes import (
PurchaseOrderStatus,
PurchaseOrderStatusGroups,
SalesOrderStatusGroups,
TransferOrderStatusGroups,
@@ -2212,6 +2204,8 @@ class Part(
If there is no PartPricing database entry defined for this Part,
it will first be created, and then returned.
"""
from pricing.models import PartPricing
try:
pricing = PartPricing.objects.get(part=self)
except PartPricing.DoesNotExist:
@@ -2618,804 +2612,6 @@ def after_save_part(sender, instance: Part, created, **kwargs):
)
class PartPricing(common.models.MetaMixin):
"""Model for caching min/max pricing information for a particular Part.
It is prohibitively expensive to calculate min/max pricing for a part "on the fly".
As min/max pricing does not change very often, we pre-calculate and cache these values.
Whenever pricing is updated, these values are re-calculated and stored.
Pricing information is cached for:
- BOM cost (min / max cost of component items)
- Purchase cost (based on purchase history)
- Internal cost (based on user-specified InternalPriceBreak data)
- Supplier price (based on supplier part data)
- Variant price (min / max cost of any variants)
- Overall best / worst (based on the values listed above)
- Sale price break min / max values
- Historical sale pricing min / max values
Note that this pricing information does not take "quantity" into account:
- This provides a simple min / max pricing range, which is quite valuable in a lot of situations
- Quantity pricing still needs to be calculated
- Quantity pricing can be viewed from the part detail page
- Detailed pricing information is very context specific in any case
"""
# When calculating assembly pricing, we limit the depth of the calculation
MAX_PRICING_DEPTH = 50
@property
def is_valid(self):
"""Return True if the cached pricing is valid."""
return self.updated is not None
def get_part(self):
"""Return the Part instance associated with this pricing object.
If the part does not exist (i.e. has been deleted), return None.
"""
from part.models import Part
try:
part = self.part
part.refresh_from_db()
return part
except (Part.DoesNotExist, IntegrityError):
return None
def convert(self, money):
"""Attempt to convert money value to default currency.
If a MissingRate error is raised, ignore it and return None
"""
if money is None:
return None
target_currency = currency_code_default()
try:
result = convert_money(money, target_currency)
except MissingRate:
logger.warning(
'No currency conversion rate available for %s -> %s',
money.currency,
target_currency,
)
result = None
return result
def schedule_for_update(self, counter: int = 0, refresh: bool = True):
"""Schedule this pricing to be updated.
Arguments:
counter: Recursion counter (used to prevent infinite recursion)
refresh: If specified, the PartPricing object will be refreshed from the database
Note:
Within a single request, repeated calls for the same part are cheap: a request-scoped
cache (see InvenTree.cache) short-circuits everything below after the first call,
avoiding redundant refresh_from_db() / existence-check queries for a part whose pricing
has already been resolved (scheduled or not) earlier in the same request. This does not
change behavior outside of a request (e.g. background tasks), where the cache is a
no-op and every call runs in full.
"""
cache_key = f'part-pricing-scheduled-{self.part_id}'
if get_session_cache(cache_key):
return
try:
self._schedule_for_update(counter=counter, refresh=refresh)
finally:
set_session_cache(cache_key, True)
def _schedule_for_update(self, counter: int = 0, refresh: bool = True):
"""Implementation of schedule_for_update(), without the request-scoped cache guard."""
import InvenTree.ready
# If importing data, skip pricing update
if InvenTree.ready.isImportingData():
return
# If running data migrations, skip pricing update
if InvenTree.ready.isRunningMigrations():
return
_part = self.get_part()
if not _part:
logger.warning(
'Referenced part instance does not exist - skipping pricing update.'
)
return
try:
if refresh and self.pk:
self.refresh_from_db()
except (PartPricing.DoesNotExist, IntegrityError):
# Error thrown if this PartPricing instance has already been removed
logger.warning("Error refreshing PartPricing instance for part '%s'", _part)
return
# Ensure that the referenced part still exists in the database
try:
if True: # refresh and p.pk:
_part.refresh_from_db()
except IntegrityError:
logger.exception(
"Could not update PartPricing as Part '%s' does not exist", _part
)
return
if self.scheduled_for_update:
# Ignore if the pricing is already scheduled to be updated
logger.debug(
'Pricing for %s already scheduled for update - skipping', _part
)
return
if counter > self.MAX_PRICING_DEPTH:
# Prevent infinite recursion / stack depth issues
logger.debug(
counter, f'Skipping pricing update for {_part} - maximum depth exceeded'
)
return
try:
self.scheduled_for_update = True
self.save()
except IntegrityError:
# An IntegrityError here likely indicates that the referenced part has already been deleted
logger.exception(
"Could not save PartPricing for part '%s' to the database", _part
)
return
import part.tasks as part_tasks
# Pricing calculations are performed in the background,
# unless the TESTING_PRICING flag is set
background = not settings.TESTING or not settings.TESTING_PRICING
# Offload task to update the pricing
# Force async, to prevent running in the foreground (unless in testing mode)
InvenTree.tasks.offload_task(
part_tasks.update_part_pricing,
self,
counter=counter,
force_async=background,
group='pricing',
)
def update_pricing(
self,
counter: int = 0,
cascade: bool = True,
previous_min=None,
previous_max=None,
):
"""Recalculate all cost data for the referenced Part instance.
Arguments:
counter: Recursion counter (used to prevent infinite recursion)
cascade: If True, update pricing for all assemblies and templates which use this part
previous_min: Previous minimum price (used to prevent further updates if unchanged)
previous_max: Previous maximum price (used to prevent further updates if unchanged)
"""
# If importing data, skip pricing update
if InvenTree.ready.isImportingData():
return
# If running data migrations, skip pricing update
if InvenTree.ready.isRunningMigrations():
return
if self.pk is not None:
try:
self.refresh_from_db()
except PartPricing.DoesNotExist:
# The underlying Part (and this pricing entry) has been deleted
# since this update was scheduled - nothing to do
return
self.update_bom_cost(save=False)
self.update_purchase_cost(save=False)
self.update_internal_cost(save=False)
self.update_supplier_cost(save=False)
self.update_variant_cost(save=False)
self.update_sale_cost(save=False)
# Clear scheduling flag
self.scheduled_for_update = False
# Note: save method calls update_overall_cost
try:
self.save()
except IntegrityError:
# Background worker processes may try to concurrently update
pass
pricing_changed = False
# Without previous pricing data, we assume that the pricing has changed
if previous_min != self.overall_min or previous_max != self.overall_max:
pricing_changed = True
# Update parent assemblies and templates
if pricing_changed and cascade:
self.update_assemblies(counter)
self.update_templates(counter)
def update_assemblies(self, counter: int = 0):
"""Schedule updates for any assemblies which use this part."""
# If the linked Part is used in any assemblies, schedule a pricing update for those assemblies
if not (_part := self.get_part()):
return
used_in_parts = _part.get_used_in()
for p in used_in_parts:
p.pricing.schedule_for_update(counter=counter + 1)
def update_templates(self, counter: int = 0):
"""Schedule updates for any template parts above this part."""
if not (_part := self.get_part()):
return
templates = _part.get_ancestors(include_self=False)
for p in templates:
p.pricing.schedule_for_update(counter=counter + 1)
def save(self, *args, **kwargs):
"""Whenever pricing model is saved, automatically update overall prices."""
# Update the currency which was used to perform the calculation
self.currency = currency_code_default()
if not (_part := self.get_part()):
return
try:
self.update_overall_cost()
super().save(*args, **kwargs)
except Exception:
log_error('PartPricing.save')
logger.error(
"Could not save PartPricing for part '%s' to the database", _part
)
def update_bom_cost(self, save=True):
"""Recalculate BOM cost for the referenced Part instance.
Iterate through the Bill of Materials, and calculate cumulative pricing:
cumulative_min: The sum of minimum costs for each line in the BOM
cumulative_max: The sum of maximum costs for each line in the BOM
Note: The cumulative costs are calculated based on the specified default currency
"""
if not (_part := self.get_part()):
return
if not _part.assembly:
# Not an assembly - no BOM pricing
self.bom_cost_min = None
self.bom_cost_max = None
if save:
self.save()
# Short circuit - no further operations required
return
currency_code = common.currency.currency_code_default()
cumulative_min = Money(0, currency_code)
cumulative_max = Money(0, currency_code)
any_min_elements = False
any_max_elements = False
for bom_item in _part.get_bom_items():
# Loop through each BOM item which is used to assemble this part
bom_item_min = None
bom_item_max = None
for sub_part in bom_item.get_valid_parts_for_allocation():
# Check each part which *could* be used
if sub_part != bom_item.sub_part and not sub_part.active:
continue
sub_part_pricing = sub_part.pricing
sub_part_min = self.convert(sub_part_pricing.overall_min)
sub_part_max = self.convert(sub_part_pricing.overall_max)
if sub_part_min is not None:
if bom_item_min is None or sub_part_min < bom_item_min:
bom_item_min = sub_part_min
if sub_part_max is not None:
if bom_item_max is None or sub_part_max > bom_item_max:
bom_item_max = sub_part_max
# Update cumulative totals
if bom_item_min is not None:
bom_item_min *= bom_item.quantity
cumulative_min += self.convert(bom_item_min)
any_min_elements = True
if bom_item_max is not None:
bom_item_max *= bom_item.quantity
cumulative_max += self.convert(bom_item_max)
any_max_elements = True
if any_min_elements:
self.bom_cost_min = cumulative_min
else:
self.bom_cost_min = None
if any_max_elements:
self.bom_cost_max = cumulative_max
else:
self.bom_cost_max = None
if save:
self.save()
def update_purchase_cost(self, save=True):
"""Recalculate historical purchase cost for the referenced Part instance.
Purchase history only takes into account "completed" purchase orders.
"""
if not (_part := self.get_part()):
return
# Find all line items for completed orders which reference this part
line_items = OrderModels.PurchaseOrderLineItem.objects.filter(
order__status=PurchaseOrderStatus.COMPLETE.value,
received__gt=0,
part__part=_part,
)
# Exclude line items which do not have an associated price
line_items = line_items.exclude(purchase_price=None)
purchase_min = None
purchase_max = None
for line in line_items:
if line.purchase_price is None:
continue
# Take supplier part pack size into account
purchase_cost = self.convert(
line.purchase_price / line.part.pack_quantity_native
)
if purchase_cost is None:
continue
if purchase_min is None or purchase_cost < purchase_min:
purchase_min = purchase_cost
if purchase_max is None or purchase_cost > purchase_max:
purchase_max = purchase_cost
# Also check if manual stock item pricing is included
if get_global_setting('PRICING_USE_STOCK_PRICING', True):
items = self.part.stock_items.all()
# Limit to stock items updated within a certain window
days = int(get_global_setting('PRICING_STOCK_ITEM_AGE_DAYS', 0))
if days > 0:
date_threshold = InvenTree.helpers.current_date() - timedelta(days=days)
items = items.filter(updated__gte=date_threshold)
for item in items:
cost = self.convert(item.purchase_price)
# Skip if the cost could not be converted (for some reason)
if cost is None:
continue
if purchase_min is None or cost < purchase_min:
purchase_min = cost
if purchase_max is None or cost > purchase_max:
purchase_max = cost
self.purchase_cost_min = purchase_min
self.purchase_cost_max = purchase_max
if save:
self.save()
def update_internal_cost(self, save=True):
"""Recalculate internal cost for the referenced Part instance."""
if not (_part := self.get_part()):
return
min_int_cost = None
max_int_cost = None
if get_global_setting('PART_INTERNAL_PRICE', False):
# Only calculate internal pricing if internal pricing is enabled
for pb in _part.internalpricebreaks.all():
cost = self.convert(pb.price)
if cost is None:
# Ignore if cost could not be converted for some reason
continue
if min_int_cost is None or cost < min_int_cost:
min_int_cost = cost
if max_int_cost is None or cost > max_int_cost:
max_int_cost = cost
self.internal_cost_min = min_int_cost
self.internal_cost_max = max_int_cost
if save:
self.save()
def update_supplier_cost(self, save=True):
"""Recalculate supplier cost for the referenced Part instance.
- The limits are simply the lower and upper bounds of available SupplierPriceBreaks
- We do not take "quantity" into account here
"""
if not (_part := self.get_part()):
return
min_sup_cost = None
max_sup_cost = None
if _part.purchaseable:
# Iterate through each available SupplierPart instance
for sp in _part.supplier_parts.all():
# Iterate through each available SupplierPriceBreak instance
for pb in sp.pricebreaks.all():
if pb.price is None:
continue
# Ensure we take supplier part pack size into account
cost = self.convert(pb.price / sp.pack_quantity_native)
if cost is None:
continue
if min_sup_cost is None or cost < min_sup_cost:
min_sup_cost = cost
if max_sup_cost is None or cost > max_sup_cost:
max_sup_cost = cost
self.supplier_price_min = min_sup_cost
self.supplier_price_max = max_sup_cost
if save:
self.save()
def update_variant_cost(self, save=True):
"""Update variant cost values.
Here we track the min/max costs of any variant parts.
"""
if not (_part := self.get_part()):
return
variant_min = None
variant_max = None
active_only = get_global_setting('PRICING_ACTIVE_VARIANTS', False)
if _part.is_template:
variants = _part.get_descendants(include_self=False)
for v in variants:
if active_only and not v.active:
# Ignore inactive variant parts
continue
v_min = self.convert(v.pricing.overall_min)
v_max = self.convert(v.pricing.overall_max)
if v_min is not None:
if variant_min is None or v_min < variant_min:
variant_min = v_min
if v_max is not None:
if variant_max is None or v_max > variant_max:
variant_max = v_max
self.variant_cost_min = variant_min
self.variant_cost_max = variant_max
if save:
self.save()
def update_overall_cost(self):
"""Update overall cost values.
Here we simply take the minimum / maximum values of the other calculated fields.
"""
overall_min = None
overall_max = None
min_costs = [self.bom_cost_min, self.purchase_cost_min, self.internal_cost_min]
max_costs = [self.bom_cost_max, self.purchase_cost_max, self.internal_cost_max]
purchase_history_override = get_global_setting(
'PRICING_PURCHASE_HISTORY_OVERRIDES_SUPPLIER', False
)
if get_global_setting('PRICING_USE_SUPPLIER_PRICING', True):
# Add supplier pricing data, *unless* historical pricing information should override
if self.purchase_cost_min is None or not purchase_history_override:
min_costs.append(self.supplier_price_min)
if self.purchase_cost_max is None or not purchase_history_override:
max_costs.append(self.supplier_price_max)
if get_global_setting('PRICING_USE_VARIANT_PRICING', True):
# Include variant pricing in overall calculations
min_costs.append(self.variant_cost_min)
max_costs.append(self.variant_cost_max)
# Calculate overall minimum cost
for cost in min_costs:
if cost is None:
continue
# Ensure we are working in a common currency
cost = self.convert(cost)
if overall_min is None or cost < overall_min:
overall_min = cost
# Calculate overall maximum cost
for cost in max_costs:
if cost is None:
continue
# Ensure we are working in a common currency
cost = self.convert(cost)
if overall_max is None or cost > overall_max:
overall_max = cost
if get_global_setting('PART_BOM_USE_INTERNAL_PRICE', False):
# Check if internal pricing should override other pricing
if self.internal_cost_min is not None:
overall_min = self.internal_cost_min
if self.internal_cost_max is not None:
overall_max = self.internal_cost_max
if self.override_min is not None:
overall_min = self.convert(self.override_min)
self.overall_min = overall_min
if self.override_max is not None:
overall_max = self.convert(self.override_max)
self.overall_max = overall_max
def update_sale_cost(self, save=True):
"""Recalculate sale cost data."""
if not (_part := self.get_part()):
return
# Iterate through the sell price breaks
min_sell_price = None
max_sell_price = None
for pb in _part.salepricebreaks.all():
cost = self.convert(pb.price)
if cost is None:
continue
if min_sell_price is None or cost < min_sell_price:
min_sell_price = cost
if max_sell_price is None or cost > max_sell_price:
max_sell_price = cost
# Record min/max values
self.sale_price_min = min_sell_price
self.sale_price_max = max_sell_price
min_sell_history = None
max_sell_history = None
# Calculate sale price history too
parts = self.part.get_descendants(include_self=True)
# Find all line items for shipped sales orders which reference this part
line_items = OrderModels.SalesOrderLineItem.objects.filter(
order__status__in=SalesOrderStatusGroups.COMPLETE, part__in=parts
)
# Exclude line items which do not have associated pricing data
line_items = line_items.exclude(sale_price=None)
for line in line_items:
cost = self.convert(line.sale_price)
if cost is None:
continue
if min_sell_history is None or cost < min_sell_history:
min_sell_history = cost
if max_sell_history is None or cost > max_sell_history:
max_sell_history = cost
self.sale_history_min = min_sell_history
self.sale_history_max = max_sell_history
if save:
self.save()
currency = models.CharField(
default=currency_code_default,
max_length=10,
verbose_name=_('Currency'),
help_text=_('Currency used to cache pricing calculations'),
validators=[validators.validate_currency_code],
)
scheduled_for_update = models.BooleanField(default=False)
part = models.OneToOneField(
Part,
on_delete=models.CASCADE,
related_name='pricing_data',
verbose_name=_('Part'),
)
bom_cost_min = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Minimum BOM Cost'),
help_text=_('Minimum cost of component parts'),
)
bom_cost_max = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Maximum BOM Cost'),
help_text=_('Maximum cost of component parts'),
)
purchase_cost_min = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Minimum Purchase Cost'),
help_text=_('Minimum historical purchase cost'),
)
purchase_cost_max = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Maximum Purchase Cost'),
help_text=_('Maximum historical purchase cost'),
)
internal_cost_min = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Minimum Internal Price'),
help_text=_('Minimum cost based on internal price breaks'),
)
internal_cost_max = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Maximum Internal Price'),
help_text=_('Maximum cost based on internal price breaks'),
)
supplier_price_min = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Minimum Supplier Price'),
help_text=_('Minimum price of part from external suppliers'),
)
supplier_price_max = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Maximum Supplier Price'),
help_text=_('Maximum price of part from external suppliers'),
)
variant_cost_min = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Minimum Variant Cost'),
help_text=_('Calculated minimum cost of variant parts'),
)
variant_cost_max = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Maximum Variant Cost'),
help_text=_('Calculated maximum cost of variant parts'),
)
override_min = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Minimum Cost'),
help_text=_('Override minimum cost'),
)
override_max = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Maximum Cost'),
help_text=_('Override maximum cost'),
)
overall_min = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Minimum Cost'),
help_text=_('Calculated overall minimum cost'),
)
overall_max = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Maximum Cost'),
help_text=_('Calculated overall maximum cost'),
)
sale_price_min = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Minimum Sale Price'),
help_text=_('Minimum sale price based on price breaks'),
)
sale_price_max = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Maximum Sale Price'),
help_text=_('Maximum sale price based on price breaks'),
)
sale_history_min = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Minimum Sale Cost'),
help_text=_('Minimum historical sale price'),
)
sale_history_max = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Maximum Sale Cost'),
help_text=_('Maximum historical sale price'),
)
class PartStocktake(models.Model):
"""Model representing a 'stock history' entry for a particular Part.
@@ -3468,51 +2664,6 @@ class PartStocktake(models.Model):
)
class PartSellPriceBreak(common.models.PriceBreak):
"""Represents a price break for selling this part."""
class Meta:
"""Metaclass providing extra model definition."""
verbose_name = _('Part Sale Price Break')
unique_together = ('part', 'quantity')
@staticmethod
def get_api_url():
"""Return the list API endpoint URL associated with the PartSellPriceBreak model."""
return reverse('api-part-sale-price-list')
part = models.ForeignKey(
Part,
on_delete=models.CASCADE,
related_name='salepricebreaks',
limit_choices_to={'salable': True},
verbose_name=_('Part'),
)
class PartInternalPriceBreak(common.models.PriceBreak):
"""Represents a price break for internally selling this part."""
class Meta:
"""Metaclass providing extra model definition."""
unique_together = ('part', 'quantity')
verbose_name = _('Part Internal Price Break')
@staticmethod
def get_api_url():
"""Return the list API endpoint URL associated with the PartInternalPriceBreak model."""
return reverse('api-part-internal-price-list')
part = models.ForeignKey(
Part,
on_delete=models.CASCADE,
related_name='internalpricebreaks',
verbose_name=_('Part'),
)
class PartStar(models.Model):
"""A PartStar object creates a subscription relationship between a User and a Part.
@@ -4379,16 +3530,8 @@ def update_bom_build_lines(sender, instance, created, **kwargs):
@receiver(post_save, sender=BomItem, dispatch_uid='post_save_bom_item')
@receiver(
post_save, sender=PartSellPriceBreak, dispatch_uid='post_save_sale_price_break'
)
@receiver(
post_save,
sender=PartInternalPriceBreak,
dispatch_uid='post_save_internal_price_break',
)
def update_pricing_after_edit(sender, instance, created, **kwargs):
"""Callback function when a part price break is created or updated."""
"""Callback function when a BomItem is created or updated."""
# Update part pricing *unless* we are importing data
if (
InvenTree.ready.canAppAccessDatabase(allow_test=settings.TESTING_PRICING)
@@ -4399,16 +3542,8 @@ def update_pricing_after_edit(sender, instance, created, **kwargs):
@receiver(post_delete, sender=BomItem, dispatch_uid='post_delete_bom_item')
@receiver(
post_delete, sender=PartSellPriceBreak, dispatch_uid='post_delete_sale_price_break'
)
@receiver(
post_delete,
sender=PartInternalPriceBreak,
dispatch_uid='post_delete_internal_price_break',
)
def update_pricing_after_delete(sender, instance, **kwargs):
"""Callback function when a part price break is deleted."""
"""Callback function when a BomItem is deleted."""
# Update part pricing *unless* we are importing data
if (
InvenTree.ready.canAppAccessDatabase(allow_test=settings.TESTING_PRICING)
+12 -251
View File
@@ -12,13 +12,9 @@ from django.urls import reverse_lazy
from django.utils.translation import gettext_lazy as _
import structlog
from djmoney.contrib.exchange.exceptions import MissingRate
from djmoney.contrib.exchange.models import convert_money
from drf_spectacular.utils import extend_schema_field
from rest_framework import serializers
from sql_util.utils import SubqueryCount
import common.currency
import common.filters
import common.serializers
import company.models
@@ -34,6 +30,8 @@ from importer.registry import register_importer
from InvenTree.mixins import DataImportExportSerializerMixin
from InvenTree.ready import isGeneratingSchema
from InvenTree.serializers import OptionalField, TreePathSerializer
from pricing.models import PartPricing
from pricing.serializers import PartSalePriceSerializer
from users.serializers import UserSerializer
from .models import (
@@ -42,10 +40,7 @@ from .models import (
Part,
PartCategory,
PartCategoryParameterTemplate,
PartInternalPriceBreak,
PartPricing,
PartRelated,
PartSellPriceBreak,
PartStar,
PartStocktake,
PartTestTemplate,
@@ -251,52 +246,6 @@ class PartTestTemplateSerializer(
return queryset.annotate(results=SubqueryCount('test_results'))
@register_importer()
class PartSalePriceSerializer(
InvenTree.serializers.FilterableSerializerMixin,
DataImportExportSerializerMixin,
InvenTree.serializers.InvenTreeModelSerializer,
):
"""Serializer for sale prices for Part model."""
no_filters = True
class Meta:
"""Metaclass defining serializer fields."""
model = PartSellPriceBreak
fields = ['pk', 'part', 'quantity', 'price', 'price_currency']
quantity = InvenTree.serializers.InvenTreeDecimalField()
price = InvenTree.serializers.InvenTreeMoneySerializer(allow_null=True)
price_currency = InvenTree.serializers.InvenTreeCurrencySerializer(
help_text=_('Purchase currency of this stock item')
)
@register_importer()
class PartInternalPriceSerializer(
DataImportExportSerializerMixin, InvenTree.serializers.InvenTreeModelSerializer
):
"""Serializer for internal prices for Part model."""
class Meta:
"""Metaclass defining serializer fields."""
model = PartInternalPriceBreak
fields = ['pk', 'part', 'quantity', 'price', 'price_currency']
quantity = InvenTree.serializers.InvenTreeDecimalField()
price = InvenTree.serializers.InvenTreeMoneySerializer(allow_null=True)
price_currency = InvenTree.serializers.InvenTreeCurrencySerializer(
help_text=_('Purchase currency of this stock item')
)
class PartThumbSerializer(serializers.Serializer):
"""Serializer for the 'image' field of the Part model.
@@ -397,6 +346,7 @@ class PartBriefSerializer(
},
default_include=True,
filter_name='pricing',
model=PartPricing,
)
pricing_max = OptionalField(
@@ -408,6 +358,7 @@ class PartBriefSerializer(
},
default_include=True,
filter_name='pricing',
model=PartPricing,
)
@@ -907,6 +858,7 @@ class PartSerializer(
},
default_include=True,
filter_name='pricing',
model=PartPricing,
)
pricing_max = OptionalField(
@@ -918,6 +870,7 @@ class PartSerializer(
},
default_include=True,
filter_name='pricing',
model=PartPricing,
)
pricing_updated = OptionalField(
@@ -929,6 +882,7 @@ class PartSerializer(
},
default_include=True,
filter_name='pricing',
model=PartPricing,
)
parameters = common.filters.enable_parameters_filter()
@@ -1379,204 +1333,6 @@ class PartStocktakeGenerateSerializer(serializers.Serializer):
)
@extend_schema_field(
serializers.CharField(
help_text=_('Select currency from available options')
+ '\n\n'
+ '\n'.join(
f'* `{value}` - {label}'
for value, label in common.currency.currency_code_mappings()
)
+ "\n\nOther valid currencies may be found in the 'CURRENCY_CODES' global setting."
)
)
class PartPricingCurrencySerializer(serializers.ChoiceField):
"""Serializer to allow annotating the schema to use String on currency fields."""
class PartPricingSerializer(InvenTree.serializers.InvenTreeModelSerializer):
"""Serializer for Part pricing information."""
class Meta:
"""Metaclass defining serializer fields."""
model = PartPricing
fields = [
'currency',
'updated',
'scheduled_for_update',
'bom_cost_min',
'bom_cost_max',
'purchase_cost_min',
'purchase_cost_max',
'internal_cost_min',
'internal_cost_max',
'supplier_price_min',
'supplier_price_max',
'variant_cost_min',
'variant_cost_max',
'override_min',
'override_min_currency',
'override_max',
'override_max_currency',
'overall_min',
'overall_max',
'sale_price_min',
'sale_price_max',
'sale_history_min',
'sale_history_max',
'update',
]
currency = InvenTree.serializers.InvenTreeCurrencySerializer(
allow_null=True, read_only=True
)
updated = serializers.DateTimeField(allow_null=True, read_only=True)
scheduled_for_update = serializers.BooleanField(read_only=True)
# Custom serializers
bom_cost_min = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
bom_cost_max = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
purchase_cost_min = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
purchase_cost_max = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
internal_cost_min = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
internal_cost_max = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
supplier_price_min = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
supplier_price_max = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
variant_cost_min = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
variant_cost_max = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
override_min = InvenTree.serializers.InvenTreeMoneySerializer(
label=_('Minimum Price'),
help_text=_('Override calculated value for minimum price'),
allow_null=True,
read_only=False,
required=False,
)
override_min_currency = PartPricingCurrencySerializer(
label=_('Minimum price currency'),
read_only=False,
required=False,
choices=common.currency.currency_code_mappings(),
)
override_max = InvenTree.serializers.InvenTreeMoneySerializer(
label=_('Maximum Price'),
help_text=_('Override calculated value for maximum price'),
allow_null=True,
read_only=False,
required=False,
)
override_max_currency = PartPricingCurrencySerializer(
label=_('Maximum price currency'),
read_only=False,
required=False,
choices=common.currency.currency_code_mappings(),
)
overall_min = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
overall_max = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
sale_price_min = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
sale_price_max = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
sale_history_min = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
sale_history_max = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
update = serializers.BooleanField(
write_only=True,
label=_('Update'),
help_text=_('Update pricing for this part'),
default=False,
required=False,
allow_null=True,
)
def validate(self, data):
"""Validate supplied pricing data."""
super().validate(data)
# Check that override_min is not greater than override_max
override_min = data.get('override_min', None)
override_max = data.get('override_max', None)
default_currency = common.currency.currency_code_default()
if override_min is not None and override_max is not None:
try:
override_min = convert_money(override_min, default_currency)
override_max = convert_money(override_max, default_currency)
except MissingRate:
raise ValidationError(
_(
f'Could not convert from provided currencies to {default_currency}'
)
)
if override_min > override_max:
raise ValidationError({
'override_min': _(
'Minimum price must not be greater than maximum price'
),
'override_max': _(
'Maximum price must not be less than minimum price'
),
})
return data
def save(self):
"""Called when the serializer is saved."""
super().save()
data = self.validated_data
if data.get('update', False):
# Update part pricing
pricing = self.instance
pricing.update_pricing()
class PartSerialNumberSerializer(InvenTree.serializers.InvenTreeModelSerializer):
"""Serializer for Part serial number information."""
@@ -1814,6 +1570,7 @@ class BomItemSerializer(
},
default_include=True,
filter_name='pricing',
model=PartPricing,
)
pricing_max = OptionalField(
@@ -1825,6 +1582,7 @@ class BomItemSerializer(
},
default_include=True,
filter_name='pricing',
model=PartPricing,
)
pricing_min_total = OptionalField(
@@ -1832,6 +1590,7 @@ class BomItemSerializer(
serializer_kwargs={'allow_null': True, 'read_only': True},
default_include=True,
filter_name='pricing',
model=PartPricing,
)
pricing_max_total = OptionalField(
@@ -1839,6 +1598,7 @@ class BomItemSerializer(
serializer_kwargs={'allow_null': True, 'read_only': True},
default_include=True,
filter_name='pricing',
model=PartPricing,
)
pricing_updated = OptionalField(
@@ -1850,6 +1610,7 @@ class BomItemSerializer(
},
default_include=True,
filter_name='pricing',
model=PartPricing,
)
# Annotated fields for available stock
+2 -1
View File
@@ -257,7 +257,8 @@ def check_missing_pricing(limit=250):
Arguments:
limit: Maximum number of parts to process at once
"""
from part.models import Part, PartPricing
from part.models import Part
from pricing.models import PartPricing
# Find any parts which have 'old' pricing information
days = int(get_global_setting('PRICING_UPDATE_DAYS', 30))
+14 -105
View File
@@ -37,9 +37,9 @@ from part.models import (
PartCategory,
PartCategoryParameterTemplate,
PartRelated,
PartSellPriceBreak,
PartTestTemplate,
)
from pricing.models import PartSellPriceBreak
from stock.models import StockItem, StockLocation
from stock.status_codes import StockStatus
@@ -822,6 +822,7 @@ class PartAPITestBase(InvenTreeAPITestCase):
'part.delete',
'part_category.change',
'part_category.add',
'pricing.view',
'stock_location.view',
]
@@ -1467,6 +1468,15 @@ class PartAPITest(PartAPITestBase):
else:
self.assertNotIn(field, response.data)
# Without the 'pricing' role, pricing fields are never included
self.clearRoles()
self.assignRole('part.view')
response = self.get(url, {'pricing': True}, expected_code=200)
for field in pricing_fields:
self.assertNotIn(field, response.data)
def test_parameters_info(self):
"""Test annotation of 'parameters' detail against a Part instance."""
part = Part.objects.first()
@@ -2480,45 +2490,6 @@ class PartNotesTests(InvenTreeAPITestCase):
self.assertEqual(response.data['content'], notes.strip())
class PartPricingDetailTests(InvenTreeAPITestCase):
"""Tests for the part pricing API endpoint."""
fixtures = ['category', 'part', 'location']
roles = ['part.change']
def url(self, pk):
"""Construct a pricing URL."""
return reverse('api-part-pricing', kwargs={'pk': pk})
def test_pricing_detail(self):
"""Test an empty pricing detail."""
response = self.get(self.url(1), expected_code=200)
# Check for expected fields
expected_fields = [
'currency',
'updated',
'bom_cost_min',
'bom_cost_max',
'purchase_cost_min',
'purchase_cost_max',
'internal_cost_min',
'internal_cost_max',
'supplier_price_min',
'supplier_price_max',
'overall_min',
'overall_max',
]
for field in expected_fields:
self.assertIn(field, response.data)
# Empty fields (no pricing by default)
for field in expected_fields[2:]:
self.assertIsNone(response.data[field])
class PartAPIAggregationTest(InvenTreeAPITestCase):
"""Tests to ensure that the various aggregation annotations are working correctly."""
@@ -3157,6 +3128,9 @@ class BomItemTest(InvenTreeAPITestCase):
# Viewing 'substitutes' requires the 'bom' role
self.assignRole('bom.view')
# Viewing cached pricing data requires the 'pricing' role
self.assignRole('pricing.view')
self.run_output_test(
reverse('api-bom-item-detail', kwargs={'pk': 3}),
[
@@ -3633,71 +3607,6 @@ class AttachmentTest(InvenTreeAPITestCase):
self.assertEqual(data['comment'], 'Hello world')
class PartInternalPriceBreakTest(InvenTreeAPITestCase):
"""Unit tests for the PartInternalPrice API endpoints."""
fixtures = [
'category',
'part',
'params',
'location',
'bom',
'company',
'test_templates',
'manufacturer_part',
'supplier_part',
'order',
'stock',
]
roles = [
'part.change',
'part.add',
'part.delete',
'part_category.change',
'part_category.add',
'part_category.delete',
]
def test_create_price_breaks(self):
"""Test we can create price breaks at various quantities."""
url = reverse('api-part-internal-price-list')
breaks = [
(1.0, 101),
(1.1, 92.555555555),
(1.5, 90.999999999),
(1.756, 89),
(2, 86),
(25, 80),
]
for q, p in breaks:
data = self.post(
url, {'part': 1, 'quantity': q, 'price': p}, expected_code=201
).data
self.assertEqual(data['part'], 1)
self.assertEqual(round(Decimal(data['quantity']), 4), round(Decimal(q), 4))
self.assertEqual(round(Decimal(data['price']), 4), round(Decimal(p), 4))
# Now, ensure that we can delete the Part via the API
# In particular this test checks that there are no circular post_delete relationships
# Ref: https://github.com/inventree/InvenTree/pull/3986
# First, ensure the part instance can be deleted
p = Part.objects.get(pk=1)
p.active = False
p.save()
InvenTreeSetting.set_setting('PART_ALLOW_DELETE_FROM_ASSEMBLY', True)
self.delete(reverse('api-part-detail', kwargs={'pk': 1}))
with self.assertRaises(Part.DoesNotExist):
p.refresh_from_db()
class PartTestTemplateTest(PartAPITestBase):
"""API unit tests for the PartTestTemplate model."""
@@ -0,0 +1 @@
"""The 'pricing' app provides models and functionality for part pricing data."""
+48
View File
@@ -0,0 +1,48 @@
"""Admin class definitions for the 'pricing' app."""
from django.contrib import admin
from pricing import models
@admin.register(models.PartPricing)
class PartPricingAdmin(admin.ModelAdmin):
"""Admin class for PartPricing model."""
list_display = ('part', 'overall_min', 'overall_max')
search_fields = ['part__name', 'part__IPN', 'part__description']
autocomplete_fields = ['part']
@admin.register(models.PartSellPriceBreak)
class PartSellPriceBreakAdmin(admin.ModelAdmin):
"""Admin class for the PartSellPriceBreak model."""
class Meta:
"""Metaclass options."""
model = models.PartSellPriceBreak
list_display = ('part', 'quantity', 'price')
search_fields = ['part__name', 'part__IPN']
autocomplete_fields = ('part',)
@admin.register(models.PartInternalPriceBreak)
class PartInternalPriceBreakAdmin(admin.ModelAdmin):
"""Admin class for the PartInternalPriceBreak model."""
class Meta:
"""Metaclass options."""
model = models.PartInternalPriceBreak
list_display = ('part', 'quantity', 'price')
search_fields = ['part__name', 'part__IPN']
autocomplete_fields = ('part',)
+108
View File
@@ -0,0 +1,108 @@
"""Provides a JSON API for the 'pricing' app."""
from django.shortcuts import get_object_or_404
from django.urls import include, path
from data_exporter.mixins import DataExportViewMixin
from InvenTree.filters import SEARCH_ORDER_FILTER
from InvenTree.mixins import ListCreateAPI, RetrieveUpdateAPI, RetrieveUpdateDestroyAPI
from part.models import Part
from . import serializers as pricing_serializers
from .models import PartInternalPriceBreak, PartPricing, PartSellPriceBreak
class PartSalePriceDetail(RetrieveUpdateDestroyAPI):
"""Detail endpoint for PartSellPriceBreak model."""
queryset = PartSellPriceBreak.objects.all()
serializer_class = pricing_serializers.PartSalePriceSerializer
class PartSalePriceList(DataExportViewMixin, ListCreateAPI):
"""API endpoint for list view of PartSalePriceBreak model."""
queryset = PartSellPriceBreak.objects.all()
serializer_class = pricing_serializers.PartSalePriceSerializer
filter_backends = SEARCH_ORDER_FILTER
filterset_fields = ['part']
ordering_fields = ['quantity', 'price']
ordering = 'quantity'
class PartInternalPriceDetail(RetrieveUpdateDestroyAPI):
"""Detail endpoint for PartInternalPriceBreak model."""
queryset = PartInternalPriceBreak.objects.all()
serializer_class = pricing_serializers.PartInternalPriceSerializer
class PartInternalPriceList(DataExportViewMixin, ListCreateAPI):
"""API endpoint for list view of PartInternalPriceBreak model."""
queryset = PartInternalPriceBreak.objects.all()
serializer_class = pricing_serializers.PartInternalPriceSerializer
filter_backends = SEARCH_ORDER_FILTER
filterset_fields = ['part']
ordering_fields = ['quantity', 'price']
ordering = 'quantity'
class PartPricingDetail(RetrieveUpdateAPI):
"""API endpoint for viewing part pricing data.
Note: The URL 'pk' refers to the linked Part instance (not the PartPricing instance).
"""
serializer_class = pricing_serializers.PartPricingSerializer
# The queryset model determines the required (model) permissions for this endpoint,
# so it must be PartPricing (rather than Part)
queryset = PartPricing.objects.all()
def get_object(self):
"""Return the PartPricing object associated with the linked Part.
If no PartPricing instance exists for the Part, a new (unsaved) instance is returned.
"""
part = get_object_or_404(Part, pk=self.kwargs.get('pk'))
pricing = part.pricing
self.check_object_permissions(self.request, pricing)
return pricing
# Note: URL names are retained from when these endpoints were part of the 'part' app
pricing_api_urls = [
# Part sale pricing
path(
'sale-price/',
include([
path(
'<int:pk>/',
PartSalePriceDetail.as_view(),
name='api-part-sale-price-detail',
),
path('', PartSalePriceList.as_view(), name='api-part-sale-price-list'),
]),
),
# Part internal pricing
path(
'internal-price/',
include([
path(
'<int:pk>/',
PartInternalPriceDetail.as_view(),
name='api-part-internal-price-detail',
),
path(
'', PartInternalPriceList.as_view(), name='api-part-internal-price-list'
),
]),
),
# Cached pricing data for a particular part
path('part/<int:pk>/', PartPricingDetail.as_view(), name='api-part-pricing'),
]
+9
View File
@@ -0,0 +1,9 @@
"""AppConfig for the 'pricing' app."""
from django.apps import AppConfig
class PricingConfig(AppConfig):
"""AppConfig class for the 'pricing' app."""
name = 'pricing'
@@ -2,14 +2,14 @@
# Price breaks for R_2K2_0805
- model: part.partsellpricebreak
- model: pricing.partsellpricebreak
pk: 1
fields:
part: 3
quantity: 1
price: 0.15
- model: part.partsellpricebreak
- model: pricing.partsellpricebreak
pk: 2
fields:
part: 3
@@ -21,14 +21,14 @@
# Internal Price breaks for R_2K2_0805
- model: part.partinternalpricebreak
- model: pricing.partinternalpricebreak
pk: 1
fields:
part: 3
quantity: 1
price: 0.08
- model: part.partinternalpricebreak
- model: pricing.partinternalpricebreak
pk: 2
fields:
part: 3
@@ -36,14 +36,14 @@
price: 0.05
# Internal Price breaks for C_22N_0805
- model: part.partinternalpricebreak
- model: pricing.partinternalpricebreak
pk: 3
fields:
part: 5
quantity: 1
price: 1
- model: part.partinternalpricebreak
- model: pricing.partinternalpricebreak
pk: 4
fields:
part: 5
@@ -0,0 +1,726 @@
# Generated by Django 5.2.17 on 2026-10-17 02:26
import InvenTree.fields
import InvenTree.validators
import common.currency
import django.core.validators
import django.db.models.deletion
import djmoney.models.fields
import djmoney.models.validators
from django.db import migrations, models
# Models which have been moved from the 'part' app
PRICING_MODELS = ['partpricing', 'partsellpricebreak', 'partinternalpricebreak']
def move_content_types(apps, schema_editor, old_app, new_app):
"""Move ContentType entries for the pricing models between apps.
This retains any database entries (e.g. permissions) which reference these content types.
"""
ContentType = apps.get_model('contenttypes', 'ContentType')
for model in PRICING_MODELS:
if ContentType.objects.filter(app_label=new_app, model=model).exists():
continue
ContentType.objects.filter(app_label=old_app, model=model).update(
app_label=new_app
)
def forward_content_types(apps, schema_editor):
"""Move ContentType entries from the 'part' app to the 'pricing' app."""
move_content_types(apps, schema_editor, 'part', 'pricing')
def reverse_content_types(apps, schema_editor):
"""Move ContentType entries from the 'pricing' app back to the 'part' app."""
move_content_types(apps, schema_editor, 'pricing', 'part')
class Migration(migrations.Migration):
initial = True
dependencies = [
('part', '0155_move_pricing_models'),
('contenttypes', '0002_remove_content_type_name'),
]
# These models were previously defined in the 'part' app (see part.0155_move_pricing_models).
# The existing database tables are retained (via Meta.db_table),
# so the models are only created in the migration state, not the database.
operations = [
migrations.SeparateDatabaseAndState(
state_operations=[
migrations.CreateModel(
name='PartPricing',
fields=[
(
'id',
models.AutoField(
auto_created=True,
primary_key=True,
serialize=False,
verbose_name='ID',
),
),
(
'updated',
models.DateTimeField(
auto_now=True,
help_text='Timestamp of last update',
null=True,
verbose_name='Updated',
),
),
(
'currency',
models.CharField(
default=common.currency.currency_code_default,
help_text='Currency used to cache pricing calculations',
max_length=10,
validators=[
InvenTree.validators.validate_currency_code
],
verbose_name='Currency',
),
),
('scheduled_for_update', models.BooleanField(default=False)),
(
'bom_cost_min_currency',
djmoney.models.fields.CurrencyField(
choices=[],
default='',
editable=False,
max_length=3,
null=True,
),
),
(
'bom_cost_min',
InvenTree.fields.InvenTreeModelMoneyField(
blank=True,
currency_choices=[],
decimal_places=6,
default_currency='',
help_text='Minimum cost of component parts',
max_digits=19,
null=True,
validators=[
djmoney.models.validators.MinMoneyValidator(0)
],
verbose_name='Minimum BOM Cost',
),
),
(
'bom_cost_max_currency',
djmoney.models.fields.CurrencyField(
choices=[],
default='',
editable=False,
max_length=3,
null=True,
),
),
(
'bom_cost_max',
InvenTree.fields.InvenTreeModelMoneyField(
blank=True,
currency_choices=[],
decimal_places=6,
default_currency='',
help_text='Maximum cost of component parts',
max_digits=19,
null=True,
validators=[
djmoney.models.validators.MinMoneyValidator(0)
],
verbose_name='Maximum BOM Cost',
),
),
(
'purchase_cost_min_currency',
djmoney.models.fields.CurrencyField(
choices=[],
default='',
editable=False,
max_length=3,
null=True,
),
),
(
'purchase_cost_min',
InvenTree.fields.InvenTreeModelMoneyField(
blank=True,
currency_choices=[],
decimal_places=6,
default_currency='',
help_text='Minimum historical purchase cost',
max_digits=19,
null=True,
validators=[
djmoney.models.validators.MinMoneyValidator(0)
],
verbose_name='Minimum Purchase Cost',
),
),
(
'purchase_cost_max_currency',
djmoney.models.fields.CurrencyField(
choices=[],
default='',
editable=False,
max_length=3,
null=True,
),
),
(
'purchase_cost_max',
InvenTree.fields.InvenTreeModelMoneyField(
blank=True,
currency_choices=[],
decimal_places=6,
default_currency='',
help_text='Maximum historical purchase cost',
max_digits=19,
null=True,
validators=[
djmoney.models.validators.MinMoneyValidator(0)
],
verbose_name='Maximum Purchase Cost',
),
),
(
'internal_cost_min_currency',
djmoney.models.fields.CurrencyField(
choices=[],
default='',
editable=False,
max_length=3,
null=True,
),
),
(
'internal_cost_min',
InvenTree.fields.InvenTreeModelMoneyField(
blank=True,
currency_choices=[],
decimal_places=6,
default_currency='',
help_text='Minimum cost based on internal price breaks',
max_digits=19,
null=True,
validators=[
djmoney.models.validators.MinMoneyValidator(0)
],
verbose_name='Minimum Internal Price',
),
),
(
'internal_cost_max_currency',
djmoney.models.fields.CurrencyField(
choices=[],
default='',
editable=False,
max_length=3,
null=True,
),
),
(
'internal_cost_max',
InvenTree.fields.InvenTreeModelMoneyField(
blank=True,
currency_choices=[],
decimal_places=6,
default_currency='',
help_text='Maximum cost based on internal price breaks',
max_digits=19,
null=True,
validators=[
djmoney.models.validators.MinMoneyValidator(0)
],
verbose_name='Maximum Internal Price',
),
),
(
'supplier_price_min_currency',
djmoney.models.fields.CurrencyField(
choices=[],
default='',
editable=False,
max_length=3,
null=True,
),
),
(
'supplier_price_min',
InvenTree.fields.InvenTreeModelMoneyField(
blank=True,
currency_choices=[],
decimal_places=6,
default_currency='',
help_text='Minimum price of part from external suppliers',
max_digits=19,
null=True,
validators=[
djmoney.models.validators.MinMoneyValidator(0)
],
verbose_name='Minimum Supplier Price',
),
),
(
'supplier_price_max_currency',
djmoney.models.fields.CurrencyField(
choices=[],
default='',
editable=False,
max_length=3,
null=True,
),
),
(
'supplier_price_max',
InvenTree.fields.InvenTreeModelMoneyField(
blank=True,
currency_choices=[],
decimal_places=6,
default_currency='',
help_text='Maximum price of part from external suppliers',
max_digits=19,
null=True,
validators=[
djmoney.models.validators.MinMoneyValidator(0)
],
verbose_name='Maximum Supplier Price',
),
),
(
'variant_cost_min_currency',
djmoney.models.fields.CurrencyField(
choices=[],
default='',
editable=False,
max_length=3,
null=True,
),
),
(
'variant_cost_min',
InvenTree.fields.InvenTreeModelMoneyField(
blank=True,
currency_choices=[],
decimal_places=6,
default_currency='',
help_text='Calculated minimum cost of variant parts',
max_digits=19,
null=True,
validators=[
djmoney.models.validators.MinMoneyValidator(0)
],
verbose_name='Minimum Variant Cost',
),
),
(
'variant_cost_max_currency',
djmoney.models.fields.CurrencyField(
choices=[],
default='',
editable=False,
max_length=3,
null=True,
),
),
(
'variant_cost_max',
InvenTree.fields.InvenTreeModelMoneyField(
blank=True,
currency_choices=[],
decimal_places=6,
default_currency='',
help_text='Calculated maximum cost of variant parts',
max_digits=19,
null=True,
validators=[
djmoney.models.validators.MinMoneyValidator(0)
],
verbose_name='Maximum Variant Cost',
),
),
(
'override_min_currency',
djmoney.models.fields.CurrencyField(
choices=[],
default='',
editable=False,
max_length=3,
null=True,
),
),
(
'override_min',
InvenTree.fields.InvenTreeModelMoneyField(
blank=True,
currency_choices=[],
decimal_places=6,
default_currency='',
help_text='Override minimum cost',
max_digits=19,
null=True,
validators=[
djmoney.models.validators.MinMoneyValidator(0)
],
verbose_name='Minimum Cost',
),
),
(
'override_max_currency',
djmoney.models.fields.CurrencyField(
choices=[],
default='',
editable=False,
max_length=3,
null=True,
),
),
(
'override_max',
InvenTree.fields.InvenTreeModelMoneyField(
blank=True,
currency_choices=[],
decimal_places=6,
default_currency='',
help_text='Override maximum cost',
max_digits=19,
null=True,
validators=[
djmoney.models.validators.MinMoneyValidator(0)
],
verbose_name='Maximum Cost',
),
),
(
'overall_min_currency',
djmoney.models.fields.CurrencyField(
choices=[],
default='',
editable=False,
max_length=3,
null=True,
),
),
(
'overall_min',
InvenTree.fields.InvenTreeModelMoneyField(
blank=True,
currency_choices=[],
decimal_places=6,
default_currency='',
help_text='Calculated overall minimum cost',
max_digits=19,
null=True,
validators=[
djmoney.models.validators.MinMoneyValidator(0)
],
verbose_name='Minimum Cost',
),
),
(
'overall_max_currency',
djmoney.models.fields.CurrencyField(
choices=[],
default='',
editable=False,
max_length=3,
null=True,
),
),
(
'overall_max',
InvenTree.fields.InvenTreeModelMoneyField(
blank=True,
currency_choices=[],
decimal_places=6,
default_currency='',
help_text='Calculated overall maximum cost',
max_digits=19,
null=True,
validators=[
djmoney.models.validators.MinMoneyValidator(0)
],
verbose_name='Maximum Cost',
),
),
(
'sale_price_min_currency',
djmoney.models.fields.CurrencyField(
choices=[],
default='',
editable=False,
max_length=3,
null=True,
),
),
(
'sale_price_min',
InvenTree.fields.InvenTreeModelMoneyField(
blank=True,
currency_choices=[],
decimal_places=6,
default_currency='',
help_text='Minimum sale price based on price breaks',
max_digits=19,
null=True,
validators=[
djmoney.models.validators.MinMoneyValidator(0)
],
verbose_name='Minimum Sale Price',
),
),
(
'sale_price_max_currency',
djmoney.models.fields.CurrencyField(
choices=[],
default='',
editable=False,
max_length=3,
null=True,
),
),
(
'sale_price_max',
InvenTree.fields.InvenTreeModelMoneyField(
blank=True,
currency_choices=[],
decimal_places=6,
default_currency='',
help_text='Maximum sale price based on price breaks',
max_digits=19,
null=True,
validators=[
djmoney.models.validators.MinMoneyValidator(0)
],
verbose_name='Maximum Sale Price',
),
),
(
'sale_history_min_currency',
djmoney.models.fields.CurrencyField(
choices=[],
default='',
editable=False,
max_length=3,
null=True,
),
),
(
'sale_history_min',
InvenTree.fields.InvenTreeModelMoneyField(
blank=True,
currency_choices=[],
decimal_places=6,
default_currency='',
help_text='Minimum historical sale price',
max_digits=19,
null=True,
validators=[
djmoney.models.validators.MinMoneyValidator(0)
],
verbose_name='Minimum Sale Cost',
),
),
(
'sale_history_max_currency',
djmoney.models.fields.CurrencyField(
choices=[],
default='',
editable=False,
max_length=3,
null=True,
),
),
(
'sale_history_max',
InvenTree.fields.InvenTreeModelMoneyField(
blank=True,
currency_choices=[],
decimal_places=6,
default_currency='',
help_text='Maximum historical sale price',
max_digits=19,
null=True,
validators=[
djmoney.models.validators.MinMoneyValidator(0)
],
verbose_name='Maximum Sale Cost',
),
),
(
'part',
models.OneToOneField(
on_delete=django.db.models.deletion.CASCADE,
related_name='pricing_data',
to='part.part',
verbose_name='Part',
),
),
],
options={'db_table': 'part_partpricing'},
),
migrations.CreateModel(
name='PartInternalPriceBreak',
fields=[
(
'id',
models.AutoField(
auto_created=True,
primary_key=True,
serialize=False,
verbose_name='ID',
),
),
(
'updated',
models.DateTimeField(
auto_now=True,
help_text='Timestamp of last update',
null=True,
verbose_name='Updated',
),
),
(
'quantity',
InvenTree.fields.RoundingDecimalField(
decimal_places=5,
default=1,
help_text='Price break quantity',
max_digits=15,
validators=[
django.core.validators.MinValueValidator(1)
],
verbose_name='Quantity',
),
),
(
'price_currency',
djmoney.models.fields.CurrencyField(
choices=[],
default='',
editable=False,
max_length=3,
null=True,
),
),
(
'price',
InvenTree.fields.InvenTreeModelMoneyField(
currency_choices=[],
decimal_places=6,
default_currency='',
help_text='Unit price at specified quantity',
max_digits=19,
null=True,
validators=[
djmoney.models.validators.MinMoneyValidator(0)
],
verbose_name='Price',
),
),
(
'part',
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
related_name='internalpricebreaks',
to='part.part',
verbose_name='Part',
),
),
],
options={
'verbose_name': 'Part Internal Price Break',
'db_table': 'part_partinternalpricebreak',
'unique_together': {('part', 'quantity')},
},
),
migrations.CreateModel(
name='PartSellPriceBreak',
fields=[
(
'id',
models.AutoField(
auto_created=True,
primary_key=True,
serialize=False,
verbose_name='ID',
),
),
(
'updated',
models.DateTimeField(
auto_now=True,
help_text='Timestamp of last update',
null=True,
verbose_name='Updated',
),
),
(
'quantity',
InvenTree.fields.RoundingDecimalField(
decimal_places=5,
default=1,
help_text='Price break quantity',
max_digits=15,
validators=[
django.core.validators.MinValueValidator(1)
],
verbose_name='Quantity',
),
),
(
'price_currency',
djmoney.models.fields.CurrencyField(
choices=[],
default='',
editable=False,
max_length=3,
null=True,
),
),
(
'price',
InvenTree.fields.InvenTreeModelMoneyField(
currency_choices=[],
decimal_places=6,
default_currency='',
help_text='Unit price at specified quantity',
max_digits=19,
null=True,
validators=[
djmoney.models.validators.MinMoneyValidator(0)
],
verbose_name='Price',
),
),
(
'part',
models.ForeignKey(
limit_choices_to={'salable': True},
on_delete=django.db.models.deletion.CASCADE,
related_name='salepricebreaks',
to='part.part',
verbose_name='Part',
),
),
],
options={
'verbose_name': 'Part Sale Price Break',
'db_table': 'part_partsellpricebreak',
'unique_together': {('part', 'quantity')},
},
),
],
database_operations=[],
),
migrations.RunPython(forward_content_types, reverse_code=reverse_content_types),
]
@@ -0,0 +1,115 @@
"""Migrate queued background tasks which reference the moved pricing models.
The PartPricing, PartSellPriceBreak and PartInternalPriceBreak models were moved from the
'part' app to the 'pricing' app (see 0001_initial).
Background tasks which were queued before this change (e.g. part pricing updates)
contain pickled model instances which reference the old ('part', <model>) identity.
These can no longer be unpickled, which causes the background worker to fail when
it attempts to process the queued task.
This migration rewrites any such queued tasks to reference the new ('pricing', <model>) identity.
"""
import pickle
from django.db import migrations
from django.db.models.base import model_unpickle
import structlog
logger = structlog.get_logger('inventree')
# Mapping of legacy model identities to their new identities
LEGACY_MODELS = {
('part', 'PartPricing'): ('pricing', 'PartPricing'),
('part', 'PartSellPriceBreak'): ('pricing', 'PartSellPriceBreak'),
('part', 'PartInternalPriceBreak'): ('pricing', 'PartInternalPriceBreak'),
}
def remapped_model_unpickle(model_id):
"""Unpickle a model instance, mapping any legacy model identity to its new identity."""
if isinstance(model_id, tuple):
model_id = LEGACY_MODELS.get(model_id, model_id)
return model_unpickle(model_id)
class RemappingUnpickler(pickle.Unpickler):
"""Unpickler which maps legacy model identities to their new identities.
Django pickles model instances as a call to 'model_unpickle(app_label, object_name)',
so the remapping is performed by substituting that function.
"""
def find_class(self, module, name):
"""Substitute the model unpickling function."""
if module == 'django.db.models.base' and name == 'model_unpickle':
return remapped_model_unpickle
return super().find_class(module, name)
class RemappingPickleSerializer:
"""Serializer for django-q signed payloads, which remaps legacy model identities."""
@staticmethod
def loads(data):
"""Unpickle the provided data, remapping legacy model identities."""
import io
return RemappingUnpickler(io.BytesIO(data)).load()
def migrate_queued_tasks(apps, schema_editor):
"""Rewrite queued task payloads which reference legacy model identities."""
from django_q import core_signing
from django_q.conf import Conf
from django_q.signing import SignedPackage
OrmQ = apps.get_model('django_q', 'OrmQ')
migrated = 0
for task in OrmQ.objects.all():
try:
SignedPackage.loads(task.payload)
# Payload can be loaded - nothing to do
continue
except LookupError:
# Payload references a model which does not exist (e.g. a legacy model identity)
pass
except Exception:
# Payload cannot be loaded for some other reason - leave it untouched
continue
try:
data = core_signing.loads(
task.payload,
key=Conf.SECRET_KEY,
salt=Conf.PREFIX,
serializer=RemappingPickleSerializer,
)
except Exception:
logger.warning('Could not migrate queued task %s', task.pk)
continue
# Re-pickle the payload, which now references the new model identity
task.payload = SignedPackage.dumps(data)
task.save()
migrated += 1
if migrated:
logger.info('pricing.0002: Migrated %s queued background tasks', migrated)
class Migration(migrations.Migration):
dependencies = [
('pricing', '0001_initial'),
('django_q', '0019_alter_task_options_alter_ormq_key_alter_ormq_lock_and_more'),
]
operations = [
migrations.RunPython(migrate_queued_tasks, reverse_code=migrations.RunPython.noop)
]
Whitespace-only changes.
+926
View File
@@ -0,0 +1,926 @@
"""Database model definitions for the 'pricing' app."""
from datetime import timedelta
from django.conf import settings
from django.db import models
from django.db.models.signals import post_delete, post_save
from django.db.utils import IntegrityError
from django.dispatch import receiver
from django.urls import reverse
from django.utils.translation import gettext_lazy as _
import structlog
from djmoney.contrib.exchange.exceptions import MissingRate
from djmoney.contrib.exchange.models import convert_money
from djmoney.money import Money
import common.currency
import common.models
import InvenTree.fields
import InvenTree.helpers
import InvenTree.ready
import InvenTree.tasks
from common.currency import currency_code_default
from common.settings import get_global_setting
from InvenTree import validators
from InvenTree.cache import get_session_cache, set_session_cache
from InvenTree.exceptions import log_error
from order import models as OrderModels
from order.status_codes import PurchaseOrderStatus, SalesOrderStatusGroups
from part.models import Part
logger = structlog.get_logger('inventree')
class PartPricing(common.models.MetaMixin):
"""Model for caching min/max pricing information for a particular Part.
It is prohibitively expensive to calculate min/max pricing for a part "on the fly".
As min/max pricing does not change very often, we pre-calculate and cache these values.
Whenever pricing is updated, these values are re-calculated and stored.
Pricing information is cached for:
- BOM cost (min / max cost of component items)
- Purchase cost (based on purchase history)
- Internal cost (based on user-specified InternalPriceBreak data)
- Supplier price (based on supplier part data)
- Variant price (min / max cost of any variants)
- Overall best / worst (based on the values listed above)
- Sale price break min / max values
- Historical sale pricing min / max values
Note that this pricing information does not take "quantity" into account:
- This provides a simple min / max pricing range, which is quite valuable in a lot of situations
- Quantity pricing still needs to be calculated
- Quantity pricing can be viewed from the part detail page
- Detailed pricing information is very context specific in any case
"""
class Meta:
"""Metaclass providing extra model definition."""
# Table name retained from when this model was part of the 'part' app
db_table = 'part_partpricing'
# When calculating assembly pricing, we limit the depth of the calculation
MAX_PRICING_DEPTH = 50
@property
def is_valid(self):
"""Return True if the cached pricing is valid."""
return self.updated is not None
def get_part(self):
"""Return the Part instance associated with this pricing object.
If the part does not exist (i.e. has been deleted), return None.
"""
try:
part = self.part
part.refresh_from_db()
return part
except (Part.DoesNotExist, IntegrityError):
return None
def convert(self, money):
"""Attempt to convert money value to default currency.
If a MissingRate error is raised, ignore it and return None
"""
if money is None:
return None
target_currency = currency_code_default()
try:
result = convert_money(money, target_currency)
except MissingRate:
logger.warning(
'No currency conversion rate available for %s -> %s',
money.currency,
target_currency,
)
result = None
return result
def schedule_for_update(self, counter: int = 0, refresh: bool = True):
"""Schedule this pricing to be updated.
Arguments:
counter: Recursion counter (used to prevent infinite recursion)
refresh: If specified, the PartPricing object will be refreshed from the database
Note:
Within a single request, repeated calls for the same part are cheap: a request-scoped
cache (see InvenTree.cache) short-circuits everything below after the first call,
avoiding redundant refresh_from_db() / existence-check queries for a part whose pricing
has already been resolved (scheduled or not) earlier in the same request. This does not
change behavior outside of a request (e.g. background tasks), where the cache is a
no-op and every call runs in full.
"""
cache_key = f'part-pricing-scheduled-{self.part_id}'
if get_session_cache(cache_key):
return
try:
self._schedule_for_update(counter=counter, refresh=refresh)
finally:
set_session_cache(cache_key, True)
def _schedule_for_update(self, counter: int = 0, refresh: bool = True):
"""Implementation of schedule_for_update(), without the request-scoped cache guard."""
# If importing data, skip pricing update
if InvenTree.ready.isImportingData():
return
# If running data migrations, skip pricing update
if InvenTree.ready.isRunningMigrations():
return
_part = self.get_part()
if not _part:
logger.warning(
'Referenced part instance does not exist - skipping pricing update.'
)
return
try:
if refresh and self.pk:
self.refresh_from_db()
except (PartPricing.DoesNotExist, IntegrityError):
# Error thrown if this PartPricing instance has already been removed
logger.warning("Error refreshing PartPricing instance for part '%s'", _part)
return
# Ensure that the referenced part still exists in the database
try:
_part.refresh_from_db()
except IntegrityError:
logger.exception(
"Could not update PartPricing as Part '%s' does not exist", _part
)
return
if self.scheduled_for_update:
# Ignore if the pricing is already scheduled to be updated
logger.debug(
'Pricing for %s already scheduled for update - skipping', _part
)
return
if counter > self.MAX_PRICING_DEPTH:
# Prevent infinite recursion / stack depth issues
logger.debug(
'Skipping pricing update for %s - maximum depth exceeded (%s)',
_part,
counter,
)
return
try:
self.scheduled_for_update = True
self.save()
except IntegrityError:
# An IntegrityError here likely indicates that the referenced part has already been deleted
logger.exception(
"Could not save PartPricing for part '%s' to the database", _part
)
return
import part.tasks as part_tasks
# Pricing calculations are performed in the background,
# unless the TESTING_PRICING flag is set
background = not settings.TESTING or not settings.TESTING_PRICING
# Offload task to update the pricing
# Force async, to prevent running in the foreground (unless in testing mode)
InvenTree.tasks.offload_task(
part_tasks.update_part_pricing,
self,
counter=counter,
force_async=background,
group='pricing',
)
def update_pricing(
self,
counter: int = 0,
cascade: bool = True,
previous_min=None,
previous_max=None,
):
"""Recalculate all cost data for the referenced Part instance.
Arguments:
counter: Recursion counter (used to prevent infinite recursion)
cascade: If True, update pricing for all assemblies and templates which use this part
previous_min: Previous minimum price (used to prevent further updates if unchanged)
previous_max: Previous maximum price (used to prevent further updates if unchanged)
"""
# If importing data, skip pricing update
if InvenTree.ready.isImportingData():
return
# If running data migrations, skip pricing update
if InvenTree.ready.isRunningMigrations():
return
if self.pk is not None:
try:
self.refresh_from_db()
except PartPricing.DoesNotExist:
# The underlying Part (and this pricing entry) has been deleted
# since this update was scheduled - nothing to do
return
self.update_bom_cost(save=False)
self.update_purchase_cost(save=False)
self.update_internal_cost(save=False)
self.update_supplier_cost(save=False)
self.update_variant_cost(save=False)
self.update_sale_cost(save=False)
# Clear scheduling flag
self.scheduled_for_update = False
# Note: save method calls update_overall_cost
try:
self.save()
except IntegrityError:
# Background worker processes may try to concurrently update
pass
pricing_changed = False
# Without previous pricing data, we assume that the pricing has changed
if previous_min != self.overall_min or previous_max != self.overall_max:
pricing_changed = True
# Update parent assemblies and templates
if pricing_changed and cascade:
self.update_assemblies(counter)
self.update_templates(counter)
def update_assemblies(self, counter: int = 0):
"""Schedule updates for any assemblies which use this part."""
# If the linked Part is used in any assemblies, schedule a pricing update for those assemblies
if not (_part := self.get_part()):
return
used_in_parts = _part.get_used_in()
for p in used_in_parts:
p.pricing.schedule_for_update(counter=counter + 1)
def update_templates(self, counter: int = 0):
"""Schedule updates for any template parts above this part."""
if not (_part := self.get_part()):
return
templates = _part.get_ancestors(include_self=False)
for p in templates:
p.pricing.schedule_for_update(counter=counter + 1)
def save(self, *args, **kwargs):
"""Whenever pricing model is saved, automatically update overall prices."""
# Update the currency which was used to perform the calculation
self.currency = currency_code_default()
if not (_part := self.get_part()):
return
try:
self.update_overall_cost()
super().save(*args, **kwargs)
except Exception:
log_error('PartPricing.save')
logger.error(
"Could not save PartPricing for part '%s' to the database", _part
)
def update_bom_cost(self, save=True):
"""Recalculate BOM cost for the referenced Part instance.
Iterate through the Bill of Materials, and calculate cumulative pricing:
cumulative_min: The sum of minimum costs for each line in the BOM
cumulative_max: The sum of maximum costs for each line in the BOM
Note: The cumulative costs are calculated based on the specified default currency
"""
if not (_part := self.get_part()):
return
if not _part.assembly:
# Not an assembly - no BOM pricing
self.bom_cost_min = None
self.bom_cost_max = None
if save:
self.save()
# Short circuit - no further operations required
return
currency_code = common.currency.currency_code_default()
cumulative_min = Money(0, currency_code)
cumulative_max = Money(0, currency_code)
any_min_elements = False
any_max_elements = False
for bom_item in _part.get_bom_items():
# Loop through each BOM item which is used to assemble this part
bom_item_min = None
bom_item_max = None
for sub_part in bom_item.get_valid_parts_for_allocation():
# Check each part which *could* be used
if sub_part != bom_item.sub_part and not sub_part.active:
continue
sub_part_pricing = sub_part.pricing
sub_part_min = self.convert(sub_part_pricing.overall_min)
sub_part_max = self.convert(sub_part_pricing.overall_max)
if sub_part_min is not None:
if bom_item_min is None or sub_part_min < bom_item_min:
bom_item_min = sub_part_min
if sub_part_max is not None:
if bom_item_max is None or sub_part_max > bom_item_max:
bom_item_max = sub_part_max
# Update cumulative totals
if bom_item_min is not None:
bom_item_min *= bom_item.quantity
cumulative_min += self.convert(bom_item_min)
any_min_elements = True
if bom_item_max is not None:
bom_item_max *= bom_item.quantity
cumulative_max += self.convert(bom_item_max)
any_max_elements = True
if any_min_elements:
self.bom_cost_min = cumulative_min
else:
self.bom_cost_min = None
if any_max_elements:
self.bom_cost_max = cumulative_max
else:
self.bom_cost_max = None
if save:
self.save()
def update_purchase_cost(self, save=True):
"""Recalculate historical purchase cost for the referenced Part instance.
Purchase history only takes into account "completed" purchase orders.
"""
if not (_part := self.get_part()):
return
# Find all line items for completed orders which reference this part
line_items = OrderModels.PurchaseOrderLineItem.objects.filter(
order__status=PurchaseOrderStatus.COMPLETE.value,
received__gt=0,
part__part=_part,
)
# Exclude line items which do not have an associated price
line_items = line_items.exclude(purchase_price=None)
purchase_min = None
purchase_max = None
for line in line_items:
if line.purchase_price is None:
continue
# Take supplier part pack size into account
purchase_cost = self.convert(
line.purchase_price / line.part.pack_quantity_native
)
if purchase_cost is None:
continue
if purchase_min is None or purchase_cost < purchase_min:
purchase_min = purchase_cost
if purchase_max is None or purchase_cost > purchase_max:
purchase_max = purchase_cost
# Also check if manual stock item pricing is included
if get_global_setting('PRICING_USE_STOCK_PRICING', True):
items = self.part.stock_items.all()
# Limit to stock items updated within a certain window
days = int(get_global_setting('PRICING_STOCK_ITEM_AGE_DAYS', 0))
if days > 0:
date_threshold = InvenTree.helpers.current_date() - timedelta(days=days)
items = items.filter(updated__gte=date_threshold)
for item in items:
cost = self.convert(item.purchase_price)
# Skip if the cost could not be converted (for some reason)
if cost is None:
continue
if purchase_min is None or cost < purchase_min:
purchase_min = cost
if purchase_max is None or cost > purchase_max:
purchase_max = cost
self.purchase_cost_min = purchase_min
self.purchase_cost_max = purchase_max
if save:
self.save()
def update_internal_cost(self, save=True):
"""Recalculate internal cost for the referenced Part instance."""
if not (_part := self.get_part()):
return
min_int_cost = None
max_int_cost = None
if get_global_setting('PART_INTERNAL_PRICE', False):
# Only calculate internal pricing if internal pricing is enabled
for pb in _part.internalpricebreaks.all():
cost = self.convert(pb.price)
if cost is None:
# Ignore if cost could not be converted for some reason
continue
if min_int_cost is None or cost < min_int_cost:
min_int_cost = cost
if max_int_cost is None or cost > max_int_cost:
max_int_cost = cost
self.internal_cost_min = min_int_cost
self.internal_cost_max = max_int_cost
if save:
self.save()
def update_supplier_cost(self, save=True):
"""Recalculate supplier cost for the referenced Part instance.
- The limits are simply the lower and upper bounds of available SupplierPriceBreaks
- We do not take "quantity" into account here
"""
if not (_part := self.get_part()):
return
min_sup_cost = None
max_sup_cost = None
if _part.purchaseable:
# Iterate through each available SupplierPart instance
for sp in _part.supplier_parts.all():
# Iterate through each available SupplierPriceBreak instance
for pb in sp.pricebreaks.all():
if pb.price is None:
continue
# Ensure we take supplier part pack size into account
cost = self.convert(pb.price / sp.pack_quantity_native)
if cost is None:
continue
if min_sup_cost is None or cost < min_sup_cost:
min_sup_cost = cost
if max_sup_cost is None or cost > max_sup_cost:
max_sup_cost = cost
self.supplier_price_min = min_sup_cost
self.supplier_price_max = max_sup_cost
if save:
self.save()
def update_variant_cost(self, save=True):
"""Update variant cost values.
Here we track the min/max costs of any variant parts.
"""
if not (_part := self.get_part()):
return
variant_min = None
variant_max = None
active_only = get_global_setting('PRICING_ACTIVE_VARIANTS', False)
if _part.is_template:
variants = _part.get_descendants(include_self=False)
for v in variants:
if active_only and not v.active:
# Ignore inactive variant parts
continue
v_min = self.convert(v.pricing.overall_min)
v_max = self.convert(v.pricing.overall_max)
if v_min is not None:
if variant_min is None or v_min < variant_min:
variant_min = v_min
if v_max is not None:
if variant_max is None or v_max > variant_max:
variant_max = v_max
self.variant_cost_min = variant_min
self.variant_cost_max = variant_max
if save:
self.save()
def update_overall_cost(self):
"""Update overall cost values.
Here we simply take the minimum / maximum values of the other calculated fields.
"""
overall_min = None
overall_max = None
min_costs = [self.bom_cost_min, self.purchase_cost_min, self.internal_cost_min]
max_costs = [self.bom_cost_max, self.purchase_cost_max, self.internal_cost_max]
purchase_history_override = get_global_setting(
'PRICING_PURCHASE_HISTORY_OVERRIDES_SUPPLIER', False
)
if get_global_setting('PRICING_USE_SUPPLIER_PRICING', True):
# Add supplier pricing data, *unless* historical pricing information should override
if self.purchase_cost_min is None or not purchase_history_override:
min_costs.append(self.supplier_price_min)
if self.purchase_cost_max is None or not purchase_history_override:
max_costs.append(self.supplier_price_max)
if get_global_setting('PRICING_USE_VARIANT_PRICING', True):
# Include variant pricing in overall calculations
min_costs.append(self.variant_cost_min)
max_costs.append(self.variant_cost_max)
# Calculate overall minimum cost
for cost in min_costs:
if cost is None:
continue
# Ensure we are working in a common currency
cost = self.convert(cost)
if overall_min is None or cost < overall_min:
overall_min = cost
# Calculate overall maximum cost
for cost in max_costs:
if cost is None:
continue
# Ensure we are working in a common currency
cost = self.convert(cost)
if overall_max is None or cost > overall_max:
overall_max = cost
if get_global_setting('PART_BOM_USE_INTERNAL_PRICE', False):
# Check if internal pricing should override other pricing
if self.internal_cost_min is not None:
overall_min = self.internal_cost_min
if self.internal_cost_max is not None:
overall_max = self.internal_cost_max
if self.override_min is not None:
overall_min = self.convert(self.override_min)
self.overall_min = overall_min
if self.override_max is not None:
overall_max = self.convert(self.override_max)
self.overall_max = overall_max
def update_sale_cost(self, save=True):
"""Recalculate sale cost data."""
if not (_part := self.get_part()):
return
# Iterate through the sell price breaks
min_sell_price = None
max_sell_price = None
for pb in _part.salepricebreaks.all():
cost = self.convert(pb.price)
if cost is None:
continue
if min_sell_price is None or cost < min_sell_price:
min_sell_price = cost
if max_sell_price is None or cost > max_sell_price:
max_sell_price = cost
# Record min/max values
self.sale_price_min = min_sell_price
self.sale_price_max = max_sell_price
min_sell_history = None
max_sell_history = None
# Calculate sale price history too
parts = self.part.get_descendants(include_self=True)
# Find all line items for shipped sales orders which reference this part
line_items = OrderModels.SalesOrderLineItem.objects.filter(
order__status__in=SalesOrderStatusGroups.COMPLETE, part__in=parts
)
# Exclude line items which do not have associated pricing data
line_items = line_items.exclude(sale_price=None)
for line in line_items:
cost = self.convert(line.sale_price)
if cost is None:
continue
if min_sell_history is None or cost < min_sell_history:
min_sell_history = cost
if max_sell_history is None or cost > max_sell_history:
max_sell_history = cost
self.sale_history_min = min_sell_history
self.sale_history_max = max_sell_history
if save:
self.save()
currency = models.CharField(
default=currency_code_default,
max_length=10,
verbose_name=_('Currency'),
help_text=_('Currency used to cache pricing calculations'),
validators=[validators.validate_currency_code],
)
scheduled_for_update = models.BooleanField(default=False)
part = models.OneToOneField(
Part,
on_delete=models.CASCADE,
related_name='pricing_data',
verbose_name=_('Part'),
)
bom_cost_min = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Minimum BOM Cost'),
help_text=_('Minimum cost of component parts'),
)
bom_cost_max = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Maximum BOM Cost'),
help_text=_('Maximum cost of component parts'),
)
purchase_cost_min = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Minimum Purchase Cost'),
help_text=_('Minimum historical purchase cost'),
)
purchase_cost_max = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Maximum Purchase Cost'),
help_text=_('Maximum historical purchase cost'),
)
internal_cost_min = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Minimum Internal Price'),
help_text=_('Minimum cost based on internal price breaks'),
)
internal_cost_max = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Maximum Internal Price'),
help_text=_('Maximum cost based on internal price breaks'),
)
supplier_price_min = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Minimum Supplier Price'),
help_text=_('Minimum price of part from external suppliers'),
)
supplier_price_max = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Maximum Supplier Price'),
help_text=_('Maximum price of part from external suppliers'),
)
variant_cost_min = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Minimum Variant Cost'),
help_text=_('Calculated minimum cost of variant parts'),
)
variant_cost_max = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Maximum Variant Cost'),
help_text=_('Calculated maximum cost of variant parts'),
)
override_min = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Minimum Cost'),
help_text=_('Override minimum cost'),
)
override_max = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Maximum Cost'),
help_text=_('Override maximum cost'),
)
overall_min = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Minimum Cost'),
help_text=_('Calculated overall minimum cost'),
)
overall_max = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Maximum Cost'),
help_text=_('Calculated overall maximum cost'),
)
sale_price_min = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Minimum Sale Price'),
help_text=_('Minimum sale price based on price breaks'),
)
sale_price_max = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Maximum Sale Price'),
help_text=_('Maximum sale price based on price breaks'),
)
sale_history_min = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Minimum Sale Cost'),
help_text=_('Minimum historical sale price'),
)
sale_history_max = InvenTree.fields.InvenTreeModelMoneyField(
null=True,
blank=True,
verbose_name=_('Maximum Sale Cost'),
help_text=_('Maximum historical sale price'),
)
class PartSellPriceBreak(common.models.PriceBreak):
"""Represents a price break for selling this part."""
class Meta:
"""Metaclass providing extra model definition."""
verbose_name = _('Part Sale Price Break')
unique_together = ('part', 'quantity')
# Table name retained from when this model was part of the 'part' app
db_table = 'part_partsellpricebreak'
@staticmethod
def get_api_url():
"""Return the list API endpoint URL associated with the PartSellPriceBreak model."""
return reverse('api-part-sale-price-list')
part = models.ForeignKey(
Part,
on_delete=models.CASCADE,
related_name='salepricebreaks',
limit_choices_to={'salable': True},
verbose_name=_('Part'),
)
class PartInternalPriceBreak(common.models.PriceBreak):
"""Represents a price break for internally selling this part."""
class Meta:
"""Metaclass providing extra model definition."""
unique_together = ('part', 'quantity')
verbose_name = _('Part Internal Price Break')
# Table name retained from when this model was part of the 'part' app
db_table = 'part_partinternalpricebreak'
@staticmethod
def get_api_url():
"""Return the list API endpoint URL associated with the PartInternalPriceBreak model."""
return reverse('api-part-internal-price-list')
part = models.ForeignKey(
Part,
on_delete=models.CASCADE,
related_name='internalpricebreaks',
verbose_name=_('Part'),
)
@receiver(
post_save, sender=PartSellPriceBreak, dispatch_uid='post_save_sale_price_break'
)
@receiver(
post_save,
sender=PartInternalPriceBreak,
dispatch_uid='post_save_internal_price_break',
)
def update_pricing_after_edit(sender, instance, created, **kwargs):
"""Callback function when a part price break is created or updated."""
# Update part pricing *unless* we are importing data
if (
InvenTree.ready.canAppAccessDatabase(allow_test=settings.TESTING_PRICING)
and not InvenTree.ready.isImportingData()
):
if instance.part:
instance.part.schedule_pricing_update(create=True)
@receiver(
post_delete, sender=PartSellPriceBreak, dispatch_uid='post_delete_sale_price_break'
)
@receiver(
post_delete,
sender=PartInternalPriceBreak,
dispatch_uid='post_delete_internal_price_break',
)
def update_pricing_after_delete(sender, instance, **kwargs):
"""Callback function when a part price break is deleted."""
# Update part pricing *unless* we are importing data
if (
InvenTree.ready.canAppAccessDatabase(allow_test=settings.TESTING_PRICING)
and not InvenTree.ready.isImportingData()
):
try:
part = instance.part
part.schedule_pricing_update(create=False)
except Part.DoesNotExist:
# The part instance may have already been deleted,
# in which case we cannot update pricing
pass
@@ -0,0 +1,266 @@
"""DRF data serializers for the 'pricing' app."""
from django.core.exceptions import ValidationError
from django.utils.translation import gettext_lazy as _
from djmoney.contrib.exchange.exceptions import MissingRate
from djmoney.contrib.exchange.models import convert_money
from drf_spectacular.utils import extend_schema_field
from rest_framework import serializers
import common.currency
import InvenTree.serializers
from importer.registry import register_importer
from InvenTree.mixins import DataImportExportSerializerMixin
from .models import PartInternalPriceBreak, PartPricing, PartSellPriceBreak
@register_importer()
class PartSalePriceSerializer(
InvenTree.serializers.FilterableSerializerMixin,
DataImportExportSerializerMixin,
InvenTree.serializers.InvenTreeModelSerializer,
):
"""Serializer for sale prices for Part model."""
no_filters = True
class Meta:
"""Metaclass defining serializer fields."""
model = PartSellPriceBreak
fields = ['pk', 'part', 'quantity', 'price', 'price_currency']
quantity = InvenTree.serializers.InvenTreeDecimalField()
price = InvenTree.serializers.InvenTreeMoneySerializer(allow_null=True)
price_currency = InvenTree.serializers.InvenTreeCurrencySerializer(
help_text=_('Currency for this price break')
)
@register_importer()
class PartInternalPriceSerializer(
DataImportExportSerializerMixin, InvenTree.serializers.InvenTreeModelSerializer
):
"""Serializer for internal prices for Part model."""
class Meta:
"""Metaclass defining serializer fields."""
model = PartInternalPriceBreak
fields = ['pk', 'part', 'quantity', 'price', 'price_currency']
quantity = InvenTree.serializers.InvenTreeDecimalField()
price = InvenTree.serializers.InvenTreeMoneySerializer(allow_null=True)
price_currency = InvenTree.serializers.InvenTreeCurrencySerializer(
help_text=_('Currency for this price break')
)
@extend_schema_field(
serializers.CharField(
help_text=_('Select currency from available options')
+ '\n\n'
+ '\n'.join(
f'* `{value}` - {label}'
for value, label in common.currency.currency_code_mappings()
)
+ "\n\nOther valid currencies may be found in the 'CURRENCY_CODES' global setting."
)
)
class PartPricingCurrencySerializer(serializers.ChoiceField):
"""Serializer to allow annotating the schema to use String on currency fields."""
class PartPricingSerializer(InvenTree.serializers.InvenTreeModelSerializer):
"""Serializer for Part pricing information."""
class Meta:
"""Metaclass defining serializer fields."""
model = PartPricing
fields = [
'currency',
'updated',
'scheduled_for_update',
'bom_cost_min',
'bom_cost_max',
'purchase_cost_min',
'purchase_cost_max',
'internal_cost_min',
'internal_cost_max',
'supplier_price_min',
'supplier_price_max',
'variant_cost_min',
'variant_cost_max',
'override_min',
'override_min_currency',
'override_max',
'override_max_currency',
'overall_min',
'overall_max',
'sale_price_min',
'sale_price_max',
'sale_history_min',
'sale_history_max',
'update',
]
currency = InvenTree.serializers.InvenTreeCurrencySerializer(
allow_null=True, read_only=True
)
updated = serializers.DateTimeField(allow_null=True, read_only=True)
scheduled_for_update = serializers.BooleanField(read_only=True)
# Custom serializers
bom_cost_min = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
bom_cost_max = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
purchase_cost_min = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
purchase_cost_max = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
internal_cost_min = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
internal_cost_max = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
supplier_price_min = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
supplier_price_max = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
variant_cost_min = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
variant_cost_max = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
override_min = InvenTree.serializers.InvenTreeMoneySerializer(
label=_('Minimum Price'),
help_text=_('Override calculated value for minimum price'),
allow_null=True,
read_only=False,
required=False,
)
override_min_currency = PartPricingCurrencySerializer(
label=_('Minimum price currency'),
read_only=False,
required=False,
choices=common.currency.currency_code_mappings(),
)
override_max = InvenTree.serializers.InvenTreeMoneySerializer(
label=_('Maximum Price'),
help_text=_('Override calculated value for maximum price'),
allow_null=True,
read_only=False,
required=False,
)
override_max_currency = PartPricingCurrencySerializer(
label=_('Maximum price currency'),
read_only=False,
required=False,
choices=common.currency.currency_code_mappings(),
)
overall_min = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
overall_max = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
sale_price_min = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
sale_price_max = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
sale_history_min = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
sale_history_max = InvenTree.serializers.InvenTreeMoneySerializer(
allow_null=True, read_only=True
)
update = serializers.BooleanField(
write_only=True,
label=_('Update'),
help_text=_('Update pricing for this part'),
default=False,
required=False,
allow_null=True,
)
def validate(self, data):
"""Validate supplied pricing data."""
super().validate(data)
# Check that override_min is not greater than override_max
# For a partial update, compare against the existing value of any field not provided
def get_value(field):
if field in data:
return data[field]
return getattr(self.instance, field, None)
override_min = get_value('override_min')
override_max = get_value('override_max')
default_currency = common.currency.currency_code_default()
if override_min is not None and override_max is not None:
try:
override_min = convert_money(override_min, default_currency)
override_max = convert_money(override_max, default_currency)
except MissingRate:
raise ValidationError(
_(
'Could not convert from provided currencies to {currency}'
).format(currency=default_currency)
)
if override_min > override_max:
raise ValidationError({
'override_min': _(
'Minimum price must not be greater than maximum price'
),
'override_max': _(
'Maximum price must not be less than minimum price'
),
})
return data
def save(self):
"""Called when the serializer is saved."""
super().save()
data = self.validated_data
if data.get('update', False):
# Update part pricing
pricing = self.instance
pricing.update_pricing()
+362
View File
@@ -0,0 +1,362 @@
"""API unit tests for the 'pricing' app."""
from decimal import Decimal
from django.urls import reverse
from rest_framework.test import APIClient
from common.models import InvenTreeSetting
from InvenTree.unit_test import InvenTreeAPITestCase
from part.models import Part
from pricing.models import PartInternalPriceBreak, PartSellPriceBreak
class PartPricingDetailTests(InvenTreeAPITestCase):
"""Tests for the part pricing API endpoint."""
fixtures = ['category', 'part', 'location']
roles = ['pricing.change']
def url(self, pk):
"""Construct a pricing URL."""
return reverse('api-part-pricing', kwargs={'pk': pk})
def test_pricing_detail(self):
"""Test an empty pricing detail."""
response = self.get(self.url(1), expected_code=200)
# Check for expected fields
expected_fields = [
'currency',
'updated',
'bom_cost_min',
'bom_cost_max',
'purchase_cost_min',
'purchase_cost_max',
'internal_cost_min',
'internal_cost_max',
'supplier_price_min',
'supplier_price_max',
'overall_min',
'overall_max',
]
for field in expected_fields:
self.assertIn(field, response.data)
# Empty fields (no pricing by default)
for field in expected_fields[2:]:
self.assertIsNone(response.data[field])
def test_pricing_permissions(self):
"""Test that the 'pricing' role (not the 'part' role) controls access."""
url = self.url(1)
self.clearRoles()
self.assignRole('part.change')
self.get(url, expected_code=403)
self.patch(url, {'update': True}, expected_code=403)
# The 'part' role is not required to view or edit pricing data
self.clearRoles()
self.assignRole('pricing.view')
self.get(url, expected_code=200)
self.patch(url, {'update': True}, expected_code=403)
self.assignRole('pricing.change')
self.patch(url, {'update': True}, expected_code=200)
def test_override_validation(self):
"""Override minimum price must not exceed override maximum price."""
url = self.url(1)
# Both values provided together
self.patch(
url,
{
'override_min': 100,
'override_min_currency': 'USD',
'override_max': 10,
'override_max_currency': 'USD',
},
expected_code=400,
)
self.patch(
url, {'override_max': 10, 'override_max_currency': 'USD'}, expected_code=200
)
# Partial update - compared against the existing override_max value
response = self.patch(
url,
{'override_min': 100, 'override_min_currency': 'USD'},
expected_code=400,
)
self.assertIn('override_min', response.data)
self.patch(
url, {'override_min': 5, 'override_min_currency': 'USD'}, expected_code=200
)
# Partial update of the maximum - compared against the existing override_min value
self.patch(
url, {'override_max': 1, 'override_max_currency': 'USD'}, expected_code=400
)
# Clearing a value removes the constraint
self.patch(url, {'override_max': None}, expected_code=200)
self.patch(
url,
{'override_min': 1000, 'override_min_currency': 'USD'},
expected_code=200,
)
class PartInternalPriceBreakTest(InvenTreeAPITestCase):
"""Unit tests for the PartInternalPrice API endpoints."""
fixtures = [
'category',
'part',
'params',
'location',
'bom',
'company',
'test_templates',
'manufacturer_part',
'supplier_part',
'order',
'stock',
]
roles = [
'part.change',
'part.add',
'part.delete',
'part_category.change',
'part_category.add',
'part_category.delete',
'pricing.add',
'pricing.delete',
]
def test_price_break_permissions(self):
"""Test that price breaks require the 'pricing' role."""
# Sale price breaks can only be created for salable parts
Part.objects.filter(pk=1).update(salable=True)
for url in [
reverse('api-part-internal-price-list'),
reverse('api-part-sale-price-list'),
]:
self.clearRoles()
self.assignRole('part.change')
self.get(url, expected_code=403)
self.post(url, {'part': 1, 'quantity': 1, 'price': 1}, expected_code=403)
self.assignRole('pricing.view')
self.get(url, expected_code=200)
self.post(url, {'part': 1, 'quantity': 1, 'price': 1}, expected_code=403)
self.assignRole('pricing.add')
self.post(url, {'part': 1, 'quantity': 1, 'price': 1}, expected_code=201)
def test_create_price_breaks(self):
"""Test we can create price breaks at various quantities."""
url = reverse('api-part-internal-price-list')
breaks = [
(1.0, 101),
(1.1, 92.555555555),
(1.5, 90.999999999),
(1.756, 89),
(2, 86),
(25, 80),
]
for q, p in breaks:
data = self.post(
url, {'part': 1, 'quantity': q, 'price': p}, expected_code=201
).data
self.assertEqual(data['part'], 1)
self.assertEqual(round(Decimal(data['quantity']), 4), round(Decimal(q), 4))
self.assertEqual(round(Decimal(data['price']), 4), round(Decimal(p), 4))
# Now, ensure that we can delete the Part via the API
# In particular this test checks that there are no circular post_delete relationships
# Ref: https://github.com/inventree/InvenTree/pull/3986
# First, ensure the part instance can be deleted
p = Part.objects.get(pk=1)
p.active = False
p.save()
InvenTreeSetting.set_setting('PART_ALLOW_DELETE_FROM_ASSEMBLY', True)
self.delete(reverse('api-part-detail', kwargs={'pk': 1}))
with self.assertRaises(Part.DoesNotExist):
p.refresh_from_db()
class PricingLegacyRedirectTests(InvenTreeAPITestCase):
"""Tests for redirects from the legacy pricing endpoints (previously under /api/part/)."""
fixtures = ['category', 'part', 'location']
roles = ['part.view', 'pricing.add', 'pricing.change', 'pricing.delete']
@classmethod
def setUpTestData(cls):
"""Sale price breaks can only be created for salable parts."""
super().setUpTestData()
Part.objects.filter(pk=1).update(salable=True)
def test_redirect_urls(self):
"""Legacy URLs redirect (HTTP 308) to the new endpoints."""
pb = PartInternalPriceBreak.objects.create(part_id=1, quantity=1, price=10)
sb = PartSellPriceBreak.objects.create(part_id=1, quantity=1, price=10)
redirects = {
'/api/part/sale-price/': reverse('api-part-sale-price-list'),
f'/api/part/sale-price/{sb.pk}/': reverse(
'api-part-sale-price-detail', kwargs={'pk': sb.pk}
),
'/api/part/internal-price/': reverse('api-part-internal-price-list'),
f'/api/part/internal-price/{pb.pk}/': reverse(
'api-part-internal-price-detail', kwargs={'pk': pb.pk}
),
'/api/part/1/pricing/': reverse('api-part-pricing', kwargs={'pk': 1}),
}
for old_url, new_url in redirects.items():
self.assertTrue(new_url.startswith('/api/pricing/'))
response = self.client.get(old_url)
self.assertEqual(response.status_code, 308, old_url)
self.assertEqual(response['Location'], new_url)
# Query parameters are passed through
response = self.client.get(old_url, {'part': 1, 'ordering': '-quantity'})
self.assertEqual(response.status_code, 308, old_url)
self.assertEqual(
response['Location'], f'{new_url}?part=1&ordering=-quantity'
)
def test_redirect_methods(self):
"""The request method (and data) is preserved when following a redirect."""
old_url = '/api/part/sale-price/'
# GET
response = self.get(old_url, {'part': 1}, follow=True)
self.assertEqual(len(response.data), 0)
# POST
response = self.post(
old_url,
{'part': 1, 'quantity': 5, 'price': 12.5},
follow=True,
expected_code=201,
)
pk = response.data['pk']
self.assertEqual(PartSellPriceBreak.objects.get(pk=pk).quantity, 5)
# PATCH
self.patch(
f'/api/part/sale-price/{pk}/',
{'quantity': 7},
follow=True,
expected_code=200,
)
self.assertEqual(PartSellPriceBreak.objects.get(pk=pk).quantity, 7)
# DELETE
self.delete(f'/api/part/sale-price/{pk}/', follow=True, expected_code=204)
self.assertFalse(PartSellPriceBreak.objects.filter(pk=pk).exists())
# Part pricing detail
response = self.patch(
'/api/part/1/pricing/',
{'override_min': 1, 'override_min_currency': 'USD'},
follow=True,
expected_code=200,
)
self.assertEqual(float(response.data['override_min']), 1)
def test_redirect_permissions(self):
"""Redirects do not bypass the permission checks of the new endpoints."""
self.clearRoles()
self.assignRole('part.view')
response = self.client.get('/api/part/sale-price/')
self.assertEqual(response.status_code, 308)
self.get('/api/part/sale-price/', follow=True, expected_code=403)
self.post(
'/api/part/sale-price/',
{'part': 1, 'quantity': 5, 'price': 12.5},
follow=True,
expected_code=403,
)
def test_redirect_csrf(self):
"""The redirect itself is not subject to CSRF checks, but the new endpoint is."""
client = APIClient(enforce_csrf_checks=True)
client.force_login(self.user)
data = {'part': 1, 'quantity': 5, 'price': 12.5}
response = client.post('/api/part/sale-price/', data, format='json')
self.assertEqual(response.status_code, 308)
# Following the redirect, the new endpoint enforces CSRF for session authentication
response = client.post(
'/api/part/sale-price/', data, format='json', follow=True
)
self.assertEqual(response.status_code, 403)
self.assertFalse(PartSellPriceBreak.objects.filter(quantity=5).exists())
class NestedPricingPermissionTests(InvenTreeAPITestCase):
"""Pricing data must not be exposed via nested (embedded) part data.
Regression test: nested serializers were constructed without the requesting user,
so permission checks for their optional fields were skipped entirely.
"""
fixtures = ['category', 'part', 'location', 'bom', 'stock']
roles = ['part.view', 'bom.view', 'stock.view']
PRICING_FIELDS = ['pricing_min', 'pricing_max']
def nested_part_data(self) -> list[dict]:
"""Return nested part data from a selection of API endpoints."""
bom_item = self.get(
reverse('api-bom-item-detail', kwargs={'pk': 3}),
{'sub_part_detail': True, 'pricing': True},
).data
stock_item = self.get(
reverse('api-stock-list'), {'part_detail': True, 'limit': 1}
).data['results'][0]
return [bom_item['sub_part_detail'], stock_item['part_detail']]
def test_nested_pricing(self):
"""Nested pricing fields depend on the 'pricing' role of the requesting user."""
for data in self.nested_part_data():
self.assertIn('name', data)
for field in self.PRICING_FIELDS:
self.assertNotIn(field, data)
self.assignRole('pricing.view')
for data in self.nested_part_data():
for field in self.PRICING_FIELDS:
self.assertIn(field, data)
@@ -0,0 +1,225 @@
"""Unit tests for the 'pricing' app database migrations."""
from django_test_migrations.contrib.unittest_case import MigratorTestCase
from InvenTree import unit_test
# Models which have been moved from the 'part' app to the 'pricing' app
MODELS = ['partpricing', 'partsellpricebreak', 'partinternalpricebreak']
class TestMovePricingModels(MigratorTestCase):
"""Test that pricing data is retained when moving models from 'part' to 'pricing'."""
migrate_from = ('part', '0154_remove_part_notes')
migrate_to = ('pricing', unit_test.getNewestMigrationFile('pricing'))
def prepare(self):
"""Create pricing data against the 'part' app models."""
Part = self.old_state.apps.get_model('part', 'part')
PartPricing = self.old_state.apps.get_model('part', 'partpricing')
PartSellPriceBreak = self.old_state.apps.get_model('part', 'partsellpricebreak')
PartInternalPriceBreak = self.old_state.apps.get_model(
'part', 'partinternalpricebreak'
)
part = Part.objects.create(
name='Priced part',
description='A part with pricing',
salable=True,
level=0,
lft=0,
rght=0,
tree_id=0,
)
PartPricing.objects.create(
part=part,
overall_min=10,
overall_min_currency='USD',
overall_max=20,
overall_max_currency='USD',
)
for qty in [1, 10, 100]:
PartSellPriceBreak.objects.create(
part=part, quantity=qty, price=qty * 2, price_currency='USD'
)
PartInternalPriceBreak.objects.create(
part=part, quantity=qty, price=qty, price_currency='USD'
)
# Replicate the content types (and permissions) present in a database
# created before these models were moved to the 'pricing' app
ContentType = self.old_state.apps.get_model('contenttypes', 'contenttype')
Permission = self.old_state.apps.get_model('auth', 'permission')
ContentType.objects.filter(app_label='pricing').delete()
for model in MODELS:
ct, _ = ContentType.objects.get_or_create(app_label='part', model=model)
Permission.objects.get_or_create(
content_type=ct, codename=f'view_{model}', defaults={'name': model}
)
def test_pricing_data(self):
"""Pricing data should be available via the 'pricing' app models."""
Part = self.new_state.apps.get_model('part', 'part')
PartPricing = self.new_state.apps.get_model('pricing', 'partpricing')
PartSellPriceBreak = self.new_state.apps.get_model(
'pricing', 'partsellpricebreak'
)
PartInternalPriceBreak = self.new_state.apps.get_model(
'pricing', 'partinternalpricebreak'
)
part = Part.objects.get(name='Priced part')
pricing = PartPricing.objects.get(part=part)
self.assertEqual(pricing.overall_min.amount, 10)
self.assertEqual(pricing.overall_max.amount, 20)
self.assertEqual(PartSellPriceBreak.objects.filter(part=part).count(), 3)
self.assertEqual(PartInternalPriceBreak.objects.filter(part=part).count(), 3)
self.assertEqual(
PartSellPriceBreak.objects.get(part=part, quantity=10).price.amount, 20
)
# Reverse relations are retained
self.assertEqual(part.salepricebreaks.count(), 3)
self.assertEqual(part.internalpricebreaks.count(), 3)
self.assertEqual(part.pricing_data.pk, pricing.pk)
def test_content_types(self):
"""Content types (and linked permissions) should move to the 'pricing' app."""
ContentType = self.new_state.apps.get_model('contenttypes', 'contenttype')
Permission = self.new_state.apps.get_model('auth', 'permission')
for model in MODELS:
self.assertFalse(
ContentType.objects.filter(app_label='part', model=model).exists()
)
ct = ContentType.objects.get(app_label='pricing', model=model)
self.assertTrue(
Permission.objects.filter(
content_type=ct, codename=f'view_{model}'
).exists()
)
class TestMigrateQueuedTasks(MigratorTestCase):
"""Test that queued tasks referencing the legacy model identities are migrated."""
migrate_from = ('pricing', '0001_initial')
migrate_to = ('pricing', '0002_migrate_queued_tasks')
def prepare(self):
"""Queue tasks with legacy (pre-migration) pickled model instances."""
from django.db.models.base import model_unpickle
from django.utils import timezone
from django_q.conf import Conf
from django_q.signing import SignedPackage
from pricing.models import PartPricing, PartSellPriceBreak
OrmQ = self.old_state.apps.get_model('django_q', 'OrmQ')
def legacy(instance, model_name):
"""Return an object which pickles exactly as a pre-migration model instance."""
state = instance.__reduce__()[2]
class Legacy:
def __reduce__(self):
return (model_unpickle, (('part', model_name),), state)
return Legacy()
# Map of task name to OrmQ primary key
self.task_ids = {}
def enqueue(name, payload):
task = OrmQ.objects.create(
key=Conf.PREFIX, payload=payload, lock=timezone.now()
)
self.task_ids[name] = task.pk
# A queued pricing update, with a legacy PartPricing instance as an argument
enqueue(
'legacy-pricing',
SignedPackage.dumps({
'id': 'legacy-pricing',
'name': 'legacy-pricing',
'func': 'part.tasks.update_part_pricing',
'args': (legacy(PartPricing(pk=123, part_id=4), 'PartPricing'),),
'kwargs': {'counter': 2},
'group': 'pricing',
}),
)
# Before migration, the legacy payload cannot be loaded
with self.assertRaises(LookupError):
SignedPackage.loads(
OrmQ.objects.get(pk=self.task_ids['legacy-pricing']).payload
)
# A legacy price break nested within the task kwargs
enqueue(
'legacy-pricebreak',
SignedPackage.dumps({
'id': 'legacy-pricebreak',
'name': 'legacy-pricebreak',
'func': 'some.task',
'args': (),
'kwargs': {
'items': [
legacy(
PartSellPriceBreak(pk=7, part_id=4), 'PartSellPriceBreak'
)
]
},
}),
)
# An unrelated task, which should not be modified
self.unrelated = SignedPackage.dumps({
'id': 'unrelated',
'name': 'unrelated',
'func': 'some.other.task',
'args': (1, 'a'),
'kwargs': {},
})
enqueue('unrelated', self.unrelated)
# A payload which cannot be decoded at all, which should not be modified
enqueue('invalid', 'not-a-valid-payload')
def test_queued_tasks(self):
"""Legacy payloads are rewritten, and other payloads are left untouched."""
from django_q.signing import SignedPackage
from pricing.models import PartPricing, PartSellPriceBreak
OrmQ = self.new_state.apps.get_model('django_q', 'OrmQ')
tasks = {name: OrmQ.objects.get(pk=pk) for name, pk in self.task_ids.items()}
self.assertEqual(len(tasks), 4)
data = SignedPackage.loads(tasks['legacy-pricing'].payload)
pricing = data['args'][0]
self.assertIsInstance(pricing, PartPricing)
self.assertEqual(pricing.pk, 123)
self.assertEqual(pricing.part_id, 4)
self.assertEqual(data['kwargs'], {'counter': 2})
self.assertEqual(data['func'], 'part.tasks.update_part_pricing')
data = SignedPackage.loads(tasks['legacy-pricebreak'].payload)
price_break = data['kwargs']['items'][0]
self.assertIsInstance(price_break, PartSellPriceBreak)
self.assertEqual(price_break.pk, 7)
self.assertEqual(tasks['unrelated'].payload, self.unrelated)
self.assertEqual(tasks['invalid'].payload, 'not-a-valid-payload')
@@ -16,6 +16,7 @@ import stock.models
from common.settings import set_global_setting
from InvenTree.unit_test import InvenTreeTestCase
from order.status_codes import PurchaseOrderStatus
from pricing.models import PartInternalPriceBreak, PartPricing
class PartPricingTests(InvenTreeTestCase):
@@ -224,7 +225,7 @@ class PartPricingTests(InvenTreeTestCase):
for ii in range(5):
# Let's add some internal price breaks
part.models.PartInternalPriceBreak.objects.create(
PartInternalPriceBreak.objects.create(
part=self.part, quantity=ii + 1, price=10 - ii, price_currency=currency
)
@@ -423,7 +424,7 @@ class PartPricingTests(InvenTreeTestCase):
self.part.delete()
# Check that the pricing was removed also
with self.assertRaises(part.models.PartPricing.DoesNotExist):
with self.assertRaises(PartPricing.DoesNotExist):
pricing.refresh_from_db()
@override_settings(TESTING_PRICING=True)
@@ -459,12 +460,12 @@ class PartPricingTests(InvenTreeTestCase):
)
# Ensure there is no pricing data
part.models.PartPricing.objects.all().delete()
PartPricing.objects.all().delete()
check_missing_pricing()
# Check that PartPricing objects have been created
self.assertEqual(part.models.PartPricing.objects.count(), 101)
self.assertEqual(PartPricing.objects.count(), 101)
def test_check_missing_pricing_batches_scheduling(self):
"""check_missing_pricing() must batch its scheduling calls, not scan the task queue per part.
@@ -486,7 +487,7 @@ class PartPricingTests(InvenTreeTestCase):
)
# Ensure there is no pricing data
part.models.PartPricing.objects.all().delete()
PartPricing.objects.all().delete()
with self.captureOnCommitCallbacks(execute=True):
with mock.patch('InvenTree.tasks.check_existing_task') as mock_check:
@@ -496,7 +497,7 @@ class PartPricingTests(InvenTreeTestCase):
mock_check.assert_not_called()
# PartPricing objects were still created and scheduled for a background update
self.assertEqual(part.models.PartPricing.objects.count(), 21)
self.assertEqual(PartPricing.objects.count(), 21)
@override_settings(TESTING_PRICING=True)
def test_delete_part_with_stock_items(self):
@@ -522,17 +523,17 @@ class PartPricingTests(InvenTreeTestCase):
p.schedule_pricing_update(create=True)
# Check that a PartPricing object exists
self.assertTrue(part.models.PartPricing.objects.filter(part=p).exists())
self.assertTrue(PartPricing.objects.filter(part=p).exists())
# Delete the part
p.delete()
# Check that the PartPricing object has been deleted
self.assertFalse(part.models.PartPricing.objects.filter(part_id=p.pk).exists())
self.assertFalse(PartPricing.objects.filter(part_id=p.pk).exists())
# Try to update pricing (should fail gracefully as the Part has been deleted)
p.schedule_pricing_update(create=False)
self.assertFalse(part.models.PartPricing.objects.filter(part_id=p.pk).exists())
self.assertFalse(PartPricing.objects.filter(part_id=p.pk).exists())
@override_settings(TESTING_PRICING=True)
def test_get_part_deleted(self):
@@ -594,7 +595,7 @@ class PartPricingTests(InvenTreeTestCase):
pricing.update_pricing()
# As the part no longer exists, no PartPricing row should have been created
self.assertFalse(part.models.PartPricing.objects.filter(part_id=p.pk).exists())
self.assertFalse(PartPricing.objects.filter(part_id=p.pk).exists())
@override_settings(TESTING_PRICING=True)
def test_multi_level_bom(self):
@@ -674,3 +675,22 @@ class PartPricingTests(InvenTreeTestCase):
self.assertEqual(A1.pricing.overall_min, Money(a_min, 'USD'))
self.assertEqual(A1.pricing.overall_max, Money(a_max, 'USD'))
def test_max_pricing_depth(self):
"""Exceeding the maximum pricing depth must skip the update cleanly.
Regression test: the debug log message was previously passed with the
arguments in the wrong order, raising a TypeError when debug logging was enabled.
"""
pricing = self.part.pricing
pricing.save()
with self.assertLogs('inventree', level='DEBUG') as logs:
pricing.schedule_for_update(counter=PartPricing.MAX_PRICING_DEPTH + 1)
self.assertTrue(
any('maximum depth exceeded' in message for message in logs.output)
)
pricing.refresh_from_db()
self.assertFalse(pricing.scheduled_for_update)
@@ -0,0 +1,66 @@
# Generated by Django 5.2.17 on 2026-10-16 23:35
import users.ruleset
from django.db import migrations, models
def copy_part_role_to_pricing(apps, schema_editor):
"""Grant the new 'pricing' role to any group which currently holds the 'part' role.
Pricing models were previously covered by the 'part' role,
so existing groups retain the same level of access to pricing data.
"""
RuleSet = apps.get_model('users', 'RuleSet')
for ruleset in RuleSet.objects.filter(name='part'):
RuleSet.objects.update_or_create(
group=ruleset.group,
name='pricing',
defaults={
'can_view': ruleset.can_view,
'can_add': ruleset.can_add,
'can_change': ruleset.can_change,
'can_delete': ruleset.can_delete,
},
)
def remove_pricing_role(apps, schema_editor):
"""Remove the 'pricing' role (reverse migration)."""
RuleSet = apps.get_model('users', 'RuleSet')
RuleSet.objects.filter(name='pricing').delete()
class Migration(migrations.Migration):
dependencies = [
("users", "0018_tippresult"),
]
operations = [
migrations.AlterField(
model_name="ruleset",
name="name",
field=models.CharField(
choices=[
(users.ruleset.RuleSetEnum["ADMIN"], "Admin"),
(users.ruleset.RuleSetEnum["PART_CATEGORY"], "Part Categories"),
(users.ruleset.RuleSetEnum["PART"], "Parts"),
(users.ruleset.RuleSetEnum["PRICING"], "Pricing"),
(users.ruleset.RuleSetEnum["BOM"], "Bills of Material"),
(users.ruleset.RuleSetEnum["STOCK_LOCATION"], "Stock Locations"),
(users.ruleset.RuleSetEnum["STOCK"], "Stock Items"),
(users.ruleset.RuleSetEnum["BUILD"], "Build Orders"),
(users.ruleset.RuleSetEnum["PURCHASE_ORDER"], "Purchase Orders"),
(users.ruleset.RuleSetEnum["SALES_ORDER"], "Sales Orders"),
(users.ruleset.RuleSetEnum["RETURN_ORDER"], "Return Orders"),
(users.ruleset.RuleSetEnum["TRANSFER_ORDER"], "Transfer Orders"),
],
help_text="Permission set",
max_length=50,
),
),
migrations.RunPython(
copy_part_role_to_pricing, reverse_code=remove_pricing_role
),
]
@@ -13,6 +13,7 @@ _roles = {
'admin': 'Role Admin',
'part_category': 'Role Part Categories',
'part': 'Role Parts',
'pricing': 'Role Pricing',
'stock_location': 'Role Stock Locations',
'stock': 'Role Stock Items',
'bom': 'Role Bills of Material',
+8 -3
View File
@@ -13,6 +13,7 @@ class RuleSetEnum(StringEnum):
ADMIN: Assigning user permissions, and other administrative tasks
PART_CATEGORY: Accessing Part Category data
PART: Accessing Part data
PRICING: Accessing Part pricing data (cached pricing, sale price breaks and internal price breaks)
BOM: Accessing Bill of Materials data
STOCK_LOCATION: Accessing Stock Location data
STOCK: Accessing Stock Item data
@@ -26,6 +27,7 @@ class RuleSetEnum(StringEnum):
ADMIN = 'admin'
PART_CATEGORY = 'part_category'
PART = 'part'
PRICING = 'pricing'
BOM = 'bom'
STOCK_LOCATION = 'stock_location'
STOCK = 'stock'
@@ -42,6 +44,7 @@ RULESET_CHOICES = [
(RuleSetEnum.ADMIN, _('Admin')),
(RuleSetEnum.PART_CATEGORY, _('Part Categories')),
(RuleSetEnum.PART, _('Parts')),
(RuleSetEnum.PRICING, _('Pricing')),
(RuleSetEnum.BOM, _('Bills of Material')),
(RuleSetEnum.STOCK_LOCATION, _('Stock Locations')),
(RuleSetEnum.STOCK, _('Stock Items')),
@@ -131,9 +134,6 @@ def get_ruleset_models() -> dict:
],
RuleSetEnum.PART: [
'part_part',
'part_partpricing',
'part_partsellpricebreak',
'part_partinternalpricebreak',
'part_parttesttemplate',
'part_partrelated',
'part_partstar',
@@ -142,6 +142,11 @@ def get_ruleset_models() -> dict:
'company_supplierpart',
'company_manufacturerpart',
],
RuleSetEnum.PRICING: [
'pricing_partpricing',
'pricing_partsellpricebreak',
'pricing_partinternalpricebreak',
],
RuleSetEnum.STOCK_LOCATION: ['stock_stocklocation', 'stock_stocklocationtype'],
RuleSetEnum.STOCK: [
'stock_stockitem',
@@ -41,3 +41,53 @@ class TestForwardMigrations(MigratorTestCase):
# a new token should be a v2 token now
new_token = ApiToken.objects.create(key='new-token', user=token.user)
self.assertEqual(new_token.token_version, 2)
class TestPricingRoleMigration(MigratorTestCase):
"""Test that the 'pricing' role is copied from the existing 'part' role."""
migrate_from = ('users', '0018_tippresult')
migrate_to = ('users', '0019_pricing_role')
def prepare(self):
"""Create groups with a variety of 'part' role permissions."""
Group = self.old_state.apps.get_model('auth', 'group')
RuleSet = self.old_state.apps.get_model('users', 'RuleSet')
viewer = Group.objects.create(name='viewer')
editor = Group.objects.create(name='editor')
nobody = Group.objects.create(name='nobody')
RuleSet.objects.create(group=viewer, name='part', can_view=True)
RuleSet.objects.create(
group=editor,
name='part',
can_view=True,
can_add=True,
can_change=True,
can_delete=True,
)
RuleSet.objects.create(group=nobody, name='part')
# A group without a 'part' ruleset at all
Group.objects.create(name='empty')
def test_pricing_role(self):
"""Pricing role permissions should match the part role for each group."""
RuleSet = self.new_state.apps.get_model('users', 'RuleSet')
self.assertEqual(RuleSet.objects.filter(name='pricing').count(), 3)
for name in ['viewer', 'editor', 'nobody']:
part = RuleSet.objects.get(group__name=name, name='part')
pricing = RuleSet.objects.get(group__name=name, name='pricing')
for perm in ['can_view', 'can_add', 'can_change', 'can_delete']:
self.assertEqual(getattr(part, perm), getattr(pricing, perm))
self.assertTrue(
RuleSet.objects.get(group__name='editor', name='pricing').can_delete
)
self.assertFalse(
RuleSet.objects.filter(group__name='empty', name='pricing').exists()
)
+3 -3
View File
@@ -116,12 +116,12 @@ export enum ApiEndpoints {
// Part API endpoints
part_list = 'part/',
part_thumbs_list = 'part/thumbs/',
part_pricing = 'part/:id/pricing/',
part_pricing = 'pricing/part/:id/',
part_requirements = 'part/:id/requirements/',
part_serial_numbers = 'part/:id/serial-numbers/',
part_scheduling = 'part/:id/scheduling/',
part_pricing_internal = 'part/internal-price/',
part_pricing_sale = 'part/sale-price/',
part_pricing_internal = 'pricing/internal-price/',
part_pricing_sale = 'pricing/sale-price/',
part_stocktake_list = 'part/stocktake/',
part_stocktake_generate = 'part/stocktake/generate/',
category_list = 'part/category/',
+3
View File
@@ -9,6 +9,7 @@ export enum UserRoles {
build = 'build',
part = 'part',
part_category = 'part_category',
pricing = 'pricing',
purchase_order = 'purchase_order',
return_order = 'return_order',
transfer_order = 'transfer_order',
@@ -37,6 +38,8 @@ export function userRoleLabel(role: UserRoles): string {
return t`Parts`;
case UserRoles.part_category:
return t`Part Categories`;
case UserRoles.pricing:
return t`Pricing`;
case UserRoles.purchase_order:
return t`Purchase Orders`;
case UserRoles.return_order:
@@ -372,6 +372,7 @@ export default function PartDetail() {
name: 'pricing',
label: t`Part Pricing`,
icon: <IconCurrencyDollar />,
hidden: !user.hasViewRole(UserRoles.pricing),
content: part ? <PartPricingPanel part={part} /> : <Skeleton />
},
{
@@ -112,7 +112,7 @@ export default function PriceBreakPanel({
onClick={() => {
newPriceBreak.open();
}}
hidden={!user.hasAddRole(UserRoles.part)}
hidden={!user.hasAddRole(UserRoles.pricing)}
/>
];
}, [user]);
@@ -121,14 +121,14 @@ export default function PriceBreakPanel({
(record: any): RowAction[] => {
return [
RowEditAction({
hidden: !user.hasChangeRole(UserRoles.part),
hidden: !user.hasChangeRole(UserRoles.pricing),
onClick: () => {
setSelectedPriceBreak(record.pk);
editPriceBreak.open();
}
}),
RowDeleteAction({
hidden: !user.hasDeleteRole(UserRoles.part),
hidden: !user.hasDeleteRole(UserRoles.pricing),
onClick: () => {
setSelectedPriceBreak(record.pk);
deletePriceBreak.open();
@@ -25,6 +25,7 @@ import { DataTable } from 'mantine-datatable';
import { type ReactNode, useCallback, useMemo } from 'react';
import { ApiEndpoints } from '@lib/enums/ApiEndpoints';
import { UserRoles } from '@lib/enums/Roles';
import { apiUrl } from '@lib/functions/Api';
import type { ApiFormFieldSet } from '@lib/types/Forms';
import { api } from '../../../App';
@@ -37,6 +38,7 @@ import { formatCurrency, formatDate } from '../../../defaults/formatters';
import { InvenTreeIcon } from '../../../functions/icons';
import { useEditApiFormModal } from '../../../hooks/UseForm';
import { useGlobalSettingsState } from '../../../states/SettingsStates';
import { useUserState } from '../../../states/UserState';
import { panelOptions } from '../PartPricingPanel';
interface PricingOverviewEntry {
@@ -65,6 +67,7 @@ export default function PricingOverviewPanel({
doNavigation: (panel: panelOptions) => void;
}>): ReactNode {
const globalSettings = useGlobalSettingsState();
const user = useUserState();
const refreshPricing = useCallback(() => {
const url = apiUrl(ApiEndpoints.part_pricing, part.pk);
@@ -305,6 +308,7 @@ export default function PricingOverviewPanel({
{
name: t`Refresh`,
tooltip: t`Refresh pricing data`,
hidden: !user.hasChangeRole(UserRoles.pricing),
icon: (
<InvenTreeIcon
icon='refresh'
@@ -316,6 +320,7 @@ export default function PricingOverviewPanel({
}
},
EditItemAction({
hidden: !user.hasChangeRole(UserRoles.pricing),
onClick: () => {
editPricing.open();
},
+1
View File
@@ -326,6 +326,7 @@ def builtin_apps():
'machine',
'order',
'part',
'pricing',
'report',
'stock',
'users',