45 lines
2.0 KiB
Python
45 lines
2.0 KiB
Python
import mysql.connector
|
||
import json
|
||
from datetime import date, datetime, timedelta
|
||
from decimal import Decimal
|
||
from create_logger import logger
|
||
from utils.format import DateEncoder, default_encoder
|
||
from conf import settings
|
||
|
||
|
||
class SqlService:
|
||
def __init__(self, service_name="数据库查询"):
|
||
# 连接数据库
|
||
self.conn = mysql.connector.connect(
|
||
host=settings.mysql_host,
|
||
port=settings.mysql_port,
|
||
user=settings.mysql_user,
|
||
password=settings.mysql_password,
|
||
database=settings.mysql_database
|
||
)
|
||
self.service_name = service_name
|
||
|
||
# 定义执行SQL查询方法，输入SQL字符串，返回JSON字符串
|
||
def execute_query(self, sql: str, no_data_message="未找数据") -> str:
|
||
try:
|
||
cursor = self.conn.cursor(dictionary=True)
|
||
cursor.execute(sql)
|
||
results = cursor.fetchall()
|
||
cursor.close()
|
||
# 格式化结果
|
||
for result in results: # 遍历每个结果字典
|
||
for key, value in result.items():
|
||
if isinstance(value, (date, datetime, timedelta, Decimal)): # 检查值是否为特殊类型
|
||
result[key] = default_encoder(value) # 使用自定义编码器格式化该值
|
||
# 序列化为JSON，如果有结果返回success，否则no_data；使用DateEncoder，非ASCII不转义
|
||
return json.dumps({"status": "success", "data": results} if results else {"status": "no_data", "message": no_data_message}, cls=DateEncoder, ensure_ascii=False)
|
||
except Exception as e:
|
||
logger.error(f"{self.service_name}错误: {str(e)}")
|
||
# 返回错误JSON响应
|
||
return json.dumps({"status": "error", "message": str(e)}, ensure_ascii=False)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
service = SqlService()
|
||
sql = "SELECT * FROM flight_tickets WHERE departure_city = '上海' AND arrival_city = '北京' AND DATE(departure_time) = '2025-10-28' AND cabin_type = '公务舱'"
|
||
print(service.execute_query(sql)) |