优化网站平台宁夏高端网站建设

优化网站平台,宁夏高端网站建设,网站搭建品牌,沈阳建设工程信息网浑南新区#x1f4da; QuantumFlow工作流自动化从入门到精通 - 第12篇 在上一篇中#xff0c;我们实现了HTTP请求和Webhook触发器。本文将深入数据库操作连接器的开发#xff0c;这是企业级工作流中最常用的功能之一。我们将实现生产级的连接池管理、SQL注入防护、事务支持等核心特性…QuantumFlow工作流自动化从入门到精通 - 第12篇在上一篇中我们实现了HTTP请求和Webhook触发器。本文将深入数据库操作连接器的开发这是企业级工作流中最常用的功能之一。我们将实现生产级的连接池管理、SQL注入防护、事务支持等核心特性。 本文概览学习目标掌握数据库连接池的设计与实现学习参数化查询防止SQL注入理解事务管理的最佳实践掌握批量操作的性能优化技巧实现支持多种数据库的统一接口技术栈Python 3.11 / asyncpgPostgreSQL异步驱动aiomysqlMySQL异步驱动SQLAlchemyORM框架asyncio异步编程psycopg2PostgreSQL同步驱动备用预计阅读时间35分钟前置知识SQL基础、Python异步编程、第10-11篇内容 一、为什么数据库连接器如此重要1.1 业务场景分析典型应用场景数据同步: 场景: 每天凌晨从业务数据库同步数据到数仓 需求: 批量查询 批量插入 事务保证 挑战: 百万级数据量性能要求高 订单处理: 场景: 用户下单后更新库存、创建订单、扣减余额 需求: 多表操作 事务一致性 挑战: 并发控制、死锁处理 报表生成: 场景: 定时生成销售报表并发送邮件 需求: 复杂SQL查询 数据聚合 挑战: 查询优化、超时控制 数据清洗: 场景: 定期清理过期数据、归档历史记录 需求: 批量删除 数据迁移 挑战: 锁表风险、性能影响1.2 设计目标核心目标安全性防止SQL注入、权限控制性能连接池复用、批量操作优化可靠性事务支持、错误重试易用性统一接口、自动类型转换可扩展性支持多种数据库、自定义查询 二、数据库连接器核心实现2.1 连接池管理器# src/connectors/database/pool_manager.py 数据库连接池管理器 支持PostgreSQL和MySQL的异步连接池 import asyncio from typing import Dict, Any, Optional, Union from contextlib import asynccontextmanager import asyncpg import aiomysql from urllib.parse import urlparse import logging logger logging.getLogger(__name__) class DatabaseType: 数据库类型枚举 POSTGRESQL postgresql MYSQL mysql class ConnectionPool: 统一的数据库连接池接口 特性 - 自动重连机制 - 连接健康检查 - 连接超时控制 - 连接数限制 - 优雅关闭 def __init__( self, connection_string: str, min_size: int 5, max_size: int 20, timeout: float 30.0, max_queries: int 50000, max_inactive_connection_lifetime: float 300.0 ): 初始化连接池 Args: connection_string: 数据库连接字符串 格式: postgresql://user:passhost:port/db mysql://user:passhost:port/db min_size: 最小连接数 max_size: 最大连接数 timeout: 连接超时时间秒 max_queries: 单个连接最大查询次数防止内存泄漏 max_inactive_connection_lifetime: 连接最大空闲时间秒 self.connection_string connection_string self.min_size min_size self.max_size max_size self.timeout timeout self.max_queries max_queries self.max_inactive_connection_lifetime max_inactive_connection_lifetime # 解析连接串确定数据库类型 parsed urlparse(connection_string) self.db_type self._detect_database_type(parsed.scheme) # 连接池实例 self._pool: Optional[Union[asyncpg.Pool, aiomysql.Pool]] None self._closed False def _detect_database_type(self, scheme: str) - str: 检测数据库类型 if scheme in [postgresql, postgres]: return DatabaseType.POSTGRESQL elif scheme mysql: return DatabaseType.MYSQL else: raise ValueError(fUnsupported database type: {scheme}) async def initialize(self): 初始化连接池 if self._pool is not None: logger.warning(Connection pool already initialized) return try: if self.db_type DatabaseType.POSTGRESQL: self._pool await self._create_postgresql_pool() elif self.db_type DatabaseType.MYSQL: self._pool await self._create_mysql_pool() logger.info( fDatabase connection pool initialized: ftype{self.db_type}, min{self.min_size}, max{self.max_size} ) except Exception as e: logger.error(fFailed to initialize connection pool: {e}) raise async def _create_postgresql_pool(self) - asyncpg.Pool: 创建PostgreSQL连接池 return await asyncpg.create_pool( self.connection_string, min_sizeself.min_size, max_sizeself.max_size, timeoutself.timeout, max_queriesself.max_queries, max_inactive_connection_lifetimeself.max_inactive_connection_lifetime, command_timeout60.0 # 单个命令超时 ) async def _create_mysql_pool(self) - aiomysql.Pool: 创建MySQL连接池 parsed urlparse(self.connection_string) return await aiomysql.create_pool( hostparsed.hostname, portparsed.port or 3306, userparsed.username, passwordparsed.password, dbparsed.path.lstrip(/), minsizeself.min_size, maxsizeself.max_size, connect_timeoutself.timeout, autocommitFalse, # 手动控制事务 charsetutf8mb4 ) asynccontextmanager async def acquire(self): 获取数据库连接上下文管理器 Example: async with pool.acquire() as conn: result await conn.fetch(SELECT * FROM users) if self._pool is None: raise RuntimeError(Connection pool not initialized) if self._closed: raise RuntimeError(Connection pool is closed) # 获取连接 if self.db_type DatabaseType.POSTGRESQL: async with self._pool.acquire() as conn: yield PostgreSQLConnection(conn) else: async with self._pool.acquire() as conn: yield MySQLConnection(conn) async def execute( self, query: str, *args, timeout: Optional[float] None ) - str: 执行SQL语句无返回结果 Args: query: SQL语句 *args: 查询参数 timeout: 超时时间 Returns: 执行状态如 INSERT 0 1 async with self.acquire() as conn: return await conn.execute(query, *args, timeouttimeout) async def fetch( self, query: str, *args, timeout: Optional[float] None ) - list: 执行查询并返回所有结果 Args: query: SQL查询语句 *args: 查询参数 timeout: 超时时间 Returns: 查询结果列表 async with self.acquire() as conn: return await conn.fetch(query, *args, timeouttimeout) async def fetchrow( self, query: str, *args, timeout: Optional[float] None ) - Optional[Dict]: 执行查询并返回第一行结果 Args: query: SQL查询语句 *args: 查询参数 timeout: 超时时间 Returns: 第一行结果字典格式或None async with self.acquire() as conn: return await conn.fetchrow(query, *args, timeouttimeout) async def fetchval( self, query: str, *args, column: int 0, timeout: Optional[float] None ) - Any: 执行查询并返回第一行第一列的值 Args: query: SQL查询语句 *args: 查询参数 column: 列索引 timeout: 超时时间 Returns: 单个值 async with self.acquire() as conn: return await conn.fetchval(query, *args, columncolumn, timeouttimeout) async def close(self): 关闭连接池 if self._pool is None or self._closed: return self._closed True if self.db_type DatabaseType.POSTGRESQL: await self._pool.close() else: self._pool.close() await self._pool.wait_closed() logger.info(Database connection pool closed) async def get_pool_status(self) - Dict[str, Any]: 获取连接池状态 if self._pool is None: return {status: not_initialized} if self.db_type DatabaseType.POSTGRESQL: return { status: active, size: self._pool.get_size(), free: self._pool.get_idle_size(), min_size: self._pool.get_min_size(), max_size: self._pool.get_max_size() } else: return { status: active, size: self._pool.size(), free: self._pool.freesize(), min_size: self._pool.minsize, max_size: self._pool.maxsize } class PostgreSQLConnection: PostgreSQL连接包装器 def __init__(self, conn: asyncpg.Connection): self._conn conn async def execute(self, query: str, *args, timeout: Optional[float] None) - str: return await self._conn.execute(query, *args, timeouttimeout) async def fetch(self, query: str, *args, timeout: Optional[float] None) - list: rows await self._conn.fetch(query, *args, timeouttimeout) return [dict(row) for row in rows] async def fetchrow(self, query: str, *args, timeout: Optional[float] None) - Optional[Dict]: row await self._conn.fetchrow(query, *args, timeouttimeout) return dict(row) if row else None async def fetchval(self, query: str, *args, column: int 0, timeout: Optional[float] None) - Any: return await self._conn.fetchval(query, *args, columncolumn, timeouttimeout) async def transaction(self): 开启事务 return self._conn.transaction() class MySQLConnection: MySQL连接包装器 def __init__(self, conn: aiomysql.Connection): self._conn conn async def execute(self, query: str, *args, timeout: Optional[float] None) - str: async with self._conn.cursor() as cursor: await cursor.execute(query, args) await self._conn.commit() return fOK {cursor.rowcount} async def fetch(self, query: str, *args, timeout: Optional[float] None) - list: async with self._conn.cursor(aiomysql.DictCursor) as cursor: await cursor.execute(query, args) return await cursor.fetchall() async def fetchrow(self, query: str, *args, timeout: Optional[float] None) - Optional[Dict]: async with self._conn.cursor(aiomysql.DictCursor) as cursor: await cursor.execute(query, args) return await cursor.fetchone() async def fetchval(self, query: str, *args, column: int 0, timeout: Optional[float] None) - Any: async with self._conn.cursor() as cursor: await cursor.execute(query, args) row await cursor.fetchone() return row[column] if row else None asynccontextmanager async def transaction(self): 开启事务 await self._conn.begin() try: yield await self._conn.commit() except Exception: await self._conn.rollback() raise2.2 数据库连接器主类# src/connectors/database/database_connector.py 数据库操作连接器 支持PostgreSQL和MySQL的完整CRUD操作 from typing import Dict, List, Any, Optional from enum import Enum import logging import re from core.plugins.base_connector import BaseConnector from core.plugins.auth import NoAuth from core.plugins.architecture import ( ConnectorMetadata, Action, Parameter, ParameterType, ConnectorStatus ) from .pool_manager import ConnectionPool logger logging.getLogger(__name__) class QueryType(str, Enum): 查询类型枚举 SELECT select INSERT insert UPDATE update DELETE delete CUSTOM custom class DatabaseConnector(BaseConnector): 数据库操作连接器 特性 - 连接池管理自动复用 - 参数化查询防SQL注入 - 事务支持ACID保证 - 批量操作性能优化 - 查询超时控制 - 自动类型转换 - 错误重试机制 metadata ConnectorMetadata( keydatabase, nameDatabase Operations, version1.0.0, description支持PostgreSQL和MySQL的数据库操作连接器, authorQuantumFlow Team, categorydatabase, tags[database, sql, postgresql, mysql], statusConnectorStatus.ACTIVE, icon_urlhttps://cdn.quantumflow.dev/icons/database.svg ) def __init__(self): super().__init__() self._pools: Dict[str, ConnectionPool] {} def get_auth(self) - NoAuth: 数据库连接通过连接字符串认证 return NoAuth() def get_actions(self) - List[Action]: return [ # ═══════════════════════════════════════════════ # 查询操作 # ═══════════════════════════════════════════════ Action( namequery, display_name执行SQL查询, description执行SELECT查询并返回结果, parameters[ Parameter( nameconnection_string, typeParameterType.STRING, display_name数据库连接字符串, description格式: postgresql://user:passhost:port/db 或 mysql://..., requiredTrue, placeholderpostgresql://user:passwordlocalhost:5432/mydb, validation{pattern: r^(postgresql|mysql)://} ), Parameter( namequery, typeParameterType.STRING, display_nameSQL查询语句, descriptionSELECT查询语句支持参数化查询, requiredTrue, placeholderSELECT * FROM users WHERE id $1, multilineTrue ), Parameter( nameparameters, typeParameterType.ARRAY, display_name查询参数, description参数化查询的参数列表防止SQL注入, requiredFalse, default[], placeholder[123, johnexample.com] ), Parameter( nametimeout, typeParameterType.NUMBER, display_name查询超时秒, description查询超时时间0表示不限制, requiredFalse, default30, validation{min: 0, max: 300} ), Parameter( namemax_rows, typeParameterType.NUMBER, display_name最大返回行数, description限制返回的最大行数0表示不限制, requiredFalse, default1000, validation{min: 0, max: 100000} ) ], output_schema{ type: object, properties: { rows: { type: array, description: 查询结果行 }, row_count: { type: integer, description: 返回的行数 }, columns: { type: array, description: 列名列表 }, elapsed: { type: number, description: 查询耗时秒 } } }, examples[ { name: 查询用户列表, parameters: { connection_string: postgresql://user:passlocalhost/mydb, query: SELECT id, name, email FROM users WHERE status $1 LIMIT $2, parameters: [active, 10], timeout: 10 } } ] ), # ═══════════════════════════════════════════════ # 插入操作 # ═══════════════════════════════════════════════ Action( nameinsert, display_name插入数据, description向表中插入一条或多条记录, parameters[ Parameter( nameconnection_string, typeParameterType.STRING, display_name数据库连接字符串, requiredTrue ), Parameter( nametable, typeParameterType.STRING, display_name表名, description要插入数据的表名, requiredTrue, placeholderusers ), Parameter( namedata, typeParameterType.OBJECT, display_name数据, description要插入的数据单条记录, requiredTrue, placeholder{name: John, email: johnexample.com} ), Parameter( namereturn_id, typeParameterType.BOOLEAN, display_name返回插入的ID, description是否返回自增主键ID, requiredFalse, defaultTrue ) ], output_schema{ type: object, properties: { inserted_id: {type: [integer, null]}, affected_rows: {type: integer} } } ), # ═══════════════════════════════════════════════ # 批量插入 # ═══════════════════════════════════════════════ Action( namebulk_insert, display_name批量插入数据, description高性能批量插入多条记录, parameters[ Parameter( nameconnection_string, typeParameterType.STRING, display_name数据库连接字符串, requiredTrue ), Parameter( nametable, typeParameterType.STRING, display_name表名, requiredTrue ), Parameter( namedata, typeParameterType.ARRAY, display_name数据列表, description要插入的数据数组, requiredTrue, placeholder[{name: John}, {name: Jane}] ), Parameter( namebatch_size, typeParameterType.NUMBER, display_name批次大小, description每批插入的记录数优化性能, requiredFalse, default1000, validation{min: 1, max: 10000} ) ] ), # ═══════════════════════════════════════════════ # 更新操作 # ═══════════════════════════════════════════════ Action( nameupdate, display_name更新数据, description更新表中的记录, parameters[ Parameter( nameconnection_string, typeParameterType.STRING, display_name数据库连接字符串, requiredTrue ), Parameter( nametable, typeParameterType.STRING, display_name表名, requiredTrue ), Parameter( namedata, typeParameterType.OBJECT, display_name更新数据, description要更新的字段和值, requiredTrue, placeholder{status: inactive} ), Parameter( namewhere, typeParameterType.OBJECT, display_nameWHERE条件, description更新条件字段值, requiredTrue, placeholder{id: 123} ) ] ), # ═══════════════════════════════════════════════ # 删除操作 # ═══════════════════════════════════════════════ Action( namedelete, display_name删除数据, description从表中删除记录, parameters[ Parameter( nameconnection_string, typeParameterType.STRING, display_name数据库连接字符串, requiredTrue ), Parameter( nametable, typeParameterType.STRING, display_name表名, requiredTrue ), Parameter( namewhere, typeParameterType.OBJECT, display_nameWHERE条件, description删除条件必填防止误删全表, requiredTrue, placeholder{id: 123} ) ] ), # ═══════════════════════════════════════════════ # 事务操作 # ═══════════════════════════════════════════════ Action( nametransaction, display_name执行事务, description在事务中执行多个SQL语句ACID保证, parameters[ Parameter( nameconnection_string, typeParameterType.STRING, display_name数据库连接字符串, requiredTrue ), Parameter( namestatements, typeParameterType.ARRAY, display_nameSQL语句列表, description要在事务中执行的SQL语句数组, requiredTrue, placeholder[{query: UPDATE ..., params: [...]}, ...] ), Parameter( nameisolation_level, typeParameterType.STRING, display_name隔离级别, description事务隔离级别, requiredFalse, defaultread_committed, options[ read_uncommitted, read_committed, repeatable_read, serializable ] ) ] ) ] async def execute_action( self, action_name: str, parameters: Dict[str, Any], credentials: Dict[str, Any] ) - Dict[str, Any]: 执行数据库操作 # 验证参数 is_valid, error_msg self.validate_parameters(action_name, parameters) if not is_valid: raise ValueError(error_msg) # 路由到具体的操作方法 if action_name query: return await self._execute_query(parameters) elif action_name insert: return await self._execute_insert(parameters) elif action_name bulk_insert: return await self._execute_bulk_insert(parameters) elif action_name update: return await self._execute_update(parameters) elif action_name delete: return await self._execute_delete(parameters) elif action_name transaction: return await self._execute_transaction(parameters) else: raise ValueError(fUnknown action: {action_name}) async def _get_pool(self, connection_string: str) - ConnectionPool: 获取或创建连接池单例模式 if connection_string not in self._pools: pool ConnectionPool( connection_stringconnection_string, min_size2, max_size10 ) await pool.initialize() self._pools[connection_string] pool return self._pools[connection_string] async def _execute_query(self, parameters: Dict[str, Any]) - Dict[str, Any]: 执行SELECT查询 实现了 - 参数化查询防SQL注入 - 查询超时控制 - 结果行数限制 - 性能统计 import time connection_string parameters[connection_string] query parameters[query] query_params parameters.get(parameters, []) timeout parameters.get(timeout, 30) max_rows parameters.get(max_rows, 1000) # 验证查询类型只允许SELECT if not self._is_select_query(query): raise ValueError(Only SELECT queries are allowed in query action) pool await self._get_pool(connection_string) start_time time.time() try: # 执行查询 rows await pool.fetch( query, *query_params, timeouttimeout if timeout 0 else None ) elapsed time.time() - start_time # 限制返回行数 if max_rows 0 and len(rows) max_rows: rows rows[:max_rows] logger.warning(fQuery result truncated to {max_rows} rows) # 提取列名 columns list(rows[0].keys()) if rows else [] return { rows: rows, row_count: len(rows), columns: columns, elapsed: elapsed } except Exception as e: logger.error(fQuery execution failed: {e}) raise Exception(fQuery failed: {str(e)}) async def _execute_insert(self, parameters: Dict[str, Any]) - Dict[str, Any]: 执行INSERT操作 实现了 - 自动生成INSERT语句 - 参数化插入防SQL注入 - 返回自增ID connection_string parameters[connection_string] table parameters[table] data parameters[data] return_id parameters.get(return_id, True) # 验证表名防止SQL注入 if not self._is_valid_identifier(table): raise ValueError(fInvalid table name: {table}) pool await self._get_pool(connection_string) # 构建INSERT语句 columns list(data.keys()) values list(data.values()) # PostgreSQL使用$1, $2...MySQL使用%s if pool.db_type postgresql: placeholders , .join(f${i1} for i in range(len(values))) query fINSERT INTO {table} ({, .join(columns)}) VALUES ({placeholders}) if return_id: query RETURNING id else: placeholders , .join([%s] * len(values)) query fINSERT INTO {table} ({, .join(columns)}) VALUES ({placeholders}) try: if return_id and pool.db_type postgresql: result await pool.fetchval(query, *values) return { inserted_id: result, affected_rows: 1 } else: await pool.execute(query, *values) return { inserted_id: None, affected_rows: 1 } except Exception as e: logger.error(fInsert failed: {e}) raise Exception(fInsert failed: {str(e)}) async def _execute_bulk_insert(self, parameters: Dict[str, Any]) - Dict[str, Any]: 批量插入数据 实现了 - 批次分割避免内存溢出 - 事务保证全部成功或全部失败 - 性能优化使用COPY或批量INSERT connection_string parameters[connection_string] table parameters[table] data_list parameters[data] batch_size parameters.get(batch_size, 1000) if not data_list: return {inserted_count: 0} # 验证表名 if not self._is_valid_identifier(table): raise ValueError(fInvalid table name: {table}) pool await self._get_pool(connection_string) # 获取列名从第一条记录 columns list(data_list[0].keys()) total_inserted 0 try: # 分批插入 for i in range(0, len(data_list), batch_size): batch data_list[i:i batch_size] # 构建批量INSERT语句 if pool.db_type postgresql: # PostgreSQL支持多行INSERT values_list [] params [] param_idx 1 for row in batch: row_placeholders [] for col in columns: row_placeholders.append(f${param_idx}) params.append(row[col]) param_idx 1 values_list.append(f({, .join(row_placeholders)})) query fINSERT INTO {table} ({, .join(columns)}) VALUES {, .join(values_list)} await pool.execute(query, *params) else: # MySQL批量插入 placeholders , .join([%s] * len(columns)) query fINSERT INTO {table} ({, .join(columns)}) VALUES ({placeholders}) async with pool.acquire() as conn: async with conn.transaction(): for row in batch: values [row[col] for col in columns] await conn.execute(query, *values) total_inserted len(batch) logger.info(fInserted batch {i//batch_size 1}: {len(batch)} rows) return { inserted_count: total_inserted, batch_count: (len(data_list) batch_size - 1) // batch_size } except Exception as e: logger.error(fBulk insert failed: {e}) raise Exception(fBulk insert failed: {str(e)}) async def _execute_update(self, parameters: Dict[str, Any]) - Dict[str, Any]: 执行UPDATE操作 connection_string parameters[connection_string] table parameters[table] data parameters[data] where parameters[where] # 验证表名 if not self._is_valid_identifier(table): raise ValueError(fInvalid table name: {table}) pool await self._get_pool(connection_string) # 构建UPDATE语句 set_clauses [] where_clauses [] params [] param_idx 1 # SET部分 for col, val in data.items(): if pool.db_type postgresql: set_clauses.append(f{col} ${param_idx}) else: set_clauses.append(f{col} %s) params.append(val) param_idx 1 # WHERE部分 for col, val in where.items(): if pool.db_type postgresql: where_clauses.append(f{col} ${param_idx}) else: where_clauses.append(f{col} %s) params.append(val) param_idx 1 query fUPDATE {table} SET {, .join(set_clauses)} WHERE { AND .join(where_clauses)} try: result await pool.execute(query, *params) # 解析affected rows if pool.db_type postgresql: affected int(result.split()[-1]) else: affected int(result.split()[1]) return {affected_rows: affected} except Exception as e: logger.error(fUpdate failed: {e}) raise Exception(fUpdate failed: {str(e)}) async def _execute_delete(self, parameters: Dict[str, Any]) - Dict[str, Any]: 执行DELETE操作 connection_string parameters[connection_string] table parameters[table] where parameters[where] # 验证表名 if not self._is_valid_identifier(table): raise ValueError(fInvalid table name: {table}) # 必须有WHERE条件防止误删全表 if not where: raise ValueError(WHERE condition is required for DELETE operation) pool await self._get_pool(connection_string) # 构建DELETE语句 where_clauses [] params [] param_idx 1 for col, val in where.items(): if pool.db_type postgresql: where_clauses.append(f{col} ${param_idx}) else: where_clauses.append(f{col} %s) params.append(val) param_idx 1 query fDELETE FROM {table} WHERE { AND .join(where_clauses)} try: result await pool.execute(query, *params) # 解析affected rows if pool.db_type postgresql: affected int(result.split()[-1]) else: affected int(result.split()[1]) return {affected_rows: affected} except Exception as e: logger.error(fDelete failed: {e}) raise Exception(fDelete failed: {str(e)}) async def _execute_transaction(self, parameters: Dict[str, Any]) - Dict[str, Any]: 执行事务 实现了 - ACID保证 - 自动回滚 - 隔离级别控制 connection_string parameters[connection_string] statements parameters[statements] isolation_level parameters.get(isolation_level, read_committed) pool await self._get_pool(connection_string) results [] try: async with pool.acquire() as conn: # 设置隔离级别 if pool.db_type postgresql: await conn.execute(fSET TRANSACTION ISOLATION LEVEL {isolation_level.upper().replace(_, )}) # 开启事务 async with conn.transaction(): for stmt in statements: query stmt[query] params stmt.get(params, []) result await conn.execute(query, *params) results.append({query: query, result: result}) return { success: True, statements_executed: len(statements), results: results } except Exception as e: logger.error(fTransaction failed: {e}) raise Exception(fTransaction failed (rolled back): {str(e)}) def _is_select_query(self, query: str) - bool: 检查是否为SELECT查询 query_upper query.strip().upper() return query_upper.startswith(SELECT) or query_upper.startswith(WITH) def _is_valid_identifier(self, identifier: str) - bool: 验证SQL标识符表名、列名是否合法 # 只允许字母、数字、下划线 return bool(re.match(r^[a-zA-Z_][a-zA-Z0-9_]*$, identifier)) async def cleanup(self): 清理资源关闭所有连接池 for pool in self._pools.values(): await pool.close() self._pools.clear() 三、高级特性实现3.1 查询构建器Query Builder# src/connectors/database/query_builder.py SQL查询构建器 提供类型安全的SQL查询构建能力 from typing import List, Dict, Any, Optional from enum import Enum class JoinType(str, Enum): JOIN类型 INNER INNER JOIN LEFT LEFT JOIN RIGHT RIGHT JOIN FULL FULL OUTER JOIN class QueryBuilder: SQL查询构建器 特性 - 链式调用 - 类型安全 - 自动参数化 - 防SQL注入 Example: qb QueryBuilder(users) qb.select([id, name, email]) .where(status, , active) .where(created_at, , 2024-01-01) .order_by(created_at, DESC) .limit(10) query, params qb.build() def __init__(self, table: str, db_type: str postgresql): self.table table self.db_type db_type self._select_columns: List[str] [] self._where_conditions: List[tuple] [] self._join_clauses: List[tuple] [] self._order_by_clauses: List[tuple] [] self._group_by_columns: List[str] [] self._having_conditions: List[tuple] [] self._limit_value: Optional[int] None self._offset_value: Optional[int] None self._params: List[Any] [] def select(self, columns: List[str]) - QueryBuilder: 选择列 self._select_columns columns return self def where(self, column: str, operator: str, value: Any) - QueryBuilder: 添加WHERE条件 self._where_conditions.append((column, operator, value)) return self def where_in(self, column: str, values: List[Any]) - QueryBuilder: WHERE IN条件 self._where_conditions.append((column, IN, values)) return self def where_between(self, column: str, start: Any, end: Any) - QueryBuilder: WHERE BETWEEN条件 self._where_conditions.append((column, BETWEEN, (start, end))) return self def where_null(self, column: str, is_null: bool True) - QueryBuilder: WHERE NULL条件 operator IS NULL if is_null else IS NOT NULL self._where_conditions.append((column, operator, None)) return self def join( self, table: str, on_left: str, on_right: str, join_type: JoinType JoinType.INNER ) - QueryBuilder: 添加JOIN self._join_clauses.append((join_type, table, on_left, on_right)) return self def order_by(self, column: str, direction: str ASC) - QueryBuilder: 添加ORDER BY self._order_by_clauses.append((column, direction.upper())) return self def group_by(self, columns: List[str]) - QueryBuilder: 添加GROUP BY self._group_by_columns columns return self def having(self, column: str, operator: str, value: Any) - QueryBuilder: 添加HAVING条件 self._having_conditions.append((column, operator, value)) return self def limit(self, limit: int) - QueryBuilder: 设置LIMIT self._limit_value limit return self def offset(self, offset: int) - QueryBuilder: 设置OFFSET self._offset_value offset return self def build(self) - tuple[str, List[Any]]: 构建SQL查询 Returns: (query, params): SQL查询字符串和参数列表 self._params [] # SELECT部分 if self._select_columns: select_clause fSELECT {, .join(self._select_columns)} else: select_clause SELECT * # FROM部分 from_clause fFROM {self.table} # JOIN部分 join_clause for join_type, table, on_left, on_right in self._join_clauses: join_clause f {join_type} {table} ON {on_left} {on_right} # WHERE部分 where_clause self._build_where_clause() # GROUP BY部分 group_by_clause if self._group_by_columns: group_by_clause fGROUP BY {, .join(self._group_by_columns)} # HAVING部分 having_clause self._build_having_clause() # ORDER BY部分 order_by_clause if self._order_by_clauses: order_parts [f{col} {direction} for col, direction in self._order_by_clauses] order_by_clause fORDER BY {, .join(order_parts)} # LIMIT/OFFSET部分 limit_clause if self._limit_value is not None: limit_clause fLIMIT {self._limit_value} offset_clause if self._offset_value is not None: offset_clause fOFFSET {self._offset_value} # 组装完整SQL query_parts [ select_clause, from_clause, join_clause, where_clause, group_by_clause, having_clause, order_by_clause, limit_clause, offset_clause ] query .join(part for part in query_parts if part) return query, self._params def _build_where_clause(self) - str: 构建WHERE子句 if not self._where_conditions: return where_parts [] for column, operator, value in self._where_conditions: if operator IN: placeholders self._get_placeholders(len(value)) where_parts.append(f{column} IN ({, .join(placeholders)})) self._params.extend(value) elif operator BETWEEN: start, end value placeholder1 self._get_placeholder() placeholder2 self._get_placeholder() where_parts.append(f{column} BETWEEN {placeholder1} AND {placeholder2}) self._params.extend([start, end]) elif operator in [IS NULL, IS NOT NULL]: where_parts.append(f{column} {operator}) else: placeholder self._get_placeholder() where_parts.append(f{column} {operator} {placeholder}) self._params.append(value) return fWHERE { AND .join(where_parts)} def _build_having_clause(self) - str: 构建HAVING子句 if not self._having_conditions: return having_parts [] for column, operator, value in self._having_conditions: placeholder self._get_placeholder() having_parts.append(f{column} {operator} {placeholder}) self._params.append(value) return fHAVING { AND .join(having_parts)} def _get_placeholder(self) - str: 获取参数占位符 if self.db_type postgresql: return f${len(self._params) 1} else: return %s def _get_placeholders(self, count: int) - List[str]: 获取多个参数占位符 return [self._get_placeholder() for _ in range(count)] # 使用示例 if __name__ __main__: # 构建复杂查询 qb QueryBuilder(orders, db_typepostgresql) qb.select([orders.id, orders.total, users.name]) .join(users, orders.user_id, users.id) .where(orders.status, , completed) .where(orders.total, , 100) .where_in(orders.payment_method, [credit_card, paypal]) .order_by(orders.created_at, DESC) .limit(20) query, params qb.build() print(Query:, query) print(Params:, params) # 输出: # Query: SELECT orders.id, orders.total, users.name FROM orders # INNER JOIN users ON orders.user_id users.id # WHERE orders.status $1 AND orders.total $2 # AND orders.payment_method IN ($3, $4) # ORDER BY orders.created_at DESC LIMIT 20 # Params: [completed, 100, credit_card, paypal]3.2 数据库迁移工具# src/connectors/database/migration.py 数据库迁移工具 支持版本化的数据库结构变更 import os import hashlib from typing import List, Dict, Any from datetime import datetime import logging logger logging.getLogger(__name__) class Migration: 数据库迁移 特性 - 版本控制 - 回滚支持 - 迁移历史记录 - 校验和验证 def __init__(self, version: str, name: str, up_sql: str, down_sql: str): self.version version self.name name self.up_sql up_sql self.down_sql down_sql self.checksum self._calculate_checksum() def _calculate_checksum(self) - str: 计算迁移文件的校验和 content f{self.version}{self.name}{self.up_sql}{self.down_sql} return hashlib.sha256(content.encode()).hexdigest() class MigrationManager: 迁移管理器 负责执行和管理数据库迁移 def __init__(self, pool: ConnectionPool): self.pool pool self.migrations: List[Migration] [] async def initialize(self): 初始化迁移表 create_table_sql CREATE TABLE IF NOT EXISTS schema_migrations ( version VARCHAR(255) PRIMARY KEY, name VARCHAR(255) NOT NULL, checksum VARCHAR(64) NOT NULL, applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) await self.pool.execute(create_table_sql) logger.info(Migration table initialized) def add_migration(self, migration: Migration): 添加迁移 self.migrations.append(migration) async def get_applied_migrations(self) - List[str]: 获取已应用的迁移版本 rows await self.pool.fetch( SELECT version FROM schema_migrations ORDER BY version ) return [row[version] for row in rows] async def migrate_up(self, target_version: Optional[str] None): 执行向上迁移 Args: target_version: 目标版本None表示迁移到最新 applied await self.get_applied_migrations() for migration in sorted(self.migrations, keylambda m: m.version): # 跳过已应用的迁移 if migration.version in applied: continue # 如果指定了目标版本只迁移到该版本 if target_version and migration.version target_version: break logger.info(fApplying migration: {migration.version} - {migration.name}) try: async with self.pool.acquire() as conn: async with conn.transaction(): # 执行迁移SQL await conn.execute(migration.up_sql) # 记录迁移历史 await conn.execute( INSERT INTO schema_migrations (version, name, checksum) VALUES ($1, $2, $3) , migration.version, migration.name, migration.checksum ) logger.info(fMigration {migration.version} applied successfully) except Exception as e: logger.error(fMigration {migration.version} failed: {e}) raise async def migrate_down(self, target_version: str): 执行向下迁移回滚 Args: target_version: 目标版本 applied await self.get_applied_migrations() for migration in sorted(self.migrations, keylambda m: m.version, reverseTrue): # 只回滚已应用的迁移 if migration.version not in applied: continue # 回滚到目标版本为止 if migration.version target_version: break logger.info(fRolling back migration: {migration.version} - {migration.name}) try: async with self.pool.acquire() as conn: async with conn.transaction(): # 执行回滚SQL await conn.execute(migration.down_sql) # 删除迁移记录 await conn.execute( DELETE FROM schema_migrations WHERE version $1, migration.version ) logger.info(fMigration {migration.version} rolled back successfully) except Exception as e: logger.error(fRollback of {migration.version} failed: {e}) raise async def verify_migrations(self) - bool: 验证迁移完整性 检查已应用迁移的校验和是否匹配 rows await self.pool.fetch( SELECT version, checksum FROM schema_migrations ) applied_checksums {row[version]: row[checksum] for row in rows} for migration in self.migrations: if migration.version in applied_checksums: if migration.checksum ! applied_checksums[migration.version]: logger.error( fMigration {migration.version} checksum mismatch! fExpected: {migration.checksum}, fGot: {applied_checksums[migration.version]} ) return False logger.info(All migrations verified successfully) return True # 使用示例 async def example_migrations(): 迁移使用示例 from .pool_manager import ConnectionPool # 创建连接池 pool ConnectionPool(postgresql://user:passlocalhost/mydb) await pool.initialize() # 创建迁移管理器 manager MigrationManager(pool) await manager.initialize() # 添加迁移 manager.add_migration(Migration( version001, namecreate_users_table, up_sql CREATE TABLE users ( id SERIAL PRIMARY KEY, name VARCHAR(100) NOT NULL, email VARCHAR(255) UNIQUE NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) , down_sqlDROP TABLE users )) manager.add_migration(Migration( version002, nameadd_users_status_column, up_sqlALTER TABLE users ADD COLUMN status VARCHAR(20) DEFAULT active, down_sqlALTER TABLE users DROP COLUMN status )) # 执行迁移 await manager.migrate_up() # 验证迁移 is_valid await manager.verify_migrations() print(fMigrations valid: {is_valid}) # 回滚到版本001 # await manager.migrate_down(001) await pool.close()3.3 连接池监控# src/connectors/database/monitoring.py 数据库连接池监控 提供实时监控和告警能力 import asyncio from typing import Dict, Any, List from datetime import datetime, timedelta import logging logger logging.getLogger(__name__) class PoolMonitor: 连接池监控器 特性 - 实时指标收集 - 性能统计 - 异常检测 - 告警通知 def __init__(self, pool: ConnectionPool): self.pool pool self.metrics: List[Dict[str, Any]] [] self._monitoring False self._monitor_task None async def start_monitoring(self, interval: int 10): 开始监控 Args: interval: 监控间隔秒 if self._monitoring: logger.warning(Monitoring already started) return self._monitoring True self._monitor_task asyncio.create_task(self._monitor_loop(interval)) logger.info(fPool monitoring started (interval: {interval}s)) async def stop_monitoring(self): 停止监控 if not self._monitoring: return self._monitoring False if self._monitor_task: self._monitor_task.cancel() try: await self._monitor_task except asyncio.CancelledError: pass logger.info(Pool monitoring stopped) async def _monitor_loop(self, interval: int): 监控循环 while self._monitoring: try: # 收集指标 metrics await self._collect_metrics() self.metrics.append(metrics) # 保留最近1小时的数据 cutoff_time datetime.now() - timedelta(hours1) self.metrics [ m for m in self.metrics if m[timestamp] cutoff_time ] # 检查异常 await self._check_anomalies(metrics) await asyncio.sleep(interval) except Exception as e: logger.error(fMonitoring error: {e}) async def _collect_metrics(self) - Dict[str, Any]: 收集连接池指标 status await self.pool.get_pool_status() return { timestamp: datetime.now(), pool_size: status.get(size, 0), free_connections: status.get(free, 0), active_connections: status.get(size, 0) - status.get(free, 0), utilization: (status.get(size, 0) - status.get(free, 0)) / max(status.get(size, 1), 1), min_size: status.get(min_size, 0), max_size: status.get(max_size, 0) } async def _check_anomalies(self, metrics: Dict[str, Any]): 检查异常情况 # 检查连接池利用率 if metrics[utilization] 0.9: logger.warning( fHigh pool utilization: {metrics[utilization]:.1%} f({metrics[active_connections]}/{metrics[pool_size]}) ) # 检查是否达到最大连接数 if metrics[pool_size] metrics[max_size]: logger.warning( fPool at maximum capacity: {metrics[pool_size]}/{metrics[max_size]} ) def get_statistics(self) - Dict[str, Any]: 获取统计信息 if not self.metrics: return {} utilizations [m[utilization] for m in self.metrics] active_conns [m[active_connections] for m in self.metrics] return { avg_utilization: sum(utilizations) / len(utilizations), max_utilization: max(utilizations), avg_active_connections: sum(active_conns) / len(active_conns), max_active_connections: max(active_conns), sample_count: len(self.metrics), time_range: { start: self.metrics[0][timestamp].isoformat(), end: self.metrics[-1][timestamp].isoformat() } } 四、完整测试套件4.1 单元测试# tests/test_database_connector.py 数据库连接器单元测试 import pytest import asyncio from connectors.database.database_connector import DatabaseConnector pytest.fixture async def db_connector(): 创建数据库连接器实例 connector DatabaseConnector() yield connector await connector.cleanup() pytest.mark.asyncio async def test_query_users(db_connector): 测试查询用户 result await db_connector.execute_action( action_namequery, parameters{ connection_string: postgresql://test:testlocalhost/testdb, query: SELECT * FROM users WHERE status $1 LIMIT $2, parameters: [active, 10], timeout: 10 }, credentials{} ) assert rows in result assert row_count in result assert result[row_count] 10 pytest.mark.asyncio async def test_insert_user(db_connector): 测试插入用户 result await db_connector.execute_action( action_nameinsert, parameters{ connection_string: postgresql://test:testlocalhost/testdb, table: users, data: { name: Test User, email: testexample.com, status: active }, return_id: True }, credentials{} ) assert inserted_id in result assert result[affected_rows] 1 pytest.mark.asyncio async def test_bulk_insert(db_connector): 测试批量插入 data [ {name: fUser {i}, email: fuser{i}example.com} for i in range(100) ] result await db_connector.execute_action( action_namebulk_insert, parameters{ connection_string: postgresql://test:testlocalhost/testdb, table: users, data: data, batch_size: 50 }, credentials{} ) assert result[inserted_count] 100 assert result[batch_count] 2 pytest.mark.asyncio async def test_transaction(db_connector): 测试事务 result await db_connector.execute_action( action_nametransaction, parameters{ connection_string: postgresql://test:testlocalhost/testdb, statements: [ { query: UPDATE users SET balance balance - $1 WHERE id $2, params: [100, 1] }, { query: UPDATE users SET balance balance $1 WHERE id $2, params: [100, 2] } ], isolation_level: serializable }, credentials{} ) assert result[success] is True assert result[statements_executed] 2 pytest.mark.asyncio async def test_sql_injection_prevention(db_connector): 测试SQL注入防护 with pytest.raises(ValueError): await db_connector.execute_action( action_nameinsert, parameters{ connection_string: postgresql://test:testlocalhost/testdb, table: users; DROP TABLE users; --, # 恶意表名 data: {name: Hacker} }, credentials{} )4.2 性能基准测试# tests/benchmark_database.py 数据库连接器性能基准测试 import asyncio import time from connectors.database.database_connector import DatabaseConnector async def benchmark_query_performance(): 测试查询性能 connector DatabaseConnector() # 预热 await connector.execute_action( query, { connection_string: postgresql://test:testlocalhost/testdb, query: SELECT 1, parameters: [] }, {} ) # 基准测试 iterations 100 start_time time.time() for _ in range(iterations): await connector.execute_action( query, { connection_string: postgresql://test:testlocalhost/testdb, query: SELECT * FROM users LIMIT 100, parameters: [] }, {} ) elapsed time.time() - start_time qps iterations / elapsed print(fQuery Performance: {qps:.2f} QPS) print(fAverage latency: {elapsed/iterations*1000:.2f} ms) await connector.cleanup() async def benchmark_bulk_insert(): 测试批量插入性能 connector DatabaseConnector() data [ {name: fUser {i}, email: fuser{i}example.com} for i in range(10000) ] start_time time.time() await connector.execute_action( bulk_insert, { connection_string: postgresql://test:testlocalhost/testdb, table: users, data: data, batch_size: 1000 }, {} ) elapsed time.time() - start_time throughput len(data) / elapsed print(fBulk Insert Performance: {throughput:.2f} rows/sec) print(fTotal time: {elapsed:.2f} seconds) await connector.cleanup() if __name__ __main__: asyncio.run(benchmark_query_performance()) asyncio.run(benchmark_bulk_insert()) 五、本文小结核心要点回顾连接池管理单例模式复用连接池自动重连机制连接健康检查优雅关闭安全性保障参数化查询防SQL注入表名/列名验证WHERE条件强制DELETE操作权限控制连接字符串性能优化批量插入分批处理连接池复用查询超时控制结果行数限制事务支持ACID保证自动回滚隔离级别控制错误处理高级特性查询构建器类型安全数据库迁移版本控制连接池监控实时告警最佳实践✅ 始终使用参数化查询✅ 验证所有用户输入✅ 设置合理的超时时间✅ 批量操作分批处理✅ 事务中包含错误处理✅ 监控连接池状态✅ 定期清理空闲连接 六、本文资源完整代码git clone https://github.com/quantumflow/database-connector.git cd database-connector pip install -r requirements.txt # 运行测试 pytest tests/test_database_connector.py -v # 性能基准测试 python tests/benchmark_database.py数据库脚本-- 创建测试表 CREATE TABLE users ( id SERIAL PRIMARY KEY, name VARCHAR(100) NOT NULL, email VARCHAR(255) UNIQUE NOT NULL, status VARCHAR(20) DEFAULT active, balance DECIMAL(10, 2) DEFAULT 0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); -- 创建索引 CREATE INDEX idx_users_status ON users(status); CREATE INDEX idx_users_email ON users(email); 七、思考题连接池优化如何根据业务负载动态调整连接池大小死锁处理在高并发场景下如何检测和处理数据库死锁读写分离如何实现主从数据库的读写分离数据分片如何支持分库分表的场景监控告警如何监控慢查询和连接池异常欢迎在评论区分享你的解决方案作者DREAMVFIA专注于企业级应用架构设计与工作流自动化技术版权声明本文为QuantumFlow专栏原创内容遵循MIT开源协议。