跳到主要内容

Python 数据库连接

在Web开发中,数据库连接是一项基础且关键的技能。Python提供了多种方式来连接和操作各类数据库,让我们能够高效地存储、检索和管理数据。本文将介绍如何使用Python连接不同类型的数据库,并执行基本的增删改查(CRUD)操作。

为什么需要学习数据库连接?

数据是任何应用程序的核心。无论是简单的待办事项应用还是复杂的社交媒体平台,它们都需要某种形式的数据存储。Python作为一种通用编程语言,提供了与各种数据库系统交互的强大工具和库。

常见的Python数据库连接方式

1. 使用SQLite

SQLite是一个轻量级的磁盘数据库,不需要单独的服务器进程。Python标准库包含了sqlite3模块,使其成为初学者的理想选择。

基本连接与操作

python
import sqlite3

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

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

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

# 插入数据
cursor.execute("INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
('张三', 'zhangsan@example.com', 25))

# 提交事务
conn.commit()

# 查询数据
cursor.execute("SELECT * FROM users")
users = cursor.fetchall()
print("所有用户:")
for user in users:
print(user)

# 关闭连接
conn.close()

输出:

所有用户:
(1, '张三', 'zhangsan@example.com', 25)
提示

SQLite非常适合以下场景:

  • 开发/测试环境
  • 小型应用
  • 嵌入式设备
  • 需要文件形式存储的应用

2. 连接MySQL

MySQL是最流行的开源关系型数据库之一。要在Python中使用MySQL,我们需要安装mysql-connector-python包。

首先安装必要的库:

bash
pip install mysql-connector-python

然后实现基本连接:

python
import mysql.connector

try:
# 建立连接
conn = mysql.connector.connect(
host="localhost",
user="yourusername",
password="yourpassword",
database="yourdatabase"
)

# 创建游标
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
)
''')

# 插入数据
sql = "INSERT INTO products (name, price, stock) VALUES (%s, %s, %s)"
values = ("笔记本电脑", 4999.99, 10)
cursor.execute(sql, values)
conn.commit()
print(f"已插入ID: {cursor.lastrowid}")

# 查询数据
cursor.execute("SELECT * FROM products")
products = cursor.fetchall()
print("\n所有产品:")
for product in products:
print(product)

except mysql.connector.Error as error:
print(f"连接失败: {error}")

finally:
# 关闭连接
if 'conn' in locals() and conn.is_connected():
cursor.close()
conn.close()
print("MySQL连接已关闭")

输出:

已插入ID: 1

所有产品:
(1, '笔记本电脑', Decimal('4999.99'), 10)
MySQL连接已关闭

3. 使用ORM: SQLAlchemy

对象关系映射(ORM)是一种将数据库表映射到对象的编程技术。SQLAlchemy是Python中最流行的ORM库之一。

安装SQLAlchemy:

bash
pip install sqlalchemy

基本用法:

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

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

# 创建基类
Base = declarative_base()

# 定义模型类
class Book(Base):
__tablename__ = 'books'

id = Column(Integer, primary_key=True)
title = Column(String, nullable=False)
author = Column(String)
price = Column(Float)

def __repr__(self):
return f"<Book(title='{self.title}', author='{self.author}', price={self.price})>"

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

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

# 添加数据
new_book = Book(title='Python编程', author='张三', price=59.9)
session.add(new_book)
session.commit()

# 查询数据
books = session.query(Book).all()
print("\n所有图书:")
for book in books:
print(book)

# 更新数据
book_to_update = session.query(Book).filter_by(title='Python编程').first()
if book_to_update:
book_to_update.price = 69.9
session.commit()
print(f"\n更新后的图书: {book_to_update}")

# 删除数据
# book_to_delete = session.query(Book).filter_by(title='Python编程').first()
# if book_to_delete:
# session.delete(book_to_delete)
# session.commit()
# print("\n删除成功")

# 关闭会话
session.close()

输出:

所有图书:
<Book(title='Python编程', author='张三', price=59.9)>

更新后的图书: <Book(title='Python编程', author='张三', price=69.9)>
备注

使用ORM的好处:

  • 减少样板代码
  • 提高代码可维护性
  • 更加面向对象
  • 更好的安全性(防止SQL注入)

4. 连接NoSQL数据库: MongoDB

MongoDB是一种流行的文档型NoSQL数据库。要在Python中使用MongoDB,我们需要安装pymongo库。

安装pymongo:

bash
pip install pymongo

基本用法:

python
from pymongo import MongoClient

# 建立连接
client = MongoClient('mongodb://localhost:27017/')

# 获取或创建数据库
db = client['blog_database']

# 获取或创建集合
posts = db.posts

# 插入文档
post_data = {
'title': 'Python与MongoDB的结合使用',
'content': '这是一篇关于Python如何连接MongoDB的文章',
'author': '李四',
'tags': ['python', 'mongodb', 'database'],
'date': '2023-09-15'
}

result = posts.insert_one(post_data)
print(f"已插入文档ID: {result.inserted_id}")

# 查询文档
print("\n所有文章:")
for post in posts.find():
print(f"标题: {post['title']}")
print(f"作者: {post['author']}")
print(f"标签: {', '.join(post['tags'])}")
print("-" * 30)

# 按条件查询
python_posts = posts.find({'tags': 'python'})
print("\n包含Python标签的文章:")
for post in python_posts:
print(f"标题: {post['title']}")

# 更新文档
update_result = posts.update_one(
{'title': 'Python与MongoDB的结合使用'},
{'$set': {'views': 100}}
)
print(f"\n已更新 {update_result.modified_count} 篇文章")

# 删除文档
# delete_result = posts.delete_one({'title': 'Python与MongoDB的结合使用'})
# print(f"\n已删除 {delete_result.deleted_count} 篇文章")

# 关闭连接
client.close()

输出:

已插入文档ID: 65f4b7c8d9e12345678901234

所有文章:
标题: Python与MongoDB的结合使用
作者: 李四
标签: python, mongodb, database
------------------------------

包含Python标签的文章:
标题: Python与MongoDB的结合使用

已更新 1 篇文章

连接池与最佳实践

在实际的Web应用中,创建和销毁数据库连接是一项昂贵的操作。连接池可以帮助管理连接并提高性能。

以下是使用SQLAlchemy创建连接池的例子:

python
from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

# 创建带连接池的引擎
engine = create_engine(
'mysql+mysqlconnector://username:password@localhost/dbname',
pool_size=5, # 池中最多保持的连接数
max_overflow=10, # 池满时可创建的额外连接数
pool_timeout=30, # 等待连接的超时时间(秒)
pool_recycle=1800 # 连接在池中可以存活的时间(秒)
)

# 使用连接
with engine.connect() as connection:
result = connection.execute("SELECT * FROM users LIMIT 5")
for row in result:
print(row)

数据库连接最佳实践

  1. 始终关闭连接: 使用try-finally块或上下文管理器确保连接被正确关闭。
  2. 使用参数化查询: 避免SQL注入攻击。
  3. 使用连接池: 在高并发应用中管理连接。
  4. 处理异常: 妥善处理数据库错误。
  5. 事务管理: 适当使用事务确保数据一致性。
python
# 良好的实践示例
import sqlite3

def get_user_by_id(user_id):
conn = None
try:
conn = sqlite3.connect('users.db')
cursor = conn.cursor()

# 使用参数化查询防止SQL注入
cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,))

user = cursor.fetchone()
return user

except sqlite3.Error as e:
print(f"数据库错误: {e}")
return None

finally:
# 确保连接被关闭
if conn:
conn.close()

实际案例: 简单的博客后端

让我们创建一个简单的博客后端API,它支持文章的增删改查操作:

python
from flask import Flask, request, jsonify
import sqlite3

app = Flask(__name__)

def get_db_connection():
conn = sqlite3.connect('blog.db')
conn.row_factory = sqlite3.Row
return conn

# 初始化数据库
def init_db():
conn = get_db_connection()
cursor = conn.cursor()

cursor.execute('''
CREATE TABLE IF NOT EXISTS posts (
id INTEGER PRIMARY KEY,
title TEXT NOT NULL,
content TEXT NOT NULL,
author TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')

conn.commit()
conn.close()

# 初始化数据库
init_db()

# 获取所有文章
@app.route('/api/posts', methods=['GET'])
def get_posts():
conn = get_db_connection()
posts = conn.execute('SELECT * FROM posts ORDER BY created_at DESC').fetchall()
conn.close()

return jsonify([dict(post) for post in posts])

# 获取单个文章
@app.route('/api/posts/<int:post_id>', methods=['GET'])
def get_post(post_id):
conn = get_db_connection()
post = conn.execute('SELECT * FROM posts WHERE id = ?', (post_id,)).fetchone()
conn.close()

if post is None:
return jsonify({'error': '文章不存在'}), 404

return jsonify(dict(post))

# 创建新文章
@app.route('/api/posts', methods=['POST'])
def create_post():
data = request.get_json()

if not data or not all(key in data for key in ['title', 'content', 'author']):
return jsonify({'error': '请提供标题、内容和作者'}), 400

conn = get_db_connection()
cursor = conn.cursor()
cursor.execute(
'INSERT INTO posts (title, content, author) VALUES (?, ?, ?)',
(data['title'], data['content'], data['author'])
)
conn.commit()
post_id = cursor.lastrowid
conn.close()

return jsonify({'id': post_id, 'message': '文章创建成功'}), 201

# 更新文章
@app.route('/api/posts/<int:post_id>', methods=['PUT'])
def update_post(post_id):
data = request.get_json()

if not data:
return jsonify({'error': '无数据提供'}), 400

conn = get_db_connection()
post = conn.execute('SELECT * FROM posts WHERE id = ?', (post_id,)).fetchone()

if post is None:
conn.close()
return jsonify({'error': '文章不存在'}), 404

updates = {}
if 'title' in data:
updates['title'] = data['title']
if 'content' in data:
updates['content'] = data['content']
if 'author' in data:
updates['author'] = data['author']

if updates:
query_parts = [f"{key} = ?" for key in updates.keys()]
query = f"UPDATE posts SET {', '.join(query_parts)} WHERE id = ?"
params = list(updates.values()) + [post_id]
conn.execute(query, params)
conn.commit()

conn.close()
return jsonify({'message': '文章更新成功'})

# 删除文章
@app.route('/api/posts/<int:post_id>', methods=['DELETE'])
def delete_post(post_id):
conn = get_db_connection()
post = conn.execute('SELECT * FROM posts WHERE id = ?', (post_id,)).fetchone()

if post is None:
conn.close()
return jsonify({'error': '文章不存在'}), 404

conn.execute('DELETE FROM posts WHERE id = ?', (post_id,))
conn.commit()
conn.close()

return jsonify({'message': '文章删除成功'})

if __name__ == '__main__':
app.run(debug=True)

通过上面的实现,我们创建了一个RESTful API,可以对博客文章进行增删改查操作。这个例子展示了如何在Web应用中使用数据库连接。

总结

Python提供了丰富的工具和库来连接各种数据库系统:

  1. SQLite: 轻量级、内置于Python标准库,适合小型应用
  2. MySQL/PostgreSQL: 强大的关系型数据库,适合大型应用
  3. SQLAlchemy: 功能丰富的ORM库,简化了数据库操作
  4. MongoDB: 流行的NoSQL选项,适合非结构化数据

无论您选择哪种数据库,理解基本的连接和操作原则都是至关重要的。这包括连接管理、查询执行、参数化查询和异常处理等。

后续学习路径

  1. 深入学习ORM框架(SQLAlchemy或Django ORM)
  2. 了解高级数据库概念(索引、事务、规范化)
  3. 学习数据库性能优化(查询优化、索引策略)
  4. 探索数据库迁移和版本控制工具

练习

  1. 创建一个简单的待办事项应用,使用SQLite存储任务列表。
  2. 实现一个使用MySQL的图书管理系统,包含图书和作者的关系。
  3. 使用SQLAlchemy创建一个学生成绩管理系统,实践一对多关系。
  4. 设计一个使用MongoDB的社交媒体帖子存储系统,存储带有评论的帖子。
警告

在实际应用中,永远不要在代码中硬编码数据库凭据。请使用环境变量或配置文件来存储敏感信息。

有用的资源

通过掌握Python数据库连接技术,您将能够开发出功能丰富、数据驱动的Web应用程序。