Python 关系型数据库操作完全指南

import sqlite3

# 若文件不存在，自动创建
conn = sqlite3.connect('data.db')

sqlite3.connect(
    database,           # 数据库文件路径，或 ':memory:' 创建内存数据库
    timeout=5.0,        # 数据库被锁时等待的时间（秒）
    detect_types=0,     # 是否自动检测列类型
    isolation_level='', # 事务隔离级别（见下文）
    check_same_thread=True,  # 是否检查线程安全
    uri=False,          # 是否将 database 参数解释为 URI
)

import sqlite3

# 推荐写法：自动管理事务
with sqlite3.connect('data.db') as conn:
    cursor = conn.cursor()
    cursor.execute('CREATE TABLE IF NOT EXISTS users(id INTEGER PRIMARY KEY, name TEXT)')
    cursor.execute('INSERT INTO users(name) VALUES(?)', ('Alice',))
    # with 块结束时自动 commit；若抛异常则自动 rollback

with sqlite3.connect('data.db') as conn:
    with conn:
        # 双层 with：
        #   conn.__exit__  → commit/rollback
        #   conn.cursor().__exit__ → close cursor
        conn.execute('CREATE TABLE IF NOT EXISTS log(ts TEXT, msg TEXT)')
        conn.execute('INSERT INTO log VALUES(datetime("now"), ?)', ('service started',))

import sqlite3

with sqlite3.connect('school.db') as conn:
    conn.execute('''
        CREATE TABLE IF NOT EXISTS student (
            id    INTEGER PRIMARY KEY AUTOINCREMENT,
            name  TEXT    NOT NULL,
            age   INTEGER CHECK(age > 0 AND age < 150),
            score REAL    DEFAULT 0.0
        )
    ''')

    # 单行插入
    conn.execute(
        'INSERT INTO student(name, age, score) VALUES(?, ?, ?)',
        ('张三', 20, 88.5)
    )

    # 多行批量插入
    students = [
        ('李四', 22, 91.0),
        ('王五', 19, 76.5),
        ('赵六', 21, 85.0),
    ]
    conn.executemany('INSERT INTO student(name, age, score) VALUES(?, ?, ?)', students)

    print(f'插入了 {conn.execute("SELECT COUNT(*) FROM student").fetchone()[0]} 条记录')

import sqlite3

with sqlite3.connect('school.db') as conn:
    # 在连接上启用 Row 工厂，使结果可以通过列名访问
    conn.row_factory = sqlite3.Row

    # 条件查询
    rows = conn.execute(
        'SELECT id, name, age, score FROM student WHERE score >= ? ORDER BY score DESC',
        (80,)
    ).fetchall()

    for row in rows:
        # 使用列名访问，比按索引更可读
        print(f'{row["id"]:>3} | {row["name"]:<6} | {row["age"]:>3} | {row["score"]:>5}')

with sqlite3.connect('school.db') as conn:
    # 更新
    affected = conn.execute(
        'UPDATE student SET score = score + 5 WHERE score < 80'
    ).rowcount
    print(f'更新了 {affected} 行')

    # 删除
    affected = conn.execute(
        'DELETE FROM student WHERE id = ?', (3,)
    ).rowcount
    print(f'删除了 {affected} 行')

# 推荐：迭代 Cursor 逐行处理，不一次性加载到内存
with sqlite3.connect('large.db') as conn:
    for row in conn.execute('SELECT * FROM huge_table'):
        process(row)

# ❌ 危险：字符串拼接，攻击者输入 ' OR 1=1 --' 即可绕过认证
user_input = "admin' OR '1'='1"
sql = f"SELECT * FROM users WHERE name='{user_input}'"
# 实际执行的 SQL:
# SELECT * FROM users WHERE name='admin' OR '1'='1'
# 返回所有用户！

# ✅ 安全：参数化查询，数据库驱动自动转义
cursor.execute('SELECT * FROM users WHERE name=?', (user_input,))

import sqlite3

with sqlite3.connect('bank.db') as conn:
    # 转账操作：必须同时成功或同时失败

    # 方案一：双层 with 自动管理
    with conn:
        conn.execute('UPDATE accounts SET balance = balance - 100 WHERE id = ?', (1,))
        conn.execute('UPDATE accounts SET balance = balance + 100 WHERE id = ?', (2,))
        # 离开 with conn 块：自动 commit
        # 若任何一步抛异常：自动 rollback

    # 方案二：手动控制
    try:
        conn.execute('UPDATE accounts SET balance = balance - 100 WHERE id = ?', (1,))
        # 模拟中间出错
        # raise ValueError('模拟失败')
        conn.execute('UPDATE accounts SET balance = balance + 100 WHERE id = ?', (2,))
        conn.commit()
        print('转账成功')
    except Exception as e:
        conn.rollback()
        print(f'转账失败，已回滚: {e}')

# 自动提交模式：每条语句立即生效
conn = sqlite3.connect('data.db', isolation_level=None)

# WAL 模式 + 立即事务：高并发场景推荐
conn = sqlite3.connect('data.db', isolation_level='IMMEDIATE')
conn.execute('PRAGMA journal_mode=WAL')  # 启用 Write-Ahead Logging

import sqlite3

with sqlite3.connect('analytics.db') as conn:
    # 开启 rollback 日志（可选）
    conn.execute('PRAGMA journal_mode=WAL')

    # 通过隔离 savepoint 保留部分回滚数据
    conn.execute("SAVEPOINT before_batch")
    try:
        conn.execute("INSERT INTO events VALUES(?, ?)", ("page_view", 100))
        conn.execute("INSERT INTO events VALUES(?, ?)", ("click", 50))
        # ...
        conn.execute("RELEASE before_batch")  # 确认保存点
    except Exception:
        conn.execute("ROLLBACK TO before_batch")  # 只回滚到保存点
        conn.execute("INSERT INTO error_log VALUES(datetime('now'), 'batch_failed')")

import sqlite3
import hashlib

def md5(text):
    return hashlib.md5(text.encode()).hexdigest()

with sqlite3.connect(':memory:') as conn:
    conn.create_function('md5', 1, md5)
    result = conn.execute("SELECT md5('hello')").fetchone()[0]
    print(result)  # 5d41402abc4b2a76b9719d911017c592

class Median:
    """自定义聚合函数：计算中位数"""
    def __init__(self):
        self.values = []

    def step(self, value):
        if value is not None:
            self.values.append(value)

    def finalize(self):
        if not self.values:
            return None
        self.values.sort()
        n = len(self.values)
        mid = n // 2
        return self.values[mid] if n % 2 else (self.values[mid-1] + self.values[mid]) / 2

with sqlite3.connect(':memory:') as conn:
    conn.create_aggregate('median', 1, Median)
    conn.execute('CREATE TABLE scores(score REAL)')
    conn.executemany('INSERT INTO scores VALUES(?)', [(85,), (90,), (78,), (92,), (88,)])
    median = conn.execute('SELECT median(score) FROM scores').fetchone()[0]
    print(f'中位数: {median}')  # 88.0

import sqlite3

def backup_database(source_path, dest_path):
    """在线备份 SQLite 数据库"""
    source = sqlite3.connect(source_path)
    dest = sqlite3.connect(dest_path)
    with dest:
        source.backup(dest)
    source.close()
    dest.close()

backup_database('production.db', 'production_backup.db')

import sqlite3

# ❌ 慢：逐条插入，每条一个事务
# 1万条数据需要约 60 秒

# ✅ 快：批量 + 单事务
with sqlite3.connect('data.db') as conn:
    conn.execute('PRAGMA synchronous=NORMAL')
    conn.execute('PRAGMA journal_mode=WAL')
    conn.execute('CREATE TABLE IF NOT EXISTS records(id INTEGER, data TEXT)')

    batch_size = 1000
    data = [(i, f'data-{i}') for i in range(10000)]

    with conn:  # 整个批量在一个事务中
        conn.executemany('INSERT INTO records VALUES(?, ?)', data)
    # 1万条数据约 0.5 秒 — 120x 提速

pip install mysql-connector-python

import mysql.connector

conn = mysql.connector.connect(
    user='root',
    password='root1234',
    database='test'
)
cursor = conn.cursor()

# 创建表
cursor.execute('''
    CREATE TABLE user (
        id   VARCHAR(20) PRIMARY KEY,
        name VARCHAR(20) NOT NULL
    )
''')

# 插入单行 — MySQL 占位符为 %s
cursor.execute(
    'INSERT INTO user (id, name) VALUES (%s, %s)',
    ['1', 'Michael']
)
print(f'影响行数: {cursor.rowcount}')

conn.commit()
cursor.close()
conn.close()

import mysql.connector

conn = mysql.connector.connect(
    user='root', password='root1234', database='test'
)
cursor = conn.cursor()

cursor.execute('SELECT * FROM user WHERE id = %s', ('1',))
values = cursor.fetchall()
print(values)  # [('1', 'Michael')]

cursor.close()
conn.close()

pip install pymysql

import pymysql

conn = pymysql.connect(
    host="localhost",
    port=3306,
    user="root",
    password="root1234",
    database="vega"
)
cursor = conn.cursor()

sql = "INSERT INTO t_user(username, password, email, role_id) VALUES(%s, %s, %s, %s)"

# 单行插入
cursor.execute(sql, ("user1", "pwd_hash", "user1@qq.com", 1))

# 批量插入 — 性能远优于逐条 execute
data = [
    ("user2", "pwd_hash_2", "user2@qq.com", 1),
    ("user3", "pwd_hash_3", "user3@qq.com", 1),
]
cursor.executemany(sql, data)

conn.commit()
cursor.close()
conn.close()

import pymysql

conn = pymysql.connect(
    host="localhost", port=3306,
    user="root", password="root1234", database="vega"
)
cursor = conn.cursor()

sql = 'UPDATE t_user SET username = %s WHERE id = %s'

# 单行更新
cursor.execute(sql, ('new_name', 23))

# 批量更新（一次性提交多组参数）
cursor.executemany(sql, [('name_22', 22), ('name_21', 21)])

conn.commit()
cursor.close()
conn.close()

sql = 'DELETE FROM t_user WHERE id = %s'

# 单行删除
cursor.execute(sql, (3,))

# 批量删除
cursor.executemany(sql, [(4,), (5,)])

conn.commit()

cursor.execute("SELECT * FROM t_user")

# 方式一：逐行获取
row = cursor.fetchone()
print(row)

# 方式二：分批获取（适合大数据集，控制内存）
rows = cursor.fetchmany(5)
for row in rows:
    print(row)

# 方式三：全部获取（适合小数据集）
results = cursor.fetchall()
for row in results:
    print(row)

import pymysql

conn = pymysql.connect(
    host="localhost", port=3306,
    user="root", password="root1234", database="vega"
)
cursor = conn.cursor()

username = "1 OR 1=1"
password = "1 OR 1=1"

# ❌ 危险：字符串拼接
sql_bad = (
    "SELECT COUNT(*) FROM t_user WHERE username=" + username +
    " AND AES_DECRYPT(UNHEX(password),'HelloWorld')=" + password
)
cursor.execute(sql_bad)
print(f'SQL注入成功，返回: {cursor.fetchone()[0]}')  # 绕过认证

# ✅ 安全：参数化查询
sql_safe = (
    "SELECT COUNT(*) FROM t_user WHERE username=%s "
    "AND AES_DECRYPT(UNHEX(password),'HelloWorld')=%s"
)
cursor.execute(sql_safe, (username, password))
print(f'参数化查询，返回: {cursor.fetchone()[0]}')  # 0 — 注入失败

cursor.close()
conn.close()

# ✅ 动态列名：白名单校验
ALLOWED_COLUMNS = {'id', 'name', 'email', 'created_at'}
ORDER_COLUMNS = {'id', 'name'}

def safe_query(table, columns, order_by):
    if table not in ALLOWED_TABLES:
        raise ValueError(f'非法的表名: {table}')
    selected = [c for c in columns if c in ALLOWED_COLUMNS]
    if order_by not in ORDER_COLUMNS:
        raise ValueError(f'非法的排序列: {order_by}')
    sql = f"SELECT {', '.join(selected)} FROM {table} ORDER BY {order_by}"
    cursor.execute(sql)

import pymysql

conn = pymysql.connect(
    host="localhost", port=3306,
    user="root", password="root1234", database="vega"
)

try:
    cursor = conn.cursor()
    conn.begin()  # 显式开启事务

    cursor.execute("INSERT INTO t_type(type) VALUES(%s)", ('test1',))
    # 更多操作...

    conn.commit()
    print('事务提交成功')
except Exception as e:
    print(f'发生错误，回滚: {e}')
    conn.rollback()
finally:
    cursor.close()
    conn.close()

pip install dbutils

import pymysql
from dbutils.pooled_db import PooledDB


class MySQLPool:
    """线程安全的 MySQL 连接池"""

    def __init__(self, host, port, user, password, database,
                 mincached=5, maxcached=20, maxconnections=50):
        self._pool = PooledDB(
            creator=pymysql,
            maxconnections=maxconnections,  # 最大连接数上限
            mincached=mincached,            # 初始化时创建的闲置连接
            maxcached=maxcached,            # 最大闲置连接数
            blocking=True,                  # 连接耗尽时阻塞等待
            host=host,
            port=port,
            user=user,
            password=password,
            database=database,
            charset='utf8mb4'
        )

    def get_conn(self):
        """从池中获取一个连接"""
        return self._pool.connection()


# 全局单例
mysql_pool = MySQLPool(
    host="localhost", port=3306,
    user="root", password="root1234",
    database="vega"
)

# 使用连接池
conn = mysql_pool.get_conn()
try:
    with conn.cursor() as cursor:
        cursor.execute('SELECT * FROM t_user')
        for row in cursor.fetchall():
            print(row)
finally:
    conn.close()  # 归还到池中，而非真正关闭 TCP 连接

pip install sqlalchemy

from sqlalchemy import Column, String, ForeignKey, create_engine
from sqlalchemy.orm import sessionmaker, relationship, declarative_base

Base = declarative_base()


class User(Base):
    __tablename__ = 'user'

    id   = Column(String(20), primary_key=True)
    name = Column(String(20))

    # 一对多关系：一个 User 拥有多个 Book
    books = relationship('Book', back_populates='owner')


class Book(Base):
    __tablename__ = 'book'

    id      = Column(String(20), primary_key=True)
    name    = Column(String(20))
    user_id = Column(String(20), ForeignKey('user.id'))

    owner = relationship('User', back_populates='books')

# 初始化引擎与会话
engine = create_engine(
    'mysql+mysqlconnector://root:root1234@localhost:3306/test',
    echo=False,          # 不打印 SQL 日志
    pool_size=10,        # 内置连接池大小
    max_overflow=20,     # 超出 pool_size 的最大额外连接
)
Session = sessionmaker(bind=engine)
session = Session()

# --- Create ---
new_user = User(id='5', name='Bob')
session.add(new_user)
session.commit()

# --- Read ---
user = session.query(User).filter(User.id == '5').one()
print(f'type: {type(user)}, name: {user.name}')

# 条件查询 + 排序 + 分页
users = (
    session.query(User)
    .filter(User.name.like('B%'))
    .order_by(User.name)
    .limit(10)
    .all()
)

# --- Update ---
user.name = 'Bob Updated'
session.commit()  # ORM 自动跟踪脏数据

# --- Delete ---
session.delete(user)
session.commit()

session.close()

# 查询 User 时自动加载其所有 Book
user = session.query(User).filter(User.id == '5').one()
for book in user.books:
    print(f'{user.name} 拥有书籍: {book.name}')

# 从 Book 反向查询 Owner
book = session.query(Book).first()
print(f'《{book.name}》的所有者是: {book.owner.name}')

dialect+driver://username:password@host:port/database

pip install openpyxl pymysql

import openpyxl
import pymysql

workbook = openpyxl.Workbook()
sheet = workbook.active
sheet.title = '员工基本信息'
sheet.append(('工号', '姓名', '职位', '月薪', '补贴', '部门'))

conn = pymysql.connect(
    host='127.0.0.1', port=3306,
    user='guest', password='Guest.618',
    database='hrs', charset='utf8mb4'
)

try:
    with conn.cursor() as cursor:
        cursor.execute(
            'SELECT eno, ename, job, sal, COALESCE(comm, 0), dname '
            'FROM tb_emp NATURAL JOIN tb_dept'
        )
        for row in cursor:
            sheet.append(row)
    workbook.save('hrs.xlsx')
    print('导出成功: hrs.xlsx')
except pymysql.MySQLError as err:
    print(f'数据库错误: {err}')
finally:
    conn.close()