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)
数据库连接最佳实践
- 始终关闭连接: 使用
try-finally块或上下文管理器确保连接被正确关闭。 - 使用参数化查询: 避免SQL注入攻击。
- 使用连接池: 在高并发应用中管理连接。
- 处理异常: 妥善处理数据库错误。
- 事务管理: 适当使用事务确保数据一致性。
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提供了丰富的工具和库来连接各种数据库系统:
- SQLite: 轻量级、内置于Python标准库，适合小型应用
- MySQL/PostgreSQL: 强大的关系型数据库，适合大型应用
- SQLAlchemy: 功能丰富的ORM库，简化了数据库操作
- MongoDB: 流行的NoSQL选项，适合非结构化数据
无论您选择哪种数据库，理解基本的连接和操作原则都是至关重要的。这包括连接管理、查询执行、参数化查询和异常处理等。
后续学习路径
- 深入学习ORM框架(SQLAlchemy或Django ORM)
- 了解高级数据库概念(索引、事务、规范化)
- 学习数据库性能优化(查询优化、索引策略)
- 探索数据库迁移和版本控制工具
练习
- 创建一个简单的待办事项应用，使用SQLite存储任务列表。
- 实现一个使用MySQL的图书管理系统，包含图书和作者的关系。
- 使用SQLAlchemy创建一个学生成绩管理系统，实践一对多关系。
- 设计一个使用MongoDB的社交媒体帖子存储系统，存储带有评论的帖子。
警告
在实际应用中，永远不要在代码中硬编码数据库凭据。请使用环境变量或配置文件来存储敏感信息。
有用的资源
通过掌握Python数据库连接技术，您将能够开发出功能丰富、数据驱动的Web应用程序。