51 lines
1.6 KiB
Python
51 lines
1.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
检查 FastAPI 返回的时间格式
|
|
"""
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
import json
|
|
|
|
project_root = Path(__file__).parent
|
|
sys.path.insert(0, str(project_root))
|
|
sys.path.insert(0, str(project_root / "src"))
|
|
|
|
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
|
|
from sqlalchemy import text, select
|
|
from config.settings import settings
|
|
from models.database import SalesOrder, Customer
|
|
|
|
|
|
async def check():
|
|
"""检查 API 返回的时间格式"""
|
|
if not settings.DATABASE_URL:
|
|
print("错误：未配置数据库连接")
|
|
return
|
|
|
|
engine = create_async_engine(settings.DATABASE_URL, echo=False)
|
|
|
|
async with AsyncSession(engine) as session:
|
|
# 查询销售订单
|
|
result = await session.execute(
|
|
select(SalesOrder, Customer)
|
|
.join(Customer, SalesOrder.customer_id == Customer.id)
|
|
.order_by(SalesOrder.id.desc())
|
|
.limit(1)
|
|
)
|
|
row = result.first()
|
|
if row:
|
|
order, customer = row
|
|
print(f"订单号: {order.order_no}")
|
|
print(f"created_at (Python): {order.created_at}")
|
|
print(f"created_at (ISO格式): {order.created_at.isoformat() if order.created_at else 'N/A'}")
|
|
print(f"created_at (带时区): {order.created_at.isoformat() if order.created_at else 'N/A'}")
|
|
print(f"order_date (Python): {order.order_date}")
|
|
print(f"order_date (ISO格式): {order.order_date.isoformat() if order.order_date else 'N/A'}")
|
|
|
|
await engine.dispose()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(check())
|