niuniu/ShiErFeng
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复：强化库存事务并发与日期边界

Browse Source
This commit is contained in:
焦龙言
2026-07-26 13:54:42 +08:00
parent 85905d85ca
commit 6c2a5fa344
5 changed files with 187 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
shierfeng-fastapi-backend/module_business/dao/inventory_dao.py
+13 -4
View File
@@ -1,4 +1,4 @@
from sqlalchemy import func, select
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from module_business.entity.do.master_do import BizPallet
@@ -15,7 +15,10 @@ class InventoryDao:
pallet_id: int,
) -> BizPalletInventory | None:
result = await db.execute(
select(BizPalletInventory).where(BizPalletInventory.pallet_id == pallet_id).with_for_update()
select(BizPalletInventory)
.where(BizPalletInventory.pallet_id == pallet_id)
.with_for_update()
.execution_options(populate_existing=True)
)
return result.scalar_one_or_none()
@@ -34,6 +37,7 @@ class InventoryDao:
.where(BizPalletInventory.pallet_id.in_(ordered_ids))
.order_by(BizPalletInventory.pallet_id)
.with_for_update()
.execution_options(populate_existing=True)
)
return sorted(result.scalars().all(), key=lambda inventory: inventory.pallet_id)
@@ -62,6 +66,7 @@ class InventoryDao:
.where(BizPallet.pallet_id.in_(ordered_ids))
.order_by(BizPallet.pallet_id)
.with_for_update()
.execution_options(populate_existing=True)
)
return sorted(result.scalars().all(), key=lambda pallet: pallet.pallet_id)
@@ -89,9 +94,13 @@ class InventoryDao:
) -> BizPalletEvent:
"""在调用方持有库存行锁时，为托盘追加下一序号事件。"""
sequence_result = await db.execute(
select(func.max(BizPalletEvent.event_seq)).where(BizPalletEvent.pallet_id == inventory.pallet_id)
select(BizPalletEvent.event_seq)
.where(BizPalletEvent.pallet_id == inventory.pallet_id)
.order_by(BizPalletEvent.event_seq.desc())
.limit(1)
.with_for_update()
)
last_sequence = sequence_result.scalar_one()
last_sequence = sequence_result.scalar_one_or_none()
event.pallet_id = inventory.pallet_id
event.event_seq = (last_sequence or 0) + 1
db.add(event)
shierfeng-fastapi-backend/module_business/dao/operation_dao.py
+13 -6
View File
@@ -1,4 +1,4 @@
from datetime import datetime, time
from datetime import datetime, time, timedelta
from typing import Any
from sqlalchemy import ColumnElement, Select, delete, select, update
@@ -45,10 +45,10 @@ class OperationDao:
conditions.append(column >= begin_at)
if end_time:
end_at = datetime.combine(
datetime.strptime(end_time, '%Y-%m-%d').date(),
time(23, 59, 59),
datetime.strptime(end_time, '%Y-%m-%d').date() + timedelta(days=1),
time.min,
)
conditions.append(column <= end_at)
conditions.append(column < end_at)
return conditions
@staticmethod
@@ -74,7 +74,12 @@ class OperationDao:
record_id: int,
*extra_conditions: ColumnElement[bool],
) -> Any | None:
result = await db.execute(select(model).where(id_column == record_id, *extra_conditions).with_for_update())
result = await db.execute(
select(model)
.where(id_column == record_id, *extra_conditions)
.with_for_update()
.execution_options(populate_existing=True)
)
return result.scalar_one_or_none()
@staticmethod
@@ -135,7 +140,7 @@ class OperationDao:
payload = {
column.key: vars(record)[column.key]
for column in model.__table__.columns
if column.key != primary_key and column.key in vars(record)
if column.key not in {primary_key, 'create_by', 'create_time'} and column.key in vars(record)
}
if payload:
await db.execute(update(model).where(id_column == getattr(record, primary_key)).values(**payload))
@@ -370,6 +375,7 @@ class OperationDao:
@classmethod
async def delete_inbound(cls, db: AsyncSession, inbound_id: int) -> None:
await db.execute(delete(BizInboundItem).where(BizInboundItem.inbound_id == inbound_id))
await cls._delete(db, BizInboundOrder, BizInboundOrder.inbound_id, inbound_id)
@classmethod
@@ -520,6 +526,7 @@ class OperationDao:
@classmethod
async def delete_shipment(cls, db: AsyncSession, shipment_id: int) -> None:
await db.execute(delete(BizShipmentItem).where(BizShipmentItem.shipment_id == shipment_id))
await cls._delete(
db,
BizShipmentOrder,
shierfeng-fastapi-backend/module_business/entity/vo/operation_vo.py
+30
View File
@@ -16,6 +16,26 @@ def _ensure_unique_ids(value: list[int]) -> list[int]:
return value
def _parse_page_query_date(value: str) -> datetime:
try:
parsed = datetime.strptime(value, '%Y-%m-%d')
except ValueError as exc:
raise ValueError('日期必须为有效的 YYYY-MM-DD 格式') from exc
if parsed.strftime('%Y-%m-%d') != value:
raise ValueError('日期必须为有效的 YYYY-MM-DD 格式')
return parsed
def _validate_page_query_date_range(value: BaseModel) -> BaseModel:
begin_value = value.begin_time
end_value = value.end_time
begin_date = _parse_page_query_date(begin_value) if begin_value else None
end_date = _parse_page_query_date(end_value) if end_value else None
if begin_date and end_date and begin_date > end_date:
raise ValueError('开始日期不能晚于结束日期')
return value
class InboundCreateModel(BaseModel):
"""首次入库草稿新增模型。"""
@@ -335,6 +355,8 @@ class InboundPageQueryModel(BaseModel):
begin_time: str | None = Field(default=None, min_length=1, pattern=r'\S', description='开始时间')
end_time: str | None = Field(default=None, min_length=1, pattern=r'\S', description='结束时间')
_validate_date_range = model_validator(mode='after')(_validate_page_query_date_range)
class FillingPageQueryModel(BaseModel):
"""托盘装填分页查询模型。"""
@@ -351,6 +373,8 @@ class FillingPageQueryModel(BaseModel):
begin_time: str | None = Field(default=None, min_length=1, pattern=r'\S', description='开始时间')
end_time: str | None = Field(default=None, min_length=1, pattern=r'\S', description='结束时间')
_validate_date_range = model_validator(mode='after')(_validate_page_query_date_range)
class ShipmentPageQueryModel(BaseModel):
"""整托发货分页查询模型。"""
@@ -366,6 +390,8 @@ class ShipmentPageQueryModel(BaseModel):
begin_time: str | None = Field(default=None, min_length=1, pattern=r'\S', description='开始时间')
end_time: str | None = Field(default=None, min_length=1, pattern=r'\S', description='结束时间')
_validate_date_range = model_validator(mode='after')(_validate_page_query_date_range)
class UsagePageQueryModel(BaseModel):
"""客户现场使用分页查询模型。"""
@@ -381,6 +407,8 @@ class UsagePageQueryModel(BaseModel):
begin_time: str | None = Field(default=None, min_length=1, pattern=r'\S', description='开始时间')
end_time: str | None = Field(default=None, min_length=1, pattern=r'\S', description='结束时间')
_validate_date_range = model_validator(mode='after')(_validate_page_query_date_range)
class ReturnPageQueryModel(BaseModel):
"""托盘退回分页查询模型。"""
@@ -396,6 +424,8 @@ class ReturnPageQueryModel(BaseModel):
begin_time: str | None = Field(default=None, min_length=1, pattern=r'\S', description='开始时间')
end_time: str | None = Field(default=None, min_length=1, pattern=r'\S', description='结束时间')
_validate_date_range = model_validator(mode='after')(_validate_page_query_date_range)
class InboundScanQueryModel(BaseModel):
"""首次入库扫码查询模型。"""
shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py
+92 -25
View File
@@ -1,11 +1,13 @@
import inspect
import unittest
from datetime import datetime
from types import SimpleNamespace
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
from sqlalchemy.dialects import mysql, postgresql
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.sql import operators
from module_business.dao import inventory_dao as inventory_dao_module
from module_business.dao import operation_dao as operation_dao_module
@@ -189,6 +191,7 @@ class TestInventoryDao(unittest.IsolatedAsyncioTestCase):
self.assertIs(inventory, actual)
statement = self.db.execute.await_args.args[0]
self.assertTrue(statement.get_final_froms()[0].compare(BizPalletInventory.__table__))
self.assertIs(True, statement.get_execution_options().get('populate_existing'))
for dialect in (mysql.dialect(), postgresql.dialect()):
sql = compile_sql(statement, dialect)
self.assertIn('biz_pallet_inventory.pallet_id = 7', sql)
@@ -209,6 +212,7 @@ class TestInventoryDao(unittest.IsolatedAsyncioTestCase):
self.assertEqual([2, 5, 9], [inventory.pallet_id for inventory in actual])
statement = self.db.execute.await_args.args[0]
self.assertTrue(statement.get_final_froms()[0].compare(BizPalletInventory.__table__))
self.assertIs(True, statement.get_execution_options().get('populate_existing'))
for dialect in (mysql.dialect(), postgresql.dialect()):
sql = compile_sql(statement, dialect)
self.assertIn('biz_pallet_inventory.pallet_id IN (2, 5, 9)', sql)
@@ -249,6 +253,7 @@ class TestInventoryDao(unittest.IsolatedAsyncioTestCase):
self.assertEqual([2, 5, 9], [pallet.pallet_id for pallet in actual])
statement = self.db.execute.await_args.args[0]
self.assertTrue(statement.get_final_froms()[0].compare(BizPallet.__table__))
self.assertIs(True, statement.get_execution_options().get('populate_existing'))
for dialect in (mysql.dialect(), postgresql.dialect()):
sql = compile_sql(statement, dialect)
self.assertIn('biz_pallet.pallet_id IN (2, 5, 9)', sql)
@@ -283,7 +288,6 @@ class TestInventoryDao(unittest.IsolatedAsyncioTestCase):
sequence_result = MagicMock()
sequence_result.scalar_one.return_value = 4
sequence_result.scalar_one_or_none.return_value = 4
sequence_result.scalar.return_value = 4
self.db.execute.return_value = sequence_result
inventory = BizPalletInventory(pallet_id=7, last_event_id=None)
event = BizPalletEvent(pallet_id=99, event_seq=99, idempotency_key='event-key')
@@ -306,15 +310,19 @@ class TestInventoryDao(unittest.IsolatedAsyncioTestCase):
self.db.add.assert_called_once_with(event)
self.assertEqual(2, self.db.flush.await_count)
statement = self.db.execute.await_args.args[0]
sql = compile_sql(statement)
self.assertIn('max(biz_pallet_event.event_seq)', sql.lower())
self.assertIn('biz_pallet_event.pallet_id = 7', sql)
self.assertTrue(statement.selected_columns[0].compare(BizPalletEvent.event_seq.property.columns[0]))
for dialect in (mysql.dialect(), postgresql.dialect()):
sql = compile_sql(statement, dialect)
self.assertNotIn('max(', sql.lower())
self.assertIn('biz_pallet_event.pallet_id = 7', sql)
self.assertIn('ORDER BY biz_pallet_event.event_seq DESC', sql)
self.assertIn('LIMIT 1', sql)
self.assertIn('FOR UPDATE', sql)
async def test_add_first_event_starts_sequence_at_one(self) -> None:
sequence_result = MagicMock()
sequence_result.scalar_one.return_value = None
sequence_result.scalar_one_or_none.return_value = None
sequence_result.scalar.return_value = None
self.db.execute.return_value = sequence_result
inventory = BizPalletInventory(pallet_id=8, last_event_id=None)
event = BizPalletEvent(pallet_id=8, idempotency_key='first-key', event_id=51)
@@ -373,6 +381,7 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
self.assertIs(record, actual)
statement = self.db.execute.await_args.args[0]
self.assertTrue(statement.get_final_froms()[0].compare(model.__table__))
self.assertIs(True, statement.get_execution_options().get('populate_existing'))
for dialect in (mysql.dialect(), postgresql.dialect()):
sql = compile_sql(statement, dialect)
self.assertIn(f'{model.__tablename__}.{id_field} = {record_id}', sql)
@@ -442,7 +451,7 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
'biz_inbound_order.warehouse_id = 3',
"biz_inbound_order.status = 'DRAFT'",
"biz_inbound_order.create_time >= '2026-07-01 00:00:00'",
"biz_inbound_order.create_time <= '2026-07-02 23:59:59'",
"biz_inbound_order.create_time < '2026-07-03 00:00:00'",
),
),
(
@@ -467,7 +476,7 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
"biz_filling_record.batch_no LIKE '%BATCH-A%'",
"biz_pallet.pallet_code LIKE '%P-00%'",
"biz_filling_record.operated_at >= '2026-07-01 00:00:00'",
"biz_filling_record.operated_at <= '2026-07-02 23:59:59'",
"biz_filling_record.operated_at < '2026-07-03 00:00:00'",
),
),
(
@@ -490,7 +499,7 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
'biz_shipment_order.target_site_id = 4',
"biz_shipment_order.status = 'CONFIRMED'",
"biz_shipment_order.create_time >= '2026-07-01 00:00:00'",
"biz_shipment_order.create_time <= '2026-07-02 23:59:59'",
"biz_shipment_order.create_time < '2026-07-03 00:00:00'",
),
),
(
@@ -513,7 +522,7 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
'biz_usage_record.line_id = 4',
"biz_pallet.pallet_code LIKE '%P-00%'",
"biz_usage_record.occurred_at >= '2026-07-01 00:00:00'",
"biz_usage_record.occurred_at <= '2026-07-02 23:59:59'",
"biz_usage_record.occurred_at < '2026-07-03 00:00:00'",
),
),
(
@@ -536,7 +545,7 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
'biz_return_order.target_warehouse_id = 4',
"biz_return_order.status = 'REQUESTED'",
"biz_return_order.requested_at >= '2026-07-01 00:00:00'",
"biz_return_order.requested_at <= '2026-07-02 23:59:59'",
"biz_return_order.requested_at < '2026-07-03 00:00:00'",
),
),
)
@@ -558,6 +567,20 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
for fragment in fragments:
self.assertIn(fragment, sql)
def test_time_conditions_include_fractional_last_second_with_exclusive_next_day(self) -> None:
lower, upper = OperationDao._time_conditions(
BizUsageRecord.occurred_at,
'2026-07-02',
'2026-07-02',
)
self.assertIs(operators.ge, lower.operator)
self.assertIs(operators.lt, upper.operator)
self.assertEqual(datetime(2026, 7, 2), lower.right.value)
self.assertEqual(datetime(2026, 7, 3), upper.right.value)
fractional_last_second = datetime(2026, 7, 2, 23, 59, 59, 500000)
self.assertTrue(lower.right.value <= fractional_last_second < upper.right.value)
async def test_order_details_load_items_explicitly_in_stable_order(self) -> None:
cases = (
(
@@ -675,11 +698,18 @@ class TestOperationDaoWrites(unittest.IsolatedAsyncioTestCase):
self.db.commit.assert_not_awaited()
self.db.rollback.assert_not_awaited()
async def test_update_methods_target_primary_key_and_flush_without_commit(self) -> None:
async def test_update_methods_target_primary_key_and_exclude_creation_audit(self) -> None:
malicious_create_time = datetime(2020, 1, 2, 3, 4, 5)
cases = (
(
OperationDao.update_inbound,
BizInboundOrder(inbound_id=11, warehouse_id=3, status='CONFIRMED'),
BizInboundOrder(
inbound_id=11,
warehouse_id=3,
status='CONFIRMED',
create_by='malicious',
create_time=malicious_create_time,
),
BizInboundOrder,
'inbound_id',
11,
@@ -687,7 +717,12 @@ class TestOperationDaoWrites(unittest.IsolatedAsyncioTestCase):
),
(
OperationDao.update_filling_record,
BizFillingRecord(filling_id=12, active_slot=None),
BizFillingRecord(
filling_id=12,
active_slot=None,
create_by='malicious',
create_time=malicious_create_time,
),
BizFillingRecord,
'filling_id',
12,
@@ -695,7 +730,12 @@ class TestOperationDaoWrites(unittest.IsolatedAsyncioTestCase):
),
(
OperationDao.update_shipment,
BizShipmentOrder(shipment_id=13, status='CONFIRMED'),
BizShipmentOrder(
shipment_id=13,
status='CONFIRMED',
create_by='malicious',
create_time=malicious_create_time,
),
BizShipmentOrder,
'shipment_id',
13,
@@ -703,7 +743,12 @@ class TestOperationDaoWrites(unittest.IsolatedAsyncioTestCase):
),
(
OperationDao.update_return_order,
BizReturnOrder(return_id=14, status='CONFIRMED'),
BizReturnOrder(
return_id=14,
status='CONFIRMED',
create_by='malicious',
create_time=malicious_create_time,
),
BizReturnOrder,
'return_id',
14,
@@ -722,27 +767,49 @@ class TestOperationDaoWrites(unittest.IsolatedAsyncioTestCase):
sql = compile_sql(statement)
self.assertIn(f'WHERE {model.__tablename__}.{id_field} = {record_id}', sql)
self.assertIn(value_fragment, sql)
self.assertNotIn('create_by', sql)
self.assertNotIn('create_time', sql)
self.db.flush.assert_awaited_once()
self.db.commit.assert_not_awaited()
self.db.rollback.assert_not_awaited()
async def test_delete_methods_target_one_order_without_commit(self) -> None:
async def test_delete_methods_remove_items_before_parent_without_commit(self) -> None:
cases = (
(OperationDao.delete_inbound, BizInboundOrder, 'inbound_id', 11),
(OperationDao.delete_shipment, BizShipmentOrder, 'shipment_id', 12),
(
OperationDao.delete_inbound,
BizInboundItem,
BizInboundOrder,
'inbound_id',
11,
),
(
OperationDao.delete_shipment,
BizShipmentItem,
BizShipmentOrder,
'shipment_id',
12,
),
)
for method, model, id_field, record_id in cases:
for method, item_model, order_model, id_field, record_id in cases:
with self.subTest(method=method.__name__):
self.db.reset_mock()
await method(self.db, record_id)
statement = self.db.execute.await_args.args[0]
self.assertTrue(statement.table.compare(model.__table__))
self.assertIn(
f'{model.__tablename__}.{id_field} = {record_id}',
compile_sql(statement),
)
self.assertEqual(2, len(self.db.execute.await_args_list))
item_statement = self.db.execute.await_args_list[0].args[0]
order_statement = self.db.execute.await_args_list[1].args[0]
self.assertTrue(item_statement.table.compare(item_model.__table__))
self.assertTrue(order_statement.table.compare(order_model.__table__))
for dialect in (mysql.dialect(), postgresql.dialect()):
self.assertIn(
f'{item_model.__tablename__}.{id_field} = {record_id}',
compile_sql(item_statement, dialect),
)
self.assertIn(
f'{order_model.__tablename__}.{id_field} = {record_id}',
compile_sql(order_statement, dialect),
)
self.db.flush.assert_awaited_once()
self.db.commit.assert_not_awaited()
self.db.rollback.assert_not_awaited()
shierfeng-fastapi-backend/tests/module_business/test_vo_contract.py
+39 -5
View File
@@ -569,6 +569,14 @@ WRITE_MODELS = (
ReturnConfirmModel,
)
OPERATION_PAGE_QUERY_MODELS = (
InboundPageQueryModel,
FillingPageQueryModel,
ShipmentPageQueryModel,
UsagePageQueryModel,
ReturnPageQueryModel,
)
PAGE_QUERY_MODELS = (
StrainPageQueryModel,
CustomerPageQueryModel,
@@ -576,11 +584,7 @@ PAGE_QUERY_MODELS = (
LinePageQueryModel,
WarehousePageQueryModel,
PalletPageQueryModel,
InboundPageQueryModel,
FillingPageQueryModel,
ShipmentPageQueryModel,
UsagePageQueryModel,
ReturnPageQueryModel,
*OPERATION_PAGE_QUERY_MODELS,
PalletInventoryReportQueryModel,
FillingReportQueryModel,
ShipmentReportQueryModel,
@@ -775,6 +779,36 @@ class BusinessVoContractTest(unittest.TestCase):
with self.assertRaises(ValidationError):
model_class.model_validate({'pageSize': 0})
def test_operation_page_query_dates_require_strict_calendar_format(self) -> None:
for model_class in OPERATION_PAGE_QUERY_MODELS:
with self.subTest(model=model_class.__name__, case='begin-only'):
model = model_class.model_validate({'beginTime': '2026-07-01'})
self.assertEqual('2026-07-01', model.begin_time)
self.assertIsNone(model.end_time)
with self.subTest(model=model_class.__name__, case='end-only'):
model = model_class.model_validate({'endTime': '2026-07-31'})
self.assertIsNone(model.begin_time)
self.assertEqual('2026-07-31', model.end_time)
for field_name in ('beginTime', 'endTime'):
for invalid_date in ('not-a-date', '2026-02-30', '2026-7-01', '2026-07-1'):
with (
self.subTest(
model=model_class.__name__,
field=field_name,
value=invalid_date,
),
self.assertRaises(ValidationError),
):
model_class.model_validate({field_name: invalid_date})
def test_operation_page_query_date_range_must_not_be_inverted(self) -> None:
for model_class in OPERATION_PAGE_QUERY_MODELS:
with self.subTest(model=model_class.__name__, case='equal'):
model = model_class.model_validate({'beginTime': '2026-07-02', 'endTime': '2026-07-02'})
self.assertEqual(model.begin_time, model.end_time)
with self.subTest(model=model_class.__name__, case='inverted'), self.assertRaises(ValidationError):
model_class.model_validate({'beginTime': '2026-07-03', 'endTime': '2026-07-02'})
def test_pallet_page_size_is_capped_at_100(self) -> None:
self.assertEqual(100, PalletPageQueryModel.model_validate({'pageSize': 100}).page_size)
with self.assertRaises(ValidationError):