要使用 Python 3.6 和 `pymssql` 模块连接并查询同一 SQL Server 服务器下的多个数据库及其表，核心在于建立连接后切换数据库或为每个数据库建立独立的连接上下文。以下方案将详细阐述连接配置、查询方法，并结合具体示例代码。 首先，确保已安装 `pymssql` 模块。虽然 Python 3.6 相对较旧，但 `pymssql` 有兼容版本。可使用 pip 安装：`pip install pymssql`。如果遇到安装问题，可能需要指定版本或安装预编译的 wheel 文件 [ref_1]。 ### 1. 连接信息与策略规划 假设服务器地址为 `localhost`，采用 SQL Server 身份验证，用户名为 `sa`，密码为 `your_password`。三个目标数据库分别命名为 `DB1`、`DB2`、`DB3`。每个数据库下有两个表，例如 `DB1` 有 `Table1_1` 和 `Table1_2`。 查询访问有两种主要策略： 1. **单连接，动态切换数据库**：建立一个到服务器默认数据库（如 `master`）的连接，然后通过执行 `USE DatabaseName` 语句来切换当前连接的数据库上下文。 2. **多连接，各库独立**：为每个目标数据库分别建立一个独立的连接对象。 | 策略 | 优点 | 缺点 | 适用场景 | | :--- | :--- | :--- | :--- | | **动态切换** | 连接开销小，只需一个连接对象。 | 需要显式执行 `USE` 语句切换，在并发或异步操作中可能因上下文冲突出错。 | 顺序、串行地处理多个数据库的任务。 | | **独立连接** | 逻辑清晰，各数据库操作互不干扰，易于管理。 | 需要创建多个连接对象，占用稍多资源。 | 需要同时或独立处理多个数据库，或代码结构要求分离度高的场景。 | ### 2. 方案一：单连接动态切换数据库 此方案先连接到服务器，然后通过 `USE` 语句切换至目标数据库进行查询。 ```python import pymssql # 服务器连接参数 server = 'localhost' user = 'sa' password = 'your_password' # 目标数据库列表 databases = ['DB1', 'DB2', 'DB3'] try: # 1. 建立连接，初始可以连接到 master 或任意一个存在的数据库 conn = pymssql.connect(server=server, user=user, password=password, database='master') cursor = conn.cursor() print("成功连接到服务器") # 2. 遍历每个数据库进行查询 for db_name in databases: print(f"\n--- 切换到数据库 [{db_name}] 进行查询 ---") # 切换数据库上下文 cursor.execute(f'USE [{db_name}]') # 使用方括号避免数据库名含特殊字符 # 注意：pymssql 的 cursor.execute() 执行 USE 后，该游标后续操作即在该数据库下 # 示例：查询当前数据库下的所有表名（假设每个数据库有两个表） # 这里使用系统视图查询表名，实际可根据已知表名直接查询 cursor.execute(""" SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_TYPE = 'BASE TABLE' """) tables = cursor.fetchall() print(f"数据库 {db_name} 下的表有: {[table[0] for table in tables]}") # 3. 对已知的每个表执行查询 (这里假设表名已知，用 TableX_Y 表示) # 例如，查询第一个表的所有数据 table_to_query = f'Table{databases.index(db_name)+1}_1' # 构造示例表名 try: cursor.execute(f'SELECT TOP 5 * FROM [{table_to_query}]') # 查询前5行数据 rows = cursor.fetchall() print(f"表 [{table_to_query}] 的前5行数据:") for row in rows: print(row) except Exception as e: print(f"查询表 {table_to_query} 时出错: {e}") # 关闭游标和连接 cursor.close() conn.close() print("\n所有操作完成，连接已关闭。") except pymssql.OperationalError as e: print(f"连接数据库时发生错误: {e}") # 可能的原因：服务器地址错误、端口未开启(默认1433)、防火墙设置、SQL Server配置未允许远程连接等 [ref_4][ref_6] ``` **关键点说明**： * `pymssql.connect()` 中的 `database` 参数用于指定初始连接的数据库，可以设为 `'master'` [ref_1]。 * `cursor.execute('USE [DBName]')` 是切换当前连接操作数据库的关键。执行后，该游标后续的所有 SQL 命令都在指定的 `DBName` 中执行。 * 查询特定表时，使用 `SELECT * FROM [TableName]`。`TOP n` 子句用于限制返回行数，避免数据量过大。 * 务必使用 `try...except` 块捕获可能出现的操作错误（如 `pymssql.OperationalError`, `pymssql.ProgrammingError`）。 ### 3. 方案二：多连接独立访问数据库 此方案为每个数据库创建一个独立的连接，更利于模块化代码或并行处理。 ```python import pymssql # 服务器连接参数 server = 'localhost' user = 'sa' password = 'your_password' # 目标数据库列表及其已知表结构（示例） db_configs = { 'DB1': ['Table1_1', 'Table1_2'], 'DB2': ['Table2_1', 'Table2_2'], 'DB3': ['Table3_1', 'Table3_2'] } # 存储所有连接和游标的字典，便于后续管理 connections = {} cursors = {} try: for db_name, tables in db_configs.items(): print(f"\n--- 连接到数据库 [{db_name}] ---") # 为每个数据库创建独立连接 conn = pymssql.connect(server=server, user=user, password=password, database=db_name) cursor = conn.cursor() connections[db_name] = conn cursors[db_name] = cursor print(f"成功连接到 {db_name}") # 查询该数据库下的所有表（验证连接和表信息） cursor.execute(""" SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_TYPE = 'BASE TABLE' """) existing_tables = [row[0] for row in cursor.fetchall()] print(f"实际存在的表: {existing_tables}") # 遍历并查询该数据库下已知的每个表 for table_name in tables: if table_name in existing_tables: try: # 执行查询，例如获取表行数 cursor.execute(f'SELECT COUNT(*) AS row_count FROM [{table_name}]') count = cursor.fetchone()[0] print(f" 表 [{table_name}] 的行数为: {count}") # 示例：查询前3行数据的具体内容 cursor.execute(f'SELECT TOP 3 * FROM [{table_name}]') sample_data = cursor.fetchall() if sample_data: print(f" 前3行数据样本:") for idx, row in enumerate(sample_data, 1): print(f" 行{idx}: {row}") except Exception as e: print(f" 查询表 {table_name} 时出错: {e}") else: print(f" 表 [{table_name}] 在数据库中不存在。") # 所有数据库操作完毕后，统一提交事务并关闭连接 for db_name, conn in connections.items(): conn.commit() # 提交所有未提交的事务 [ref_2] cursors[db_name].close() conn.close() print("\n所有数据库操作已完成，连接已关闭。") except pymssql.Error as e: print(f"发生数据库错误: {e}") # 发生错误时，尝试关闭所有已打开的连接 for db_name, conn in connections.items(): try: conn.close() except: pass ``` **关键点说明**： * 在 `pymssql.connect()` 中直接指定目标 `database=db_name`，建立到特定数据库的专有连接 [ref_1]。 * 每个连接对象都是独立的，对其游标的操作不会影响其他数据库，逻辑更清晰。 * 使用字典 (`db_configs`) 来管理数据库名和其对应的表列表，使配置更灵活。 * 操作完成后，需要遍历并提交每个连接的事务 (`conn.commit()`) 以确保数据修改（如果有）被保存，然后关闭所有游标和连接 [ref_2]。 * 对于纯查询操作，`commit()` 不是必须的，但养成显式提交或关闭的习惯是良好的。 ### 4. 扩展：使用 pandas 进行数据查询与分析 如果需要对查询结果进行进一步的数据分析，结合 `pandas` 和 `sqlalchemy`（通过 `pymssql` 驱动）或直接使用 `pandas.read_sql` 与 `pymssql` 连接会更加高效 [ref_5]。 首先安装 `pandas`：`pip install pandas`。 ```python import pandas as pd import pymssql server = 'localhost' user = 'sa' password = 'your_password' database = 'DB1' # 指定要操作的数据库 # 建立连接 conn = pymssql.connect(server=server, user=user, password=password, database=database) # 使用 pandas 直接读取 SQL 查询结果到 DataFrame query = "SELECT * FROM Table1_1 WHERE some_column > 100" # 示例查询 try: df = pd.read_sql(query, conn) print(f"查询成功，获取到 {len(df)} 行数据。") print(df.head()) # 查看前几行数据 # 接下来可以使用 pandas 进行各种数据分析操作 # 例如：print(df.describe()) # 统计描述 except Exception as e: print(f"使用 pandas 查询时出错: {e}") finally: conn.close() ``` 此方法特别适合于数据探索、清洗和可视化前的数据准备阶段 [ref_5]。 ### 5. 常见连接问题与排查 若连接失败，请按以下步骤排查： 1. **网络与配置**：确认服务器 IP、端口（默认 1433）正确，防火墙已放行该端口，且 SQL Server 已配置允许远程连接（通过 SQL Server Configuration Manager 启用 TCP/IP 协议）[ref_4][ref_6]。 2. **身份验证**：确认用户名和密码正确。如果是 Windows 身份验证，连接字符串不同（通常使用 `trusted_connection=True` 参数，但 `pymssql` 对 Windows 身份验证的支持取决于环境）[ref_1]。 3. **驱动与编码**：确保 `pymssql` 版本与 Python 3.6 兼容。若查询结果出现中文乱码，可在连接字符串中添加 `charset='utf8'` 参数，或确保数据库表的字符集兼容 [ref_2][ref_3][ref_5]。 4. **数据库存在性**：确认 `database` 参数指定的数据库名在服务器上确实存在。 通过以上两种核心策略（动态切换与独立连接）并结合具体示例代码，可以灵活地实现使用 Python 3.6 和 `pymssql` 对同一 SQL Server 服务器下多个数据库及其表的查询访问。选择哪种策略取决于具体的应用场景和对资源、复杂度的权衡。