Codex
69 lines
1.7 KiB
Python
69 lines
1.7 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
"""Handler 共享工具"""
|
|
import logging
|
|
from typing import Any, Dict
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def format_size(size_bytes: int) -> str:
|
|
"""格式化字节大小"""
|
|
if not size_bytes:
|
|
return "0 bytes"
|
|
if size_bytes >= 1073741824:
|
|
return f"{size_bytes / 1073741824:.2f} GB"
|
|
if size_bytes >= 1048576:
|
|
return f"{size_bytes / 1048576:.2f} MB"
|
|
if size_bytes >= 1024:
|
|
return f"{size_bytes / 1024:.2f} KB"
|
|
return f"{size_bytes} bytes"
|
|
|
|
|
|
def serialize_row(row: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""序列化行数据"""
|
|
for key, value in row.items():
|
|
if hasattr(value, "isoformat"):
|
|
row[key] = value.isoformat()
|
|
elif isinstance(value, bytes):
|
|
row[key] = value.decode("utf-8", errors="replace")
|
|
elif isinstance(value, (set, frozenset)):
|
|
row[key] = list(value)
|
|
return row
|
|
|
|
|
|
def format_connection_error(exc: Exception) -> str:
|
|
"""将驱动异常转为可读的错误详情"""
|
|
if exc is None:
|
|
return "未知错误"
|
|
msg = str(exc).strip()
|
|
if msg:
|
|
return msg
|
|
return exc.__class__.__name__
|
|
|
|
|
|
def log_database_connect_failure(
|
|
*,
|
|
db_type: str,
|
|
host: str,
|
|
port: int,
|
|
user: str = "",
|
|
database: str = "",
|
|
db_name: str = "",
|
|
detail: str,
|
|
action: str = "connect",
|
|
) -> None:
|
|
"""记录数据库连接失败详情到后台日志"""
|
|
logger.error(
|
|
"Database connection failed [%s]: db_name=%s db_type=%s target=%s:%s "
|
|
"database=%s user=%s error=%s",
|
|
action,
|
|
db_name or "-",
|
|
db_type,
|
|
host,
|
|
port,
|
|
database or "-",
|
|
user or "-",
|
|
detail,
|
|
)
|