第29篇：Python开发进阶：数据库操作与ORM

第29篇：数据库操作与ORM

目录

1. 数据库操作概述
   • 什么是数据库
   • 数据库管理系统（DBMS）
   • 常见的数据库类型
2. Python中的数据库操作
   • 使用sqlite3模块
   • 连接到MySQL数据库
   • 连接到PostgreSQL数据库
3. SQL基础
   • 数据库基本操作
   • 表的创建与管理
   • 数据的插入、查询、更新和删除
4. ORM（对象关系映射）
   • 什么是ORM
   • ORM的优势
   • 常见的Python ORM框架
5. 使用SQLAlchemy进行ORM操作
   • SQLAlchemy简介
   • 安装与设置
   • 定义模型
   • 数据库迁移
   • CRUD操作
6. Django ORM
   • Django的ORM简介
   • 模型的定义与迁移
   • 查询操作
7. 示例代码
   • 使用sqlite3操作数据库
   • 使用SQLAlchemy进行ORM操作
   • 使用Django ORM管理数据库
8. 常见问题及解决方法
   • 问题1：连接数据库失败
   • 问题2：查询效率低
   • 问题3：数据库迁移问题
   • 问题4：ORM性能问题
9. 总结

数据库操作概述

什么是数据库

数据库是按照一定的数据模型组织起来的、可以长期存储并管理数据的集合。它允许用户高效地存储、查询、更新和管理数据，广泛应用于各种应用程序和系统中，如网站、企业管理系统、移动应用等。

数据库管理系统（DBMS）

**数据库管理系统（DBMS）**是用于创建和管理数据库的软件工具。它提供了数据的存储、查询、更新和管理功能，确保数据的一致性、完整性和安全性。常见的DBMS包括：

• 关系型数据库管理系统（RDBMS）：如MySQL、PostgreSQL、SQLite、Oracle、SQL Server等。
• 非关系型数据库管理系统（NoSQL DBMS）：如MongoDB、Redis、Cassandra、CouchDB等。

常见的数据库类型

Python中的数据库操作

Python提供了多种方式与不同类型的数据库进行交互。以下将介绍使用内置的sqlite3模块以及连接MySQL和PostgreSQL数据库的方法。

使用sqlite3模块

SQLite是一个轻量级的、嵌入式的关系型数据库，Python内置了sqlite3模块，支持SQLite数据库的操作。适用于小型应用、测试和开发环境。

示例：使用sqlite3创建和操作数据库

import sqlite3

# 连接到SQLite数据库（如果数据库不存在，会自动创建）
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
)
''')

# 插入数据
cursor.execute('INSERT INTO users (name, age) VALUES (?, ?)', ('Alice', 30))
cursor.execute('INSERT INTO users (name, age) VALUES (?, ?)', ('Bob', 25))

# 提交事务
conn.commit()

# 查询数据
cursor.execute('SELECT * FROM users')
rows = cursor.fetchall()
for row in rows:
    print(row)

# 关闭连接
conn.close()

输出：

(1, 'Alice', 30)
(2, 'Bob', 25)

连接到MySQL数据库

要连接到MySQL数据库，可以使用第三方库PyMySQL或mysql-connector-python。以下示例使用PyMySQL。

安装PyMySQL

pip install PyMySQL

示例：使用PyMySQL连接MySQL

import pymysql

# 连接到MySQL数据库
connection = pymysql.connect(
    host='localhost',
    user='your_username',
    password='your_password',
    database='your_database',
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor
)

try:
    with connection.cursor() as cursor:
        # 创建表
        sql = '''
        CREATE TABLE IF NOT EXISTS employees (
            id INT AUTO_INCREMENT PRIMARY KEY,
            name VARCHAR(255) NOT NULL,
            position VARCHAR(255),
            salary DECIMAL(10, 2)
        )
        '''
        cursor.execute(sql)

        # 插入数据
        sql = 'INSERT INTO employees (name, position, salary) VALUES (%s, %s, %s)'
        cursor.execute(sql, ('Charlie', 'Engineer', 70000))
        cursor.execute(sql, ('Diana', 'Manager', 90000))

    # 提交事务
    connection.commit()

    with connection.cursor() as cursor:
        # 查询数据
        sql = 'SELECT * FROM employees'
        cursor.execute(sql)
        result = cursor.fetchall()
        for row in result:
            print(row)
finally:
    connection.close()

输出：

{'id': 1, 'name': 'Charlie', 'position': 'Engineer', 'salary': Decimal('70000.00')}
{'id': 2, 'name': 'Diana', 'position': 'Manager', 'salary': Decimal('90000.00')}

连接到PostgreSQL数据库

要连接到PostgreSQL数据库，可以使用第三方库psycopg2。

安装psycopg2

pip install psycopg2

示例：使用psycopg2连接PostgreSQL

import psycopg2
from psycopg2 import sql

# 连接到PostgreSQL数据库
connection = psycopg2.connect(
    host='localhost',
    user='your_username',
    password='your_password',
    dbname='your_database'
)

try:
    with connection.cursor() as cursor:
        # 创建表
        cursor.execute('''
        CREATE TABLE IF NOT EXISTS products (
            id SERIAL PRIMARY KEY,
            name VARCHAR(255) NOT NULL,
            price NUMERIC(10, 2)
        )
        ''')

        # 插入数据
        cursor.execute('INSERT INTO products (name, price) VALUES (%s, %s)', ('Laptop', 1200.50))
        cursor.execute('INSERT INTO products (name, price) VALUES (%s, %s)', ('Smartphone', 800.00))

    # 提交事务
    connection.commit()

    with connection.cursor() as cursor:
        # 查询数据
        cursor.execute('SELECT * FROM products')
        result = cursor.fetchall()
        for row in result:
            print(row)
finally:
    connection.close()

输出：

(1, 'Laptop', Decimal('1200.50'))
(2, 'Smartphone', Decimal('800.00'))

SQL基础

数据库基本操作

**SQL（结构化查询语言）**是用于管理和操作关系型数据库的标准语言。通过SQL，可以执行数据的创建、查询、更新和删除等操作。

表的创建与管理

创建表

CREATE TABLE IF NOT EXISTS students (
    id SERIAL PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    grade INTEGER
);

删除表

DROP TABLE IF EXISTS students;

修改表

-- 添加列
ALTER TABLE students ADD COLUMN email VARCHAR(100);

-- 删除列
ALTER TABLE students DROP COLUMN email;

-- 修改列类型
ALTER TABLE students ALTER COLUMN grade TYPE SMALLINT;

数据的插入、查询、更新和删除

插入数据

INSERT INTO students (name, grade) VALUES ('Emma', 85);
INSERT INTO students (name, grade) VALUES ('Liam', 90);

查询数据

-- 查询所有数据
SELECT * FROM students;

-- 条件查询
SELECT * FROM students WHERE grade > 80;

-- 排序查询
SELECT * FROM students ORDER BY grade DESC;

-- 聚合查询
SELECT AVG(grade) AS average_grade FROM students;

更新数据

UPDATE students SET grade = 95 WHERE name = 'Emma';

删除数据

DELETE FROM students WHERE name = 'Liam';

ORM（对象关系映射）

什么是ORM

**对象关系映射（ORM，Object-Relational Mapping）**是一种编程技术，将对象编程语言中的对象与关系型数据库中的表进行映射。通过ORM，开发者可以使用面向对象的方式操作数据库，而无需编写原始的SQL语句。

ORM的优势

• 提高开发效率：减少手写SQL，提高代码可读性和可维护性。
• 跨数据库兼容性：通过ORM层，可以轻松切换不同的数据库后端。
• 自动化管理：ORM框架通常提供自动化的迁移、验证和查询优化功能。
• 安全性：ORM框架能够自动处理SQL注入等安全问题。

常见的Python ORM框架

使用SQLAlchemy进行ORM操作

SQLAlchemy简介

SQLAlchemy是一个功能强大且灵活的Python ORM框架，支持多种数据库后端。它提供了两种主要的API：

• 核心（Core）：低级别的数据库操作，提供SQL表达式语言。
• ORM：高级别的对象关系映射，支持自动生成SQL语句。

安装与设置

使用pip安装SQLAlchemy：

pip install SQLAlchemy

定义模型

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

Base = declarative_base()

class Product(Base):
    __tablename__ = 'products'

    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    price = Column(Float)

    def __repr__(self):
        return f"<Product(name='{self.name}', price={self.price})>"

数据库迁移

在SQLAlchemy中，可以手动创建数据库表，也可以使用迁移工具如Alembic进行自动化迁移。

手动创建表

from sqlalchemy import create_engine

# 创建引擎
engine = create_engine('sqlite:///products.db', echo=True)

# 创建所有表
Base.metadata.create_all(engine)

CRUD操作

创建会话

from sqlalchemy.orm import sessionmaker

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

# 创建会话实例
session = Session()

插入数据

# 创建新产品
new_product = Product(name='Laptop', price=1200.50)

# 添加到会话
session.add(new_product)

# 提交事务
session.commit()

查询数据

# 查询所有产品
products = session.query(Product).all()
for product in products:
    print(product)

# 条件查询
expensive_products = session.query(Product).filter(Product.price > 1000).all()
for product in expensive_products:
    print(product)

更新数据

# 查询特定产品
product = session.query(Product).filter_by(name='Laptop').first()
if product:
    product.price = 1100.00
    session.commit()

删除数据

# 查询特定产品
product = session.query(Product).filter_by(name='Laptop').first()
if product:
    session.delete(product)
    session.commit()

Django ORM

Django的ORM简介

Django ORM是Django Web框架内置的ORM系统，提供了与数据库交互的高级抽象层。它允许开发者使用Python类定义数据库模型，自动生成SQL语句，简化数据库操作。

模型的定义与迁移

定义模型

在Django应用的models.py中定义模型：

from django.db import models

class Article(models.Model):
    title = models.CharField(max_length=200)
    content = models.TextField()
    published_date = models.DateTimeField(auto_now_add=True)

    def __str__(self):
        return self.title

进行数据库迁移

1. 创建迁移文件

   python manage.py makemigrations
   

2. 应用迁移

   python manage.py migrate
   

查询操作

from blog.models import Article

# 创建文章
article = Article.objects.create(title='Django ORM', content='内容简介')

# 查询所有文章
all_articles = Article.objects.all()
for article in all_articles:
    print(article)

# 条件查询
django_articles = Article.objects.filter(title__icontains='Django')
for article in django_articles:
    print(article)

# 更新文章
article = Article.objects.get(id=1)
article.title = '更新后的标题'
article.save()

# 删除文章
article = Article.objects.get(id=1)
article.delete()

示例代码

使用sqlite3操作数据库

import sqlite3

def sqlite_example():
    # 连接到SQLite数据库
    conn = sqlite3.connect('students.db')
    cursor = conn.cursor()

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

    # 插入数据
    cursor.execute('INSERT INTO students (name, age) VALUES (?, ?)', ('Emma', 22))
    cursor.execute('INSERT INTO students (name, age) VALUES (?, ?)', ('Liam', 24))
    conn.commit()

    # 查询数据
    cursor.execute('SELECT * FROM students')
    rows = cursor.fetchall()
    for row in rows:
        print(row)

    # 关闭连接
    conn.close()

if __name__ == '__main__':
    sqlite_example()

运行结果：

(1, 'Emma', 22)
(2, 'Liam', 24)

使用SQLAlchemy进行ORM操作

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 Employee(Base):
    __tablename__ = 'employees'

    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    position = Column(String(100))
    salary = Column(Float)

    def __repr__(self):
        return f"<Employee(name='{self.name}', position='{self.position}', salary={self.salary})>"

def sqlalchemy_example():
    # 创建引擎
    engine = create_engine('sqlite:///employees.db', echo=True)

    # 创建所有表
    Base.metadata.create_all(engine)

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

    # 插入数据
    emp1 = Employee(name='Alice', position='Engineer', salary=75000)
    emp2 = Employee(name='Bob', position='Manager', salary=85000)
    session.add_all([emp1, emp2])
    session.commit()

    # 查询数据
    employees = session.query(Employee).all()
    for emp in employees:
        print(emp)

    # 更新数据
    emp = session.query(Employee).filter_by(name='Alice').first()
    if emp:
        emp.salary = 80000
        session.commit()

    # 删除数据
    emp = session.query(Employee).filter_by(name='Bob').first()
    if emp:
        session.delete(emp)
        session.commit()

    # 关闭会话
    session.close()

if __name__ == '__main__':
    sqlalchemy_example()

运行结果：

<Employee(name='Alice', position='Engineer', salary=75000.0)>
<Employee(name='Bob', position='Manager', salary=85000.0)>
<Employee(name='Alice', position='Engineer', salary=80000.0)>

使用Django ORM管理数据库

设置Django项目

1. 创建项目和应用

   django-admin startproject mysite
   cd mysite
   python manage.py startapp blog
   

2. 定义模型

   在blog/models.py中：

   from django.db import models
   
   class Post(models.Model):
       title = models.CharField(max_length=200)
       content = models.TextField()
       published_date = models.DateTimeField(auto_now_add=True)
   
       def __str__(self):
           return self.title
   

3. 注册应用和模型

   • 在mysite/settings.py中，将blog添加到INSTALLED_APPS：

     INSTALLED_APPS = [
         # ...
         'blog',
     ]
     

   • 进行迁移：

     python manage.py makemigrations
     python manage.py migrate
     

4. 使用Django Shell进行操作

   python manage.py shell
   

   在Django Shell中：

   from blog.models import Post
   
   # 创建文章
   post1 = Post.objects.create(title='Django ORM', content='Django的ORM功能强大。')
   post2 = Post.objects.create(title='SQLAlchemy', content='SQLAlchemy是一个灵活的ORM框架。')
   
   # 查询所有文章
   posts = Post.objects.all()
   for post in posts:
       print(post)
   
   # 更新文章
   post = Post.objects.get(title='Django ORM')
   post.content = 'Django ORM提供了高级的数据库操作功能。'
   post.save()
   
   # 删除文章
   post = Post.objects.get(title='SQLAlchemy')
   post.delete()
   

   输出：

   Django ORM
   SQLAlchemy
   

常见问题及解决方法

问题1：连接数据库失败

原因：数据库服务未启动、连接参数错误（如主机名、端口、用户名、密码）、网络问题等。

解决方法：

1. 检查数据库服务是否运行：确保数据库服务器正在运行。
2. 验证连接参数：确认主机名、端口、用户名、密码和数据库名称是否正确。
3. 检查网络连接：确保客户端与数据库服务器之间的网络连接正常。
4. 查看错误日志：检查数据库服务器的错误日志，获取详细的错误信息。
5. 防火墙设置：确保防火墙未阻止数据库端口的访问。

问题2：查询效率低

原因：缺乏索引、查询语句不优化、数据量过大等。

解决方法：

1. 创建索引：为经常查询的列创建索引，提升查询速度。

   CREATE INDEX idx_name ON employees (name);
   

2. 优化查询语句：避免使用SELECT *，仅选择必要的列。使用JOIN代替子查询。
3. 分页查询：对于大量数据，使用分页技术分批查询，减少单次查询的数据量。

   SELECT * FROM employees LIMIT 100 OFFSET 200;
   

4. 使用缓存：对于频繁访问的数据，使用缓存机制（如Redis）减少数据库负载。
5. 数据库优化：定期进行数据库维护，如碎片整理、统计信息更新等。

问题3：数据库迁移问题

原因：模型更改未正确迁移、迁移文件冲突、数据库结构不一致等。

解决方法：

1. 正确创建和应用迁移：

   python manage.py makemigrations
   python manage.py migrate
   

2. 检查迁移文件：确保迁移文件按顺序正确生成，避免手动修改迁移文件。
3. 解决迁移冲突：如果多个开发者同时修改模型，可能导致迁移文件冲突。使用--merge选项合并迁移。

   python manage.py makemigrations --merge
   

4. 备份数据库：在进行重大迁移前，备份数据库以防数据丢失。
5. 使用迁移工具：对于复杂迁移，使用专业工具如Alembic（针对SQLAlchemy）进行管理。

问题4：ORM性能问题

原因：过多的数据库查询、未使用批量操作、N+1查询问题等。

解决方法：

1. 使用批量操作：尽量使用批量插入、更新和删除，减少数据库交互次数。

   # SQLAlchemy示例
   session.bulk_save_objects([obj1, obj2, obj3])
   session.commit()
   

2. 避免N+1查询：使用预加载（joinedload或subqueryload）减少查询次数。

   from sqlalchemy.orm import joinedload
   session.query(User).options(joinedload(User.addresses)).all()
   

3. 优化查询逻辑：合并相似的查询，避免重复的数据获取。
4. 使用缓存：对于频繁读取的数据，使用缓存减少数据库访问。
5. 分析和调优：使用性能分析工具（如Django Debug Toolbar、SQLAlchemy Profiling）识别瓶颈并优化。

总结

在本篇文章中，我们深入探讨了Python中的数据库操作与ORM。通过理解数据库的基本概念、学习如何使用sqlite3、MySQL和PostgreSQL进行数据库操作，掌握SQL基础知识，以及学习如何使用SQLAlchemy和Django ORM进行对象关系映射，您可以高效地进行数据管理和操作。数据库操作与ORM是构建数据驱动应用的核心技能，掌握这些内容将大大提升您的开发能力和项目效率。

学习建议：

1. 实践数据库操作：通过创建和管理不同类型的数据库，巩固数据库连接和基本操作知识。
2. 深入学习SQL：掌握复杂查询、事务管理、索引优化等高级SQL技术，提升数据库操作能力。
3. 掌握ORM框架：选择SQLAlchemy或Django ORM，深入学习其高级特性和最佳实践，编写高效的数据库操作代码。
4. 学习数据库设计：了解数据库范式、关系建模和规范化设计，构建高效且可维护的数据库结构。
5. 探索数据库性能优化：学习索引优化、查询优化、缓存机制等技术，提升数据库性能。
6. 阅读相关书籍和文档：如《SQLAlchemy文档》、《Django官方文档》，系统性地提升数据库操作与ORM能力。

接下来的系列文章将继续深入探讨Python的网络安全与测试，帮助您进一步掌握Python编程的核心概念和技巧。保持学习的热情，持续实践，您将逐步成为一名优秀的Python开发者！

如果您有任何问题或需要进一步的帮助，请随时在评论区留言或联系相关技术社区。