1088c261
 tangwang
mv files
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
|
"""
数据库连接服务模块
提供统一的数据库连接接口
"""
from sqlalchemy import create_engine
from urllib.parse import quote_plus
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def create_db_connection(host, port, database, username, password):
"""
创建数据库连接
Args:
host: 数据库主机地址
port: 端口
database: 数据库名
username: 用户名
password: 密码
Returns:
SQLAlchemy engine对象
"""
try:
# 对密码进行URL编码，处理特殊字符
encoded_password = quote_plus(password)
# 构建连接字符串
connection_string = f'mysql+pymysql://{username}:{encoded_password}@{host}:{port}/{database}'
# 创建引擎
engine = create_engine(
connection_string,
pool_pre_ping=True, # 连接池预检
pool_recycle=3600, # 连接回收时间
echo=False
)
logger.info(f"Database connection created successfully: {host}:{port}/{database}")
return engine
except Exception as e:
logger.error(f"Failed to create database connection: {e}")
raise
|