42 lines
1.4 KiB
Python
42 lines
1.4 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
import asyncio
|
|
import sys
|
|
sys.path.insert(0, 'ruoyi-fastapi-backend')
|
|
|
|
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
|
|
from urllib.parse import quote_plus
|
|
|
|
DB_URL = f'mysql+asyncmy://cpy_admin:{quote_plus("Tgzz2025+")}@localhost:3307/ruoyi-fastapi'
|
|
|
|
async def test():
|
|
from sqlalchemy import text
|
|
|
|
engine = create_async_engine(DB_URL, echo=False)
|
|
AsyncSession = async_sessionmaker(engine, expire_on_commit=False)
|
|
|
|
async with AsyncSession() as db:
|
|
# 查询最新的 RJ 格式单号
|
|
result = await db.execute(text("""
|
|
SELECT receipt_id, receipt_no FROM warehouse_receipt
|
|
WHERE receipt_no LIKE '2026RJ%'
|
|
ORDER BY receipt_id DESC LIMIT 3
|
|
"""))
|
|
rows = result.fetchall()
|
|
print("RJ format receipts:")
|
|
for row in rows:
|
|
print(f" ID={row[0]}, No={row[1]}")
|
|
|
|
# 查询最新的 内检 格式单号
|
|
result2 = await db.execute(text("""
|
|
SELECT receipt_id, receipt_no FROM warehouse_receipt
|
|
WHERE receipt_no LIKE '2026内检%'
|
|
ORDER BY receipt_id DESC LIMIT 3
|
|
"""))
|
|
rows2 = result2.fetchall()
|
|
print("\n内检 format receipts:")
|
|
for row in rows2:
|
|
print(f" ID={row[0]}, No={row[1]}")
|
|
|
|
asyncio.run(test())
|