85 lines
3.5 KiB
Python
85 lines
3.5 KiB
Python
|
#!/usr/bin/env python
|
||
|
|
# -*- coding: utf-8 -*-
|
||
|
|
import asyncio
|
||
|
|
import sys
|
||
|
|
import traceback
|
||
|
|
sys.path.insert(0, 'ruoyi-fastapi-backend')
|
||
|
|
|
||
|
|
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
|
||
|
|
from sqlalchemy import select
|
||
|
|
from urllib.parse import quote_plus
|
||
|
|
|
||
|
|
# 数据库配置
|
||
|
|
DB_URL = f'mysql+asyncmy://cpy_admin:{quote_plus("Tgzz2025+")}@localhost:3307/ruoyi-fastapi'
|
||
|
|
|
||
|
|
async def test_receipt():
|
||
|
|
engine = create_async_engine(DB_URL, echo=False)
|
||
|
|
AsyncSession = async_sessionmaker(engine, expire_on_commit=False)
|
||
|
|
|
||
|
|
async with AsyncSession() as db:
|
||
|
|
try:
|
||
|
|
# 导入模型
|
||
|
|
from module_admin.entity.do.warehouse_receipt_do import WarehouseReceipt
|
||
|
|
from module_admin.entity.do.warehouse_sample_do import WarehouseSample
|
||
|
|
from module_admin.dao.warehouse_sample_dao import WarehouseSampleDao
|
||
|
|
from module_admin.entity.vo.warehouse_sample_vo import WarehouseSamplePageQueryModel
|
||
|
|
from utils.common_util import CamelCaseUtil
|
||
|
|
from utils.response_util import ResponseUtil
|
||
|
|
|
||
|
|
# 查询入库单
|
||
|
|
result = await db.execute(
|
||
|
|
select(WarehouseReceipt).where(WarehouseReceipt.receipt_id == 685)
|
||
|
|
)
|
||
|
|
receipt = result.scalars().first()
|
||
|
|
|
||
|
|
if receipt:
|
||
|
|
print(f"Receipt found: {receipt.receipt_id}, {receipt.receipt_no}")
|
||
|
|
|
||
|
|
receipt_dict = CamelCaseUtil.transform_result(receipt)
|
||
|
|
|
||
|
|
# 使用 WarehouseSampleDao 查询样品 - 这是 service 里的方式
|
||
|
|
sample_query = WarehouseSamplePageQueryModel(receiptId=receipt.receipt_id, pageNum=1, pageSize=1000)
|
||
|
|
print(f"sample_query.receipt_id={sample_query.receipt_id}")
|
||
|
|
|
||
|
|
samples = await WarehouseSampleDao.get_sample_list(db, sample_query, is_page=False)
|
||
|
|
print(f"Found {len(samples)} samples via DAO")
|
||
|
|
|
||
|
|
# 转换样品
|
||
|
|
samples_list = [CamelCaseUtil.transform_result(s) for s in samples]
|
||
|
|
|
||
|
|
# 序列化日期
|
||
|
|
def serialize_dates(obj):
|
||
|
|
if isinstance(obj, dict):
|
||
|
|
for key, value in obj.items():
|
||
|
|
if hasattr(value, 'isoformat'):
|
||
|
|
obj[key] = value.isoformat()
|
||
|
|
elif isinstance(value, list):
|
||
|
|
for item in value:
|
||
|
|
serialize_dates(item)
|
||
|
|
return obj
|
||
|
|
|
||
|
|
receipt_dict = serialize_dates(receipt_dict)
|
||
|
|
samples_list = [serialize_dates(s) for s in samples_list]
|
||
|
|
receipt_dict['samples'] = samples_list
|
||
|
|
|
||
|
|
print(f"Final receipt_dict ready")
|
||
|
|
|
||
|
|
# 测试 ResponseUtil.success
|
||
|
|
try:
|
||
|
|
response = ResponseUtil.success(data=receipt_dict)
|
||
|
|
print(f"ResponseUtil.success success!")
|
||
|
|
print(f"Response body: {response.body[:500]}...")
|
||
|
|
except Exception as e:
|
||
|
|
print(f"ResponseUtil.success error: {e}")
|
||
|
|
traceback.print_exc()
|
||
|
|
|
||
|
|
else:
|
||
|
|
print("Receipt not found")
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print(f"Error: {e}")
|
||
|
|
traceback.print_exc()
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
asyncio.run(test_receipt())
|