Python脚本如何操作数据库包

wen 实用脚本 2026-07-07 6

本文目录导读：

Python脚本如何操作数据库包

关系型数据库
ORM（对象关系映射）框架
NoSQL数据库
数据库连接池
异步数据库操作
最佳实践

我来详细介绍Python操作数据库的常用方法和包：

关系型数据库

SQLite（内置，无需安装）

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
    )
''')
# 插入数据
cursor.execute('INSERT INTO users (name, email) VALUES (?, ?)', 
               ('张三', 'zhang@example.com'))
# 查询数据
cursor.execute('SELECT * FROM users')
rows = cursor.fetchall()
# 提交事务
conn.commit()
# 关闭连接
conn.close()

MySQL（需要安装 mysql-connector-python 或 pymysql）

# 安装：pip install mysql-connector-python
import mysql.connector
# 连接数据库
conn = mysql.connector.connect(
    host='localhost',
    user='root',
    password='password',
    database='mydb'
)
cursor = conn.cursor()
# 执行SQL
cursor.execute("SELECT * FROM users")
results = cursor.fetchall()
# 使用参数化查询防止SQL注入
cursor.execute("INSERT INTO users (name, age) VALUES (%s, %s)", 
               ('李四', 25))
conn.commit()
conn.close()

PostgreSQL（需要安装 psycopg2）

# 安装：pip install psycopg2-binary
import psycopg2
# 连接数据库
conn = psycopg2.connect(
    host='localhost',
    database='mydb',
    user='postgres',
    password='password'
)
cursor = conn.cursor()
# 执行查询
cursor.execute("SELECT * FROM users WHERE age > %s", (18,))
rows = cursor.fetchall()
conn.commit()
conn.close()

ORM（对象关系映射）框架

SQLAlchemy（推荐）

# 安装：pip install sqlalchemy
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
Base = declarative_base()
# 定义模型
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)
# 创建数据库引擎
engine = create_engine('sqlite:///example.db', echo=True)
Base.metadata.create_all(engine)
# 创建会话
Session = sessionmaker(bind=engine)
session = Session()
# 添加数据
new_user = User(name='王五', email='wang@example.com')
session.add(new_user)
session.commit()
# 查询数据
users = session.query(User).all()
for user in users:
    print(user.name, user.email)
session.close()

Django ORM（Web开发框架自带）

# 需要先安装Django：pip install django
# models.py
from django.db import models
class User(models.Model):
    name = models.CharField(max_length=100)
    email = models.EmailField(unique=True)
    created_at = models.DateTimeField(auto_now_add=True)
# 使用ORM
from myapp.models import User
# 创建
user = User.objects.create(name='赵六', email='zhao@example.com')
# 查询
users = User.objects.filter(name__contains='张')
user = User.objects.get(id=1)
# 更新
user.name = '新名字'
user.save()
# 删除
user.delete()

NoSQL数据库

MongoDB（需要安装 pymongo）

# 安装：pip install pymongo
from pymongo import MongoClient
# 连接MongoDB
client = MongoClient('mongodb://localhost:27017/')
db = client['mydatabase']
collection = db['users']
# 插入文档
user = {'name': '张三', 'age': 30, 'city': '北京'}
result = collection.insert_one(user)
# 查询文档
user = collection.find_one({'name': '张三'})
users = collection.find({'age': {'$gt': 25}})
# 更新文档
collection.update_one(
    {'name': '张三'},
    {'$set': {'age': 31}}
)
# 删除文档
collection.delete_one({'name': '张三'})

Redis（需要安装 redis）

# 安装：pip install redis
import redis
# 连接Redis
r = redis.Redis(host='localhost', port=6379, db=0)
# 字符串操作
r.set('key', 'value')
value = r.get('key')
# 哈希操作
r.hset('user:1', 'name', '张三')
r.hset('user:1', 'age', 30)
user = r.hgetall('user:1')
# 列表操作
r.lpush('mylist', 'item1', 'item2')
items = r.lrange('mylist', 0, -1)
# 设置过期时间
r.setex('temp_key', 3600, 'expires in 1 hour')

数据库连接池

使用连接池（提高性能）

# 安装：pip install DBUtils
from dbutils.pooled_db import PooledDB
import pymysql
# 创建连接池
pool = PooledDB(
    creator=pymysql,
    maxconnections=6,
    mincached=2,
    maxcached=5,
    blocking=True,
    host='localhost',
    user='root',
    password='password',
    database='mydb'
)
# 从池中获取连接
conn = pool.connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM users")
result = cursor.fetchall()
cursor.close()
conn.close()  # 返回连接池而不是真正关闭

异步数据库操作

使用 async/await（Python 3.7+）

# 安装：pip install databases
import asyncio
from databases import Database
async def main():
    # 连接数据库
    database = Database('sqlite:///example.db')
    await database.connect()
    # 执行查询
    query = "SELECT * FROM users"
    rows = await database.fetch_all(query)
    # 执行插入
    await database.execute("INSERT INTO users (name) VALUES (:name)", 
                          {"name": "异步用户"})
    await database.disconnect()
# 运行异步函数
asyncio.run(main())

最佳实践

使用上下文管理器

import sqlite3
from contextlib import contextmanager
@contextmanager
def get_db():
    conn = sqlite3.connect('database.db')
    try:
        yield conn
        conn.commit()
    except:
        conn.rollback()
        raise
    finally:
        conn.close()
# 使用示例
with get_db() as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users")
    print(cursor.fetchall())

配置文件管理

# config.py
DB_CONFIG = {
    'host': 'localhost',
    'port': 3306,
    'user': 'root',
    'password': 'password',
    'database': 'mydb',
    'charset': 'utf8mb4'
}
# main.py
import pymysql
from config import DB_CONFIG
conn = pymysql.connect(**DB_CONFIG)

# 关系型数据库
pip install mysql-connector-python  # MySQL
pip install psycopg2-binary          # PostgreSQL
pip install pymysql                  # MySQL替代方案
# ORM框架
pip install sqlalchemy              # SQLAlchemy
pip install django                  # Django ORM
# NoSQL数据库
pip install pymongo                 # MongoDB
pip install redis                   # Redis
# 连接池
pip install DBUtils                 # 数据库连接池
# 异步操作
pip install databases               # 异步数据库操作

根据你的具体需求选择合适的数据库包，对于中小型项目推荐使用SQLAlchemy,它支持多种数据库且性能优秀。