From 89cc999c423f5b7b7fb15e8a6b5e04c78415863a Mon Sep 17 00:00:00 2001 From: caoqianming Date: Thu, 30 Jul 2026 13:22:28 +0800 Subject: [PATCH] Prevent duplicate warehouse inventory creation --- apps/inm/models.py | 111 ++++++++++++++++++++++++++++++++++- apps/inm/services.py | 6 +- apps/inm/services_daoru.py | 4 +- apps/inm/tests.py | 116 ++++++++++++++++++++++++++++++++++++- apps/qm/services.py | 2 +- 5 files changed, 230 insertions(+), 9 deletions(-) diff --git a/apps/inm/models.py b/apps/inm/models.py index e0d2cd3d..5148c42c 100644 --- a/apps/inm/models.py +++ b/apps/inm/models.py @@ -1,4 +1,6 @@ -from django.db import models +import json + +from django.db import connection, models from apps.utils.models import BaseModel, CommonBModel, CommonBDModel, CommonADModel from apps.pum.models import Supplier, PuOrder from apps.sam.models import Customer, Order @@ -38,6 +40,113 @@ class MaterialBatch(BaseModel): material_ofrom = models.ForeignKey(Material, verbose_name='原料物料', on_delete=models.SET_NULL, null=True, blank=True, related_name='mb_mofrom') defect = models.ForeignKey('qm.defect', verbose_name='缺陷', on_delete=models.PROTECT, null=True, blank=True) + INVENTORY_KEY_FIELDS = ( + 'material', + 'batch', + 'warehouse', + 'state', + 'defect', + ) + + @classmethod + def _normalize_inventory_lookup(cls, **kwargs): + """生成唯一、完整的仓库库存业务键。""" + unknown_fields = set(kwargs) - set(cls.INVENTORY_KEY_FIELDS) + if unknown_fields: + fields = ', '.join(sorted(unknown_fields)) + raise TypeError(f'不支持的仓库库存定位字段: {fields}') + + required_fields = ('material', 'batch', 'warehouse') + missing_fields = [ + field for field in required_fields + if kwargs.get(field) is None + ] + if missing_fields: + fields = ', '.join(missing_fields) + raise ValueError(f'仓库库存业务键必须包含: {fields}') + + lookup = { + field: kwargs.get(field) + for field in cls.INVENTORY_KEY_FIELDS + } + if lookup['state'] is None: + lookup['state'] = cls._meta.get_field('state').get_default() + return lookup + + @classmethod + def _inventory_advisory_lock_payload(cls, lookup): + lock_values = {} + for name in cls.INVENTORY_KEY_FIELDS: + field = cls._meta.get_field(name) + value = lookup[name] + if field.is_relation and value is not None: + value = getattr(value, 'pk', value) + lock_values[field.attname] = value + return json.dumps( + {'model': cls._meta.label_lower, 'lookup': lock_values}, + sort_keys=True, + ensure_ascii=False, + default=str, + separators=(',', ':'), + ) + + @classmethod + def locked_get_or_create_inventory(cls, defaults=None, **kwargs): + """ + 在事务中按完整业务键获取或创建仓库库存。 + + 已存在记录使用行锁；首次创建使用 PostgreSQL 事务级 advisory lock， + 并在取得锁后重新查询，避免并发创建重复库存。 + """ + if not connection.in_atomic_block: + raise RuntimeError( + 'locked_get_or_create_inventory 必须在事务中调用' + ) + if connection.vendor != 'postgresql': + raise RuntimeError( + 'locked_get_or_create_inventory 仅支持 PostgreSQL' + ) + + defaults = defaults or {} + lookup = cls._normalize_inventory_lookup(**kwargs) + create_defaults = { + key: value + for key, value in defaults.items() + if key not in cls.INVENTORY_KEY_FIELDS + } + + rows = list( + cls.objects.select_for_update().filter(**lookup)[:2] + ) + if len(rows) > 1: + raise RuntimeError( + f'{cls.__name__} 数据异常：库存业务键 {lookup} 命中多条' + ) + if rows: + return rows[0], False + + lock_payload = cls._inventory_advisory_lock_payload(lookup) + with connection.cursor() as cursor: + cursor.execute( + 'SELECT pg_advisory_xact_lock(hashtextextended(%s, 0))', + [lock_payload], + ) + + rows = list( + cls.objects.select_for_update().filter(**lookup)[:2] + ) + if len(rows) > 1: + raise RuntimeError( + f'{cls.__name__} 数据异常：库存业务键 {lookup} 命中多条' + ) + if rows: + return rows[0], False + + return cls.objects.create( + **lookup, + **create_defaults, + ), True + @property def count_mioing(self): diff --git a/apps/inm/services.py b/apps/inm/services.py index eaf2fd24..965e2413 100644 --- a/apps/inm/services.py +++ b/apps/inm/services.py @@ -223,7 +223,7 @@ def do_in(item: MIOItem): # 增加mb if not is_zhj: - mb, _ = MaterialBatch.objects.get_or_create( + mb, _ = MaterialBatch.locked_get_or_create_inventory( material=xmaterial, warehouse=item.warehouse, batch=xbatch, @@ -258,7 +258,7 @@ def do_in(item: MIOItem): if is_zhj: # 组合件单独处理并且不做追踪单个处理 - mb, is_created = MaterialBatch.objects.get_or_create( + mb, is_created = MaterialBatch.locked_get_or_create_inventory( material=item.material, warehouse=item.warehouse, batch=item.batch, @@ -415,7 +415,7 @@ class InmService: state = WMaterial.WM_OK if defect and defect.okcate in [Defect.DEFECT_NOTOK]: state = WMaterial.WM_NOTOK - mb, _ = MaterialBatch.objects.get_or_create( + mb, _ = MaterialBatch.locked_get_or_create_inventory( material=material, warehouse=warehouse, batch=batch, diff --git a/apps/inm/services_daoru.py b/apps/inm/services_daoru.py index 34da62d6..7d913615 100644 --- a/apps/inm/services_daoru.py +++ b/apps/inm/services_daoru.py @@ -55,7 +55,7 @@ def daoru_mb(path: str): process=process, defaults={"type": type, "name": name, "specification": specification, "model": model, "process": process, "number": ranstr(6), "id": idWorker.get_id()}, ) - MaterialBatch.objects.get_or_create( + MaterialBatch.locked_get_or_create_inventory( material=material, batch=batch, warehouse=warehouse, defaults={"material": material, "batch": batch, "warehouse": warehouse, "count": count, "id": idWorker.get_id()} ) i = i + 1 @@ -174,4 +174,4 @@ def daoru_mioitems(path:str, mio:MIO): unit_price=material.unit_price, id=idWorker.get_id())) ind = ind + 1 - MIOItem.objects.bulk_create(mioitems) \ No newline at end of file + MIOItem.objects.bulk_create(mioitems) diff --git a/apps/inm/tests.py b/apps/inm/tests.py index 7ce503c2..c00fa963 100644 --- a/apps/inm/tests.py +++ b/apps/inm/tests.py @@ -1,3 +1,115 @@ -from django.test import TestCase +from concurrent.futures import ThreadPoolExecutor +from decimal import Decimal +from threading import Barrier +from unittest import skipUnless -# Create your tests here. +from django.db import connection, connections, transaction +from django.test import SimpleTestCase, TransactionTestCase + +from apps.inm.models import MaterialBatch, WareHouse +from apps.mtm.models import Material + + +class MaterialBatchInventoryKeyTests(SimpleTestCase): + def setUp(self): + self.material = Material(id='100', name='测试物料') + self.warehouse = WareHouse( + id='200', + number='TEST', + name='测试仓库', + place='测试地点', + ) + + def test_inventory_key_normalizes_omitted_optional_fields(self): + omitted = MaterialBatch._normalize_inventory_lookup( + material=self.material, + batch='BATCH-001', + warehouse=self.warehouse, + ) + explicit = MaterialBatch._normalize_inventory_lookup( + material=self.material, + batch='BATCH-001', + warehouse=self.warehouse, + state=10, + defect=None, + ) + + self.assertEqual(omitted, explicit) + self.assertEqual( + MaterialBatch._inventory_advisory_lock_payload(omitted), + MaterialBatch._inventory_advisory_lock_payload(explicit), + ) + + def test_inventory_key_requires_material_batch_and_warehouse(self): + required_values = { + 'material': self.material, + 'batch': 'BATCH-001', + 'warehouse': self.warehouse, + } + + for field in required_values: + with self.subTest(field=field): + lookup = required_values.copy() + lookup[field] = None + with self.assertRaisesRegex(ValueError, field): + MaterialBatch._normalize_inventory_lookup(**lookup) + + def test_inventory_key_rejects_unknown_fields(self): + with self.assertRaisesRegex(TypeError, 'supplier'): + MaterialBatch._normalize_inventory_lookup( + material=self.material, + batch='BATCH-001', + warehouse=self.warehouse, + supplier=None, + ) + + +@skipUnless( + connection.vendor == 'postgresql', + 'advisory lock concurrency test requires PostgreSQL', +) +class MaterialBatchConcurrencyTests(TransactionTestCase): + def test_concurrent_first_create_uses_one_inventory_record(self): + material = Material.objects.create(name='仓库并发测试物料') + warehouse = WareHouse.objects.create( + number='CONCURRENT', + name='并发测试仓库', + place='测试地点', + ) + barrier = Barrier(2) + + def create_inventory(): + connections.close_all() + try: + barrier.wait() + with transaction.atomic(): + mb, created = ( + MaterialBatch.locked_get_or_create_inventory( + material=material, + batch='CONCURRENT-001', + warehouse=warehouse, + defaults={'count': Decimal('0')}, + ) + ) + mb.count += Decimal('1') + mb.save(update_fields=['count']) + return created + finally: + connections.close_all() + + with ThreadPoolExecutor(max_workers=2) as executor: + created_results = list(executor.map( + lambda _: create_inventory(), + range(2), + )) + + queryset = MaterialBatch.objects.filter( + material=material, + batch='CONCURRENT-001', + warehouse=warehouse, + state=10, + defect=None, + ) + self.assertEqual(queryset.count(), 1) + self.assertEqual(queryset.get().count, Decimal('2')) + self.assertCountEqual(created_results, [True, False]) diff --git a/apps/qm/services.py b/apps/qm/services.py index fdae864f..62cf8574 100644 --- a/apps/qm/services.py +++ b/apps/qm/services.py @@ -146,7 +146,7 @@ def ftestwork_submit(ins:FtestWork, user: User): mbstate = WMaterial.WM_OK if item.defect.okcate == Defect.DEFECT_NOTOK: mbstate = WMaterial.WM_NOTOK - mbx, new_create = MaterialBatch.objects.get_or_create( + mbx, new_create = MaterialBatch.locked_get_or_create_inventory( material=mb.material, warehouse=mb.warehouse, batch=mb.batch,