数据库操作 - Python高级课程

数据库操作简介

数据库操作是现代应用程序开发中不可或缺的一部分。Python提供了丰富的数据库操作工具和库，使得开发者能够方便地与各种类型的数据库进行交互。

为什么要学习数据库操作？

实现数据的持久化存储
高效管理大量结构化数据
保证数据的一致性和安全性
支持复杂的数据查询和分析

常见数据库类型

关系型数据库：SQLite、MySQL、PostgreSQL
文档型数据库：MongoDB
键值存储：Redis
列式数据库：Cassandra

SQLite数据库

SQLite是一个轻量级的关系型数据库，它不需要独立的服务器进程，适合嵌入式应用和小型项目。Python标准库内置了sqlite3模块。

基本操作

import sqlite3

# 连接数据库（如果不存在则创建）
conn = sqlite3.connect('example.db')
cursor = conn.cursor()

# 创建表
cursor.execute('''
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        age INTEGER,
        email TEXT UNIQUE
    )
''')

# 插入数据
def insert_user(name, age, email):
    cursor.execute('''
        INSERT INTO users (name, age, email)
        VALUES (?, ?, ?)
    ''', (name, age, email))
    conn.commit()

# 查询数据
def get_users():
    cursor.execute('SELECT * FROM users')
    return cursor.fetchall()

# 使用示例
insert_user('张三', 25, '[email protected]')
users = get_users()
print(users)

# 关闭连接
conn.close()

SQLite特点

轻量级，零配置
支持事务和ACID特性
适合单文件应用
并发访问限制

MySQL数据库

MySQL是最流行的开源关系型数据库之一，适合各种规模的应用。Python可以通过mysql-connector-python等库与MySQL交互。

连接与基本操作

import mysql.connector

# 建立连接
def get_connection():
    return mysql.connector.connect(
        host="localhost",
        user="your_username",
        password="your_password",
        database="your_database"
    )

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

# 插入数据
def add_product(conn, name, price, stock):
    cursor = conn.cursor()
    sql = "INSERT INTO products (name, price, stock) VALUES (%s, %s, %s)"
    cursor.execute(sql, (name, price, stock))
    conn.commit()
    return cursor.lastrowid

# 查询数据
def get_products(conn):
    cursor = conn.cursor(dictionary=True)
    cursor.execute("SELECT * FROM products")
    return cursor.fetchall()

# 使用事务
def transfer_stock(conn, from_id, to_id, amount):
    cursor = conn.cursor()
    try:
        cursor.execute("START TRANSACTION")
        
        # 减少源产品库存
        cursor.execute("""
            UPDATE products 
            SET stock = stock - %s 
            WHERE id = %s AND stock >= %s
        """, (amount, from_id, amount))
        
        if cursor.rowcount == 0:
            raise Exception("库存不足")
        
        # 增加目标产品库存
        cursor.execute("""
            UPDATE products 
            SET stock = stock + %s 
            WHERE id = %s
        """, (amount, to_id))
        
        conn.commit()
        return True
    except Exception as e:
        conn.rollback()
        print(f"转移库存失败：{e}")
        return False

注意事项

始终使用参数化查询防止SQL注入
及时关闭数据库连接
正确处理事务
注意连接池的使用

ORM框架

ORM（对象关系映射）框架可以让开发者用面向对象的方式操作数据库，而不需要直接写SQL语句。SQLAlchemy是Python最流行的ORM框架之一。

SQLAlchemy基础

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

# 创建基类
Base = declarative_base()

# 定义模型
class Product(Base):
    __tablename__ = 'products'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(255), nullable=False)
    price = Column(Float)
    stock = Column(Integer, default=0)
    
    def __repr__(self):
        return f""

# 创建数据库引擎
engine = create_engine('sqlite:///products.db')
Base.metadata.create_all(engine)

# 创建会话
Session = sessionmaker(bind=engine)
session = Session()

# 添加数据
def add_product(name, price, stock):
    product = Product(name=name, price=price, stock=stock)
    session.add(product)
    session.commit()
    return product

# 查询数据
def get_products():
    return session.query(Product).all()

# 更新数据
def update_product_price(product_id, new_price):
    product = session.query(Product).get(product_id)
    if product:
        product.price = new_price
        session.commit()
        return True
    return False

# 删除数据
def delete_product(product_id):
    product = session.query(Product).get(product_id)
    if product:
        session.delete(product)
        session.commit()
        return True
    return False

ORM的优势

更直观的数据操作方式
自动处理数据库差异
内置防SQL注入机制
支持数据库迁移

NoSQL数据库

NoSQL数据库提供了一种更灵活的数据存储方式，适合处理非结构化数据和大规模数据。这里以MongoDB为例介绍NoSQL数据库的使用。

MongoDB操作

from pymongo import MongoClient

# 连接MongoDB
client = MongoClient('mongodb://localhost:27017/')
db = client['example_db']
collection = db['users']

# 插入文档
def insert_user(user_data):
    return collection.insert_one(user_data)

# 查询文档
def find_users(query=None):
    return list(collection.find(query or {}))

# 更新文档
def update_user(user_id, update_data):
    return collection.update_one(
        {'_id': user_id},
        {'$set': update_data}
    )

# 删除文档
def delete_user(user_id):
    return collection.delete_one({'_id': user_id})

# 使用示例
user = {
    'name': '张三',
    'age': 25,
    'skills': ['Python', 'MongoDB'],
    'contact': {
        'email': '[email protected]',
        'phone': '1234567890'
    }
}

# 插入用户
result = insert_user(user)
print(f"插入的文档ID: {result.inserted_id}")

# 查询用户
users = find_users({'age': {'$gt': 20}})
for user in users:
    print(user)

# 更新用户
update_user(result.inserted_id, {
    'age': 26,
    'skills': ['Python', 'MongoDB', 'Redis']
})

NoSQL特点

灵活的数据模型
高度可扩展性
适合大数据应用
支持复杂的查询操作

最佳实践

数据库操作建议

使用连接池管理数据库连接
正确处理事务和异常
使用参数化查询防止SQL注入
定期备份重要数据
优化查询性能
合理设计数据库结构

连接池示例

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import QueuePool

engine = create_engine('mysql+mysqlconnector://user:pass@localhost/db',
                      poolclass=QueuePool,
                      pool_size=5,
                      max_overflow=10,
                      pool_timeout=30)

Session = sessionmaker(bind=engine)

def get_session():
    return Session()

查询优化示例

from sqlalchemy import func

# 使用索引
def get_user_by_email(email):
    return session.query(User).filter(
        User.email == email
    ).first()

# 选择需要的列
def get_user_names():
    return session.query(User.name).all()

# 使用聚合函数
def get_user_stats():
    return session.query(
        func.count(User.id).label('total_users'),
        func.avg(User.age).label('avg_age')
    ).first()

# 批量操作
def bulk_insert_users(users):
    session.bulk_insert_mappings(User, users)
    session.commit()

练习与实践

通过以下练习，你可以加深对数据库操作的理解，并掌握在实际场景中应用数据库技术的技巧。

学习目标

掌握基本的SQL操作
理解ORM的使用方法
能够处理复杂的数据关系
学会数据库性能优化

练习1：图书管理系统

实现一个简单的图书管理系统，包含图书、作者、分类等基本信息的管理。

提示：

设计合适的数据库结构
实现基本的CRUD操作
处理一对多、多对多关系
添加搜索和过滤功能

练习2：电商数据库

设计并实现一个简单的电商系统数据库，包含商品、用户、订单等核心功能。

提示：

实现商品库存管理
处理订单状态流转
实现购物车功能
添加用户评价系统

练习3：博客系统

使用ORM框架实现一个博客系统的数据层，支持文章、评论、标签等功能。

提示：

使用SQLAlchemy设计模型
实现文章分类和标签
添加用户评论功能
实现文章搜索

练习4：数据迁移工具

开发一个数据迁移工具，支持不同数据库之间的数据转换和迁移。

提示：

支持多种数据库类型
处理数据类型转换
实现断点续传
添加数据验证

练习5：缓存系统

实现一个数据库缓存系统，使用Redis作为缓存层，提高查询性能。

提示：

实现缓存策略
处理缓存失效
添加并发控制
监控缓存效果

挑战任务

完成基础练习后，你可以尝试以下挑战任务来提升你的数据库开发技能：

实现一个简单的ORM框架
开发数据库监控工具
实现分布式数据库方案
设计数据库分片策略