<think>我们之前已经详细讨论了Python建立数据库连接的两种主要方法：原生SQL连接（如PyMySQL）和ORM框架（如SQLAlchemy）。现在，根据用户的问题，他想要查找Python建立与数据库连接方法的解决方案。因此，我们需要再次总结并提供清晰的步骤和代码示例。 由于用户提到“参考站内引用”，但引用内容实际上是关于如何全局管理数据库连接的问题（避免重复创建连接和变量重定义），因此我们将重点放在如何安全地管理全局数据库连接上，同时提供两种连接方式的实现。 ### 全局数据库连接管理的最佳实践 根据引用[^1]中的问题，用户希望有一个全局的数据库连接，可以在脚本的任何地方访问，但又不容易出错（比如避免意外重新定义连接变量）。通常，我们可以使用模块级别的变量来保存连接，并通过函数来获取连接。同时，为了确保连接在不需要时关闭，我们可以使用上下文管理器（如`with`语句）或连接池。 #### 方案一：使用PyMySQL实现全局连接（单例模式） ```python import pymysql # 模块级别全局变量 _db_connection = None def get_connection(): global _db_connection if _db_connection is None: try: _db_connection = pymysql.connect( host='localhost', user='your_username', password='your_pwd', database='your_db', charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor ) except pymysql.Error as e: print(f"数据库连接失败: {e}") raise return _db_connection def close_connection(): global _db_connection if _db_connection is not None: _db_connection.close() _db_connection = None # 使用示例 def query_users(): conn = get_connection() try: with conn.cursor() as cursor: cursor.execute("SELECT * FROM users") return cursor.fetchall() except pymysql.Error as e: print(f"查询失败: {e}") # 注意：这里不关闭连接，因为全局连接在其他地方还要用 # 在程序退出时关闭连接 if __name__ == "__main__": try: # 执行一些操作 users = query_users() print(users) finally: close_connection() ``` **注意**：这种全局连接的方式在长时间运行的程序中可能不是最佳，因为连接可能会中断。更健壮的做法是使用连接池或每次需要时创建连接（但开销大）。另外，多线程环境下需要加锁。 #### 方案二：使用SQLAlchemy的全局会话管理（推荐） SQLAlchemy提供了连接池和会话管理，可以很好地处理全局连接问题。 ```python from sqlalchemy import create_engine from sqlalchemy.orm import scoped_session, sessionmaker from sqlalchemy.ext.declarative import declarative_base # 创建引擎（连接池默认开启） engine = create_engine('mysql+pymysql://user:pwd@localhost/db_name', pool_recycle=3600) # 使用scoped_session创建线程安全的会话工厂 db_session = scoped_session(sessionmaker(autocommit=False, autoflush=False, bind=engine)) Base = declarative_base() Base.query = db_session.query_property() # 定义模型 class User(Base): __tablename__ = 'users' id = Column(Integer, primary_key=True) name = Column(String(50)) email = Column(String(100)) # 使用示例 def get_all_users(): return User.query.all() # 应用结束时关闭所有连接 def shutdown_session(exception=None): db_session.remove() # 注册关闭函数，比如在Flask中可以使用teardown_appcontext # 或者直接在使用完后调用db_session.remove()（在每个请求结束时） if __name__ == "__main__": try: users = get_all_users() print(users) finally: shutdown_session() ``` **说明**： - `scoped_session`为每个线程（或协程）创建一个独立的会话，避免了多线程冲突。 - 使用`db_session.remove()`会在当前上下文结束后将会话返回连接池，并关闭连接（如果超过连接池大小）。 ### 两种方法的比较 | 方法 | 优点 | 缺点 | |----------------|--------------------------------------------|--------------------------------------| | PyMySQL全局连接 | 简单直接，适合小型脚本 | 需手动管理连接关闭，多线程不安全 | | SQLAlchemy | 自动连接池，线程安全，支持事务和ORM | 学习曲线较陡，需要理解ORM和会话管理 | ### 最佳实践建议 1. **小型脚本**：使用PyMySQL，但考虑使用上下文管理器（`with`语句）来确保连接关闭。 2. **Web应用或长期运行服务**：使用SQLAlchemy的连接池和会话管理，避免连接泄漏。 3. **多线程环境**：必须使用线程安全的连接管理（如`scoped_session`）。 ### 相关问题 §§相关问题§§ 1. 如何用Python实现数据库连接池管理？ 2. SQLAlchemy中的Session对象有哪些高级用法？ 3. 异步Python中如何连接数据库（如aiomysql）？ 4. 如何用Python实现数据库事务的回滚机制？ 5. 不同数据库（MySQL/PostgreSQL/SQLite）的连接参数有何区别？ </think> ### Python建立数据库连接的详细方法 在Python中建立数据库连接主要有两种方式：**原生SQL连接**（如PyMySQL）和**ORM框架**（如SQLAlchemy）。以下是具体实现方法： --- #### 一、原生SQL连接（以PyMySQL为例） 通过PyMySQL库直接操作数据库，适合需要精细控制SQL语句的场景。 ```python import pymysql # 建立数据库连接 def create_connection(): try: # 关键参数：主机、用户名、密码、数据库名 conn = pymysql.connect( host='localhost', # 数据库主机地址 user='your_username', # 数据库用户名 password='your_pwd', # 数据库密码 database='your_db', # 数据库名称 charset='utf8mb4', # 字符编码 cursorclass=pymysql.cursors.DictCursor # 返回字典形式结果 ) print("数据库连接成功") return conn except pymysql.Error as e: print(f"连接失败: {e}") return None # 执行查询 def query_data(sql): conn = create_connection() if conn: try: with conn.cursor() as cursor: cursor.execute(sql) # 执行SQL语句 result = cursor.fetchall() # 获取所有结果 return result finally: conn.close() # 确保关闭连接 # 使用示例 data = query_data("SELECT * FROM users WHERE age > 30") print(data) ``` **关键点说明**： 1. 使用`pymysql.connect()`建立连接[^1] 2. 通过`cursor.execute()`执行SQL命令 3. 用`fetchall()`/`fetchone()`获取结果 4. **必须显式关闭连接**（避免资源泄漏） 5. 参数化查询防止SQL注入： ```python cursor.execute("SELECT * FROM users WHERE name = %s", (user_input,)) ``` --- #### 二、ORM框架连接（推荐使用SQLAlchemy） 通过对象关系映射（ORM）管理数据库，提高代码可维护性和安全性。 ```python from sqlalchemy import create_engine, Column, Integer, String from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.declarative import declarative_base # 1. 创建引擎（自动管理连接池） engine = create_engine('mysql+pymysql://user:pwd@localhost/db_name') # 2. 定义数据模型 Base = declarative_base() class User(Base): __tablename__ = 'users' id = Column(Integer, primary_key=True) name = Column(String(50)) email = Column(String(100)) # 3. 创建会话工厂 Session = sessionmaker(bind=engine) # 4. 执行数据库操作 def add_user(name, email): session = Session() # 从连接池获取连接 try: new_user = User(name=name, email=email) session.add(new_user) session.commit() # 自动提交事务 except Exception as e: session.rollback() print(f"操作失败: {e}") finally: session.close() # 连接返回到连接池 # 使用示例 add_user("Alice", "alice@example.com") ``` **SQLAlchemy核心优势**： 1. **自动连接管理**：内置连接池，无需手动关闭 2. **数据库无关性**：相同代码支持MySQL/PostgreSQL/SQLite等 3. **防止SQL注入**：自动转义参数 4. **面向对象操作**：用Python类替代原生SQL 5. **事务自动处理**：简化`commit`/`rollback`逻辑 --- ### 两种方法对比 | 特性 | 原生SQL (PyMySQL) | ORM (SQLAlchemy) | |---------------------|-------------------------|-------------------------| | 学习曲线 | 低（需懂SQL） | 中（需理解ORM概念） | | 性能 | 高（直接执行SQL） | 中（有ORM转换开销） | | 代码可维护性 | 低（SQL混杂在代码中） | 高（清晰的OOP结构） | | 安全性 | 需手动防注入 | 自动参数化查询 | | 适合场景 | 简单查询/性能敏感场景 | 中大型项目/团队协作 | > **最佳实践建议**： > 小型项目可用PyMySQL快速实现；中大型项目强烈推荐SQLAlchemy，其连接池和对象映射能显著提升开发效率和代码质量[^1]。