更新时间：2026-05-19 | 阅读时长：12分钟🗄️

---

前言

数据存在哪里？数据库。

无论是你刷抖音时的推荐算法、购物网站的订单记录，还是公司财务报表，底层都是数据库在支撑。

Python最大的优势之一，就是能和各种数据库无缝对接。今天这篇文章，带你用Python玩转数据库。

我们从最轻量的SQLite开始，到MySQL、PostgreSQL，最后用SQLAlchemy实现"写一次代码，切换任意数据库"。学完这篇，你处理数据的方式会彻底改变。

---

一、为什么学数据库？

先回答一个问题：数据存在Excel里不好吗？

Excel适合小数据、个人使用。但遇到这些情况就头疼了：

• 数据量超过100万行 → Excel卡死
• 多人同时编辑 → 版本混乱
• 需要精确查询 → 筛选条件复杂
• 数据需要实时更新 → Excel做不到

数据库就是为这些问题而生的：海量存储、并发访问、精确查询、实时更新。

---

二、轻量级首选：SQLite

SQLite是Python内置的数据库引擎，不需要安装任何服务，文件即数据库。

2.1 创建数据库

python

复制

import sqlite3

# 连接到数据库（不存在则自动创建）
conn = sqlite3.connect('my_database.db')

# 创建游标对象
cursor = conn.cursor()

print("数据库创建成功！")

运行后，当前目录会多出一个 my_database.db 文件——这就是你的数据库。

2.2 创建表

python

复制

# 创建员工表
cursor.execute('''
CREATE TABLE IF NOT EXISTS employees (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    age INTEGER,
    department TEXT,
    salary REAL,
    hire_date TEXT
)
''')

conn.commit()
print("表创建成功！")

2.3 插入数据

python

复制

# 插入单条数据
cursor.execute('''
INSERT INTO employees (name, age, department, salary, hire_date)
VALUES ('张三', 28, '技术部', 15000, '2023-01-15')
''')

# 插入多条数据
employees = [
    ('李四', 32, '市场部', 12000, '2022-06-20'),
    ('王五', 26, '技术部', 11000, '2023-03-10'),
    ('赵六', 35, '财务部', 18000, '2020-11-05'),
    ('钱七', 29, '人事部', 10000, '2023-07-01'),
]

cursor.executemany('''
INSERT INTO employees (name, age, department, salary, hire_date)
VALUES (?, ?, ?, ?, ?)
''', employees)

conn.commit()
print(f"成功插入 {cursor.rowcount + 1} 条数据！")

2.4 查询数据

python

复制

# 查询所有员工
cursor.execute('SELECT * FROM employees')
rows = cursor.fetchall()

print("\n所有员工信息：")
print("-" * 60)
for row in rows:
    print(f"ID: {row[0]} | 姓名: {row[1]} | 年龄: {row[2]} | 部门: {row[3]} | 薪资: {row[4]}")

python

复制

# 条件查询：工资大于12000的技术部员工
cursor.execute('''
SELECT * FROM employees
WHERE department = '技术部' AND salary > 12000
''')
results = cursor.fetchall()

print("\n工资 > 12000 的技术部员工：")
for row in results:
    print(f"  {row[1]} - {row[4]}元")

python

复制

# 排序查询：按工资降序
cursor.execute('''
SELECT name, department, salary FROM employees
ORDER BY salary DESC
''')
sorted_rows = cursor.fetchall()

print("\n按工资排序（高→低）：")
for name, dept, sal in sorted_rows:
    print(f"  {name} | {dept} | {sal}元")

2.5 更新和删除

python

复制

# 更新数据：给所有技术部员工加薪10%
cursor.execute('''
UPDATE employees
SET salary = salary * 1.1
WHERE department = '技术部'
''')
conn.commit()
print(f"更新了 {cursor.rowcount} 条记录")

# 删除数据：删除薪资低于10000的员工
cursor.execute('''
DELETE FROM employees
WHERE salary < 10000
''')
conn.commit()
print(f"删除了 {cursor.rowcount} 条记录")

2.6 聚合函数

python

复制

# 统计各部门人数和平均工资
cursor.execute('''
SELECT department,
       COUNT(*) as headcount,
       AVG(salary) as avg_salary
FROM employees
GROUP BY department
ORDER BY avg_salary DESC
''')

print("\n各部门统计：")
print("-" * 40)
print(f"{'部门':<10} {'人数':<8} {'平均工资':<12}")
print("-" * 40)
for dept, count, avg_sal in cursor.fetchall():
    print(f"{dept:<10} {count:<8} {avg_sal:<12.2f}")

---

三、用pandas操作SQLite

直接用SQL很灵活，但处理数据还是pandas更方便。好消息是，两者可以无缝结合。

python

复制

import pandas as pd

# 读取数据库为DataFrame
df = pd.read_sql_query('SELECT * FROM employees', conn)

print("数据库中的数据：")
print(df)

# DataFrame写入数据库
new_employees = pd.DataFrame({
    'name': ['孙八', '周九'],
    'age': [31, 27],
    'department': ['市场部', '技术部'],
    'salary': [13500, 12500],
    'hire_date': ['2023-02-15', '2023-08-20']
})

new_employees.to_sql('employees', conn, if_exists='append', index=False)
print("\n新增2条数据！")

# 用SQL查询，pandas呈现
df_filtered = pd.read_sql_query('''
SELECT department, AVG(salary) as avg_salary
FROM employees
GROUP BY department
''', conn)
print("\n各部门平均工资：")
print(df_filtered)

---

四、连接MySQL

MySQL是企业级数据库，市场占有率极高。下面演示完整连接流程。

4.1 安装驱动

bash

复制

pip install pymysql

4.2 连接MySQL

python

复制

import pymysql

# 连接MySQL数据库
conn = pymysql.connect(
    host='localhost',          # 数据库地址
    port=3306,                 # 端口号
    user='root',               # 用户名
    password='your_password',  # 密码
    database='test_db',        # 数据库名
    charset='utf8mb4'          # 字符集（支持中文）
)

cursor = conn.cursor()
print("MySQL连接成功！")

4.3 CRUD操作

python

复制

# 创建表
cursor.execute('''
CREATE TABLE IF NOT EXISTS products (
    id INT PRIMARY KEY AUTO_INCREMENT,
    name VARCHAR(100) NOT NULL,
    category VARCHAR(50),
    price DECIMAL(10, 2),
    stock INT DEFAULT 0
)
''')
conn.commit()

# 插入数据
products = [
    ('iPhone 15', '手机', 6999.00, 100),
    ('MacBook Pro', '电脑', 12999.00, 50),
    ('AirPods Pro', '耳机', 1899.00, 200),
    ('iPad Air', '平板', 4599.00, 80),
]

cursor.executemany('''
INSERT INTO products (name, category, price, stock)
VALUES (%s, %s, %s, %s)
''', products)
conn.commit()
print(f"插入 {cursor.rowcount} 条商品记录")

# 查询
cursor.execute('SELECT * FROM products')
for row in cursor.fetchall():
    print(f"{row[1]} - {row[3]}元 - 库存{row[4]}件")

4.4 关闭连接

python

复制

# 操作完成后务必关闭
cursor.close()
conn.close()
print("连接已关闭")

---

五、用SQLAlchemy统一操作

不同数据库有不同的驱动和语法，写一套代码换数据库就要改一堆东西。

SQLAlchemy解决了这个问题——一套代码，切换任意数据库。

5.1 安装

bash

复制

pip install sqlalchemy pymysql

5.2 连接不同数据库

python

复制

from sqlalchemy import create_engine, text

# SQLite（最简单）
engine_sqlite = create_engine('sqlite:///myapp.db')

# MySQL（使用pymysql驱动）
engine_mysql = create_engine(
    'mysql+pymysql://root:password@localhost:3306/test_db'
)

# PostgreSQL
engine_pg = create_engine(
    'postgresql://user:password@localhost:5432/test_db'
)

# 查看当前连接的数据库类型
print(f"当前数据库：{engine_sqlite.url.drivername}")

5.3 用SQLAlchemy执行SQL

python

复制

from sqlalchemy import text

# 创建引擎
engine = create_engine('sqlite:///company.db')

# 执行原生SQL
with engine.connect() as conn:
    # 创建表
    conn.execute(text('''
    CREATE TABLE IF NOT EXISTS departments (
        id INTEGER PRIMARY KEY,
        name TEXT NOT NULL,
        location TEXT
    )
    '''))
    conn.commit()

    # 插入数据
    conn.execute(text('''
    INSERT INTO departments (id, name, location)
    VALUES (:id, :name, :loc)
    '''), [
        {'id': 1, 'name': '技术部', 'loc': '北京'},
        {'id': 2, 'name': '市场部', 'loc': '上海'},
        {'id': 3, 'name': '销售部', 'loc': '广州'},
    ])
    conn.commit()

    # 查询
    result = conn.execute(text('SELECT * FROM departments'))
    print("\n部门列表：")
    for row in result:
        print(f"  {row}")

5.4 用ORM方式操作数据库

ORM（对象关系映射）让你用操作对象的方式操作数据库，不用写SQL。

python

复制

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, declarative_base

Base = declarative_base()

class Employee(Base):
    __tablename__ = 'employees_orm'

    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    department = Column(String(50))
    salary = Column(Integer)

    def __repr__(self):
        return f"<Employee {self.name} - {self.department}>"

# 创建引擎和会话
engine = create_engine('sqlite:///orm_demo.db', echo=False)
Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

# 插入数据
emp1 = Employee(name='Tom', department='Engineering', salary=15000)
emp2 = Employee(name='Lucy', department='Marketing', salary=12000)
emp3 = Employee(name='Jack', department='Engineering', salary=14000)

session.add_all([emp1, emp2, emp3])
session.commit()
print("ORM插入成功！")

# 查询（不用写SQL）
eng_employees = session.query(Employee).filter_by(department='Engineering').all()
print("\n工程部员工：")
for emp in eng_employees:
    print(f"  {emp.name} - {emp.salary}元")

# 更新
emp1.salary = 16000
session.commit()
print("\nTom涨薪后：")
print(session.get(Employee, emp1.id))

# 删除
session.delete(emp2)
session.commit()

# 关闭会话
session.close()
print("\nORM操作完成！")

---

六、实战：爬虫数据存储到数据库

结合前面几篇学过的爬虫知识，把爬到的数据直接存进数据库，形成完整的数据采集→存储→分析管线。

python

复制

import sqlite3
import requests
from bs4 import BeautifulSoup

def crawl_books_to_db():
    """爬取豆瓣读书TOP250，存入SQLite数据库"""

    # 连接数据库
    conn = sqlite3.connect('books.db')
    cursor = conn.cursor()

    # 创建表
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS books (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        title TEXT NOT NULL,
        rating REAL,
        author TEXT,
        publisher TEXT,
        publish_year TEXT
    )
    ''')
    conn.commit()

    base_url = 'https://book.douban.com/top250?start='
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
    }

    print("开始爬取豆瓣读书TOP250...")

    for page in range(0, 250, 25):
        url = base_url + str(page)
        try:
            response = requests.get(url, headers=headers, timeout=10)
            soup = BeautifulSoup(response.text, 'html.parser')

            for item in soup.select('tr.item'):
                title = item.select_one('td:nth-child(2) a').get_text(strip=True)
                rating_tag = item.select_one('span.rating_nums')
                rating = float(rating_tag.get_text()) if rating_tag else 0

                info = item.select_one('td:nth-child(2) p').get_text(strip=True)
                parts = info.split('/')
                author = parts[0] if len(parts) > 0 else ''
                publisher_year = '/'.join(parts[-2:]) if len(parts) > 2 else ''

                cursor.execute('''
                INSERT INTO books (title, rating, author, publisher)
                VALUES (?, ?, ?, ?)
                ''', (title, rating, author, publisher_year))

            print(f"  第{page // 25 + 1}页完成...")
        except Exception as e:
            print(f"  第{page // 25 + 1}页出错：{e}")

    conn.commit()

    # 统计
    cursor.execute('SELECT COUNT(*), AVG(rating) FROM books')
    count, avg_rating = cursor.fetchone()
    print(f"\n爬取完成！共{count}本书，平均评分{avg_rating:.2f}")

    # 关闭连接
    cursor.close()
    conn.close()

# 运行爬虫
# crawl_books_to_db()

python

复制

# 查询爬取结果
conn = sqlite3.connect('books.db')
cursor = conn.cursor()

print("\n评分最高的10本书：")
cursor.execute('''
SELECT title, rating, author
FROM books
ORDER BY rating DESC
LIMIT 10
''')

for i, (title, rating, author) in enumerate(cursor.fetchall(), 1):
    print(f"{i}. {title}")
    print(f"   评分：{rating} | 作者：{author}")
    print()

conn.close()

---

七、数据库最佳实践

7.1 使用上下文管理器

每次操作后手动关闭连接容易遗漏，用上下文管理器自动管理：

python

复制

import sqlite3

# 自动管理连接和游标
with sqlite3.connect('demo.db') as conn:
    cursor = conn.cursor()
    cursor.execute('SELECT * FROM employees')
    results = cursor.fetchall()
    # 离开with块时自动commit和close

7.2 防止SQL注入

永远不要用字符串拼接构建SQL语句：

python

复制

# ❌ 危险！会被SQL注入攻击
user_input = "' OR '1'='1"
cursor.execute(f"SELECT * FROM users WHERE name = '{user_input}'")

# ✅ 安全！用参数化查询
cursor.execute("SELECT * FROM users WHERE name = ?", (user_input,))

7.3 事务处理

多个操作需要同时成功或同时失败时，用事务：

python

复制

try:
    cursor.execute("UPDATE accounts SET balance = balance - 1000 WHERE id = 1")
    cursor.execute("UPDATE accounts SET balance = balance + 1000 WHERE id = 2")
    conn.commit()  # 两笔转账同时成功
    print("转账成功！")
except Exception as e:
    conn.rollback()  # 任何一步出错就全部回滚
    print(f"转账失败：{e}")

7.4 索引加速查询

对于大表，添加索引能大幅加速查询：

python

复制

# 给经常查询的字段加索引
cursor.execute('CREATE INDEX IF NOT EXISTS idx_department ON employees(department)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_salary ON employees(salary)')
conn.commit()
print("索引创建成功！")

---

总结

今天这篇文章，涵盖了Python操作数据库的核心技能：

工具	适用场景	复杂度
SQLite	个人项目、小型应用、内嵌数据库	⭐ 最简单
PyMySQL	企业级MySQL连接	⭐⭐
SQLAlchemy	多数据库切换、ORM开发	⭐⭐⭐
ORM	大型项目、面向对象开发	⭐⭐⭐

整套流程就三步：连接 → 操作 → 关闭。

结合前面学过的爬虫和数据分析，你现在拥有了一条完整的数据管线：采集（爬虫）→ 存储（数据库）→ 分析（pandas/机器学习）。

下一步可以深入学习：数据库优化（索引、分表）、多表关联查询（JOIN）、数据库备份恢复等进阶技能。

---

👍 如果这篇文章对你有帮助，记得点赞！ 在评论区说说，你项目中用的是什么数据库？

我是AI实验室2026，专注AI大模型与Python技术，我们下篇见！