Python案例如何连接数据库？

wen python案例 2026-06-10 247

本文目录导读：

Python案例如何连接数据库？

连接MySQL数据库
连接PostgreSQL数据库
连接SQLite数据库（内置模块）
使用ORM框架（SQLAlchemy）
数据库连接池（提高性能）
完整的数据库操作封装
注意事项
最佳实践

我来详细介绍Python连接数据库的几种常见方法。

连接MySQL数据库

安装依赖

pip install pymysql
# 或者
pip install mysql-connector-python

使用PyMySQL连接

import pymysql
# 连接数据库
connection = pymysql.connect(
    host='localhost',      # 数据库主机地址
    port=3306,             # MySQL端口号
    user='root',           # 用户名
    password='123456',     # 密码
    database='test_db',    # 数据库名称
    charset='utf8mb4'      # 字符集
)
try:
    # 创建游标对象
    with connection.cursor() as cursor:
        # 执行SQL查询
        sql = "SELECT * FROM users WHERE age > %s"
        cursor.execute(sql, (20,))
        # 获取结果
        results = cursor.fetchall()
        for row in results:
            print(f"ID: {row[0]}, Name: {row[1]}, Age: {row[2]}")
    # 提交事务
    connection.commit()
finally:
    # 关闭连接
    connection.close()

使用上下文管理器（推荐方式）

from contextlib import closing
import pymysql
def query_database():
    """使用上下文管理器自动管理连接"""
    with closing(pymysql.connect(
        host='localhost',
        user='root',
        password='123456',
        database='test_db'
    )) as connection:
        with connection.cursor() as cursor:
            cursor.execute("SELECT * FROM users")
            return cursor.fetchall()
# 使用
users = query_database()
print(users)

连接PostgreSQL数据库

安装依赖

pip install psycopg2

示例代码

import psycopg2
# 连接数据库
conn = psycopg2.connect(
    host="localhost",
    port="5432",
    database="test_db",
    user="postgres",
    password="123456"
)
# 创建游标
cur = conn.cursor()
# 执行查询
cur.execute("SELECT * FROM employees WHERE salary > %s", (50000,))
# 获取数据
rows = cur.fetchall()
for row in rows:
    print(row)
# 关闭连接
cur.close()
conn.close()

连接SQLite数据库（内置模块）

import sqlite3
# 连接数据库（如果不存在会自动创建）
conn = sqlite3.connect('example.db')
# 创建游标
cursor = conn.cursor()
# 创建表
cursor.execute('''
    CREATE TABLE IF NOT EXISTS students (
        id INTEGER PRIMARY KEY,
        name TEXT NOT NULL,
        age INTEGER
    )
''')
# 插入数据
cursor.execute("INSERT INTO students (name, age) VALUES (?, ?)", 
               ("张三", 20))
# 查询数据
cursor.execute("SELECT * FROM students")
rows = cursor.fetchall()
for row in rows:
    print(row)
# 提交事务
conn.commit()
# 关闭连接
conn.close()

使用ORM框架（SQLAlchemy）

安装依赖

pip install sqlalchemy

示例代码

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 创建基类
Base = declarative_base()
# 定义模型
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    age = Column(Integer)
    def __repr__(self):
        return f"<User(name={self.name}, age={self.age})>"
# 创建数据库引擎
# MySQL: engine = create_engine('mysql+pymysql://user:password@localhost/dbname')
# PostgreSQL: engine = create_engine('postgresql://user:password@localhost/dbname')
# SQLite: engine = create_engine('sqlite:///example.db')
engine = create_engine('sqlite:///example.db')
# 创建表
Base.metadata.create_all(engine)
# 创建会话
Session = sessionmaker(bind=engine)
session = Session()
# 添加数据
new_user = User(name='李四', age=25)
session.add(new_user)
session.commit()
# 查询数据
users = session.query(User).filter(User.age > 20).all()
for user in users:
    print(user)
# 关闭会话
session.close()

数据库连接池（提高性能）

from dbutils.pooled_db import PooledDB
import pymysql
# 创建连接池
pool = PooledDB(
    creator=pymysql,
    maxconnections=6,      # 最大连接数
    mincached=2,           # 初始化时创建的最小连接数
    maxcached=5,           # 最大空闲连接数
    maxusage=None,         # 单个连接的最大复用次数
    blocking=True,         # 无可用连接时是否等待
    host='localhost',
    user='root',
    password='123456',
    database='test_db',
    charset='utf8mb4'
)
# 从连接池获取连接
def get_user_by_id(user_id):
    conn = pool.connection()
    try:
        with conn.cursor() as cursor:
            cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
            return cursor.fetchone()
    finally:
        conn.close()  # 不是真正关闭，而是返回到连接池
# 使用
user = get_user_by_id(1)
print(user)

完整的数据库操作封装

import pymysql
from typing import List, Tuple, Optional
class DatabaseManager:
    """数据库操作管理器"""
    def __init__(self, config: dict):
        self.config = config
        self.connection = None
    def connect(self) -> None:
        """建立数据库连接"""
        self.connection = pymysql.connect(**self.config)
    def disconnect(self) -> None:
        """关闭数据库连接"""
        if self.connection:
            self.connection.close()
    def execute_query(self, sql: str, params: tuple = None) -> List[Tuple]:
        """执行查询操作"""
        with self.connection.cursor() as cursor:
            cursor.execute(sql, params)
            return cursor.fetchall()
    def execute_update(self, sql: str, params: tuple = None) -> int:
        """执行更新操作"""
        with self.connection.cursor() as cursor:
            affected_rows = cursor.execute(sql, params)
            self.connection.commit()
            return affected_rows
    def insert_data(self, table: str, data: dict) -> int:
        """插入数据"""
        columns = ', '.join(data.keys())
        values = ', '.join(['%s'] * len(data))
        sql = f"INSERT INTO {table} ({columns}) VALUES ({values})"
        with self.connection.cursor() as cursor:
            cursor.execute(sql, tuple(data.values()))
            self.connection.commit()
            return cursor.lastrowid
# 使用示例
config = {
    'host': 'localhost',
    'user': 'root',
    'password': '123456',
    'database': 'test_db',
    'charset': 'utf8mb4'
}
db = DatabaseManager(config)
db.connect()
# 查询数据
users = db.execute_query("SELECT * FROM users WHERE active = %s", (1,))
# 插入数据
new_id = db.insert_data('users', {
    'name': '王五',
    'age': 28,
    'email': 'wangwu@example.com'
})
db.disconnect()

注意事项

敏感信息管理：不要硬编码数据库密码，使用环境变量或配置文件
错误处理：始终使用 try-except 处理数据库异常
资源释放：确保连接和游标被正确关闭
SQL注入防护：使用参数化查询而不是字符串拼接
连接池：生产环境建议使用连接池提高性能

最佳实践

# 使用环境变量
import os
from dotenv import load_dotenv
load_dotenv()
DB_CONFIG = {
    'host': os.getenv('DB_HOST', 'localhost'),
    'user': os.getenv('DB_USER'),
    'password': os.getenv('DB_PASSWORD'),
    'database': os.getenv('DB_NAME'),
    'charset': 'utf8mb4'
}

就是Python连接数据库的完整案例,包括了常见数据库类型和最佳实践。

本文地址： https://www.bmjup.com/post/4749.html