【Python 进阶 | 第四篇】Psycopg3 + Flask 实现 PostgreSQL CRUD 全流程：从连接池到RESTful接口

文章目录

• • 1. 环境准备：依赖安装
  • • 1.1 安装依赖
    • 1.2 项目结构
    • 1.3 环境变量配置
  • 2. Psycopg3：新一代 PostgreSQL 驱动
  • • 2.1 基本连接
    • 2.2 执行 SQL：cursor 的使用
    • 2.3 SQL 注入防护：参数化查询
    • 2.4 查询结果处理
    • 2.5 事务管理：commit 与 rollback
  • 3. 连接池管理：性能与资源的关键
  • • 3.1 为什么需要连接池
    • 3.2 连接池配置详解
    • 3.3 完整连接池管理器
    • 3.4 上下文管理器封装
  • 4. Flask 路由设计：RESTful 接口规范
  • • 4.1 RESTful 路由设计原则
    • 4.2 Flask 基础结构
    • 4.3 统一响应格式
    • 4.4 请求参数验证
  • 5. 完整 CRUD 流程：用户管理实战
  • • 5.1 数据库初始化
    • 5.2 用户模型
    • 5.3 用户路由（CRUD 完整实现）
    • 5.4 PUT vs PATCH 的区别
    • 5.5 常见 HTTP 状态码
  • 6. 总结

前言   此文是我学习 Python Web 开发过程中整理的数据库操作笔记，涵盖 Psycopg3 连接 PostgreSQL、连接池管理、Flask 路由设计以及完整的 CRUD 操作流程，结合代码实践总结而来，分享给大家。一是为了记录，二是用于日后的回顾，三也是希望能给其他初学者带来一点点帮助。

目录

• 环境准备：依赖安装
• Psycopg3：新一代 PostgreSQL 驱动
• 连接池管理：性能与资源的关键
• Flask 路由设计：RESTful 接口规范
• 完整 CRUD 流程：用户管理实战
• 总结

1. 环境准备：依赖安装

1.1 安装依赖

pip install psycopg[pip] # Psycopg3 核心驱动
pip install psycopg-binary # 二进制版本（无需编译）
pip install flask # Web 框架
pip install python-dotenv # 环境变量管理

1.2 项目结构

project/
├── app.py # Flask 应用入口
├── config.py # 配置文件
├── models/
│ └── user.py # 用户模型
├── routes/
│ └── user_routes.py # 用户路由
├── db/
│ ├── __init__.py
│ ├── pool.py # 连接池管理
│ └── init_db.py # 数据库初始化
├── .env # 环境变量
└── requirements.txt

1.3 环境变量配置

# .env
DB_HOST=localhost
DB_PORT=5432
DB_NAME=testdb
DB_USER=postgres
DB_PASSWORD=your_password
FLASK_ENV=development
FLASK_DEBUG=True

# db_config.py
import os
from dotenv import load_dotenv

load_dotenv()

class Config:
DB_HOST = os.getenv("DB_HOST", "localhost")
DB_PORT = int(os.getenv("DB_PORT", 5432))
DB_NAME = os.getenv("DB_NAME", "testdb")
DB_USER = os.getenv("DB_USER", "postgres")
DB_PASSWORD = os.getenv("DB_PASSWORD", "")
DB_MIN_CONN = int(os.getenv("DB_MIN_CONN", 2))
DB_MAX_CONN = int(os.getenv("DB_MAX_CONN", 10))

一句话总结：Psycopg3 是 PostgreSQL 的最新官方驱动，相比 Psycopg2，它支持 async/await、更现代的连接池、更好的类型提示，是未来趋势。

2. Psycopg3：新一代 PostgreSQL 驱动

2.1 基本连接

import psycopg

# 最简单的连接方式
conn = psycopg.connect(
host="localhost",
port=5432,
dbname="testdb",
user="postgres",
password="your_password"
)
print(conn) # <Connection …>

# 使用连接字符串（类似 SQLAlchemy）
conn = psycopg.connect("postgresql://postgres:your_password@localhost:5432/testdb")

# 关闭连接
conn.close()

2.2 执行 SQL：cursor 的使用

conn = psycopg.connect("postgresql://postgres:your_password@localhost:5432/testdb")

# 创建 cursor
cursor = conn.cursor()

# 执行查询
cursor.execute("SELECT version();")
result = cursor.fetchone()
print(f"PostgreSQL 版本: {result[0]}")

# 执行 DML 操作（需要 commit）
cursor.execute("INSERT INTO users (name, email) VALUES (%s, %s)", ("张三", "zhangsan@example.com"))
conn.commit() # 不要忘记 commit！

# 批量插入
users = [
("李四", "lisi@example.com"),
("王五", "wangwu@example.com"),
("赵六", "zhaoliu@example.com")
]
cursor.executemany(
"INSERT INTO users (name, email) VALUES (%s, %s)",
users
)
conn.commit()

# 关闭
cursor.close()
conn.close()

2.3 SQL 注入防护：参数化查询

Psycopg3 永远使用参数化查询，永远不要用字符串拼接！

# ❌ 错误：SQL 注入风险
user_input = "' OR '1'='1"
cursor.execute(f"SELECT * FROM users WHERE name = '{user_input}'") # 危险！

# ✅ 正确：参数化查询
cursor.execute("SELECT * FROM users WHERE name = %s", (user_input,))
# 或者用 $1, $2 占位符
cursor.execute("SELECT * FROM users WHERE name = $1", (user_input,))

2.4 查询结果处理

# 查询单条
cursor.execute("SELECT * FROM users WHERE id = %s", (1,))
user = cursor.fetchone() # 返回元组: (1, "张三", "zhangsan@example.com")
user = cursor.fetchone() # 返回字典: {'id': 1, 'name': '张三', …}

# 查询多条
cursor.execute("SELECT * FROM users LIMIT 10")
users = cursor.fetchall() # 返回列表

# 使用 named tuple（更易读）
cursor.execute("SELECT * FROM users WHERE id = %s", (1,))
user = cursor.fetchone(named=True) # 返回 asyncpg.Row 对象，支持 user.name 访问

# 遍历查询结果
cursor.execute("SELECT * FROM users")
for user in cursor:
print(user['name'], user['email'])

2.5 事务管理：commit 与 rollback

try:
cursor.execute("BEGIN")
cursor.execute("INSERT INTO users (name, email) VALUES (%s, %s)", ("张三", "zhangsan@example.com"))
cursor.execute("UPDATE users SET email = %s WHERE name = %s", ("new@example.com", "张三"))
conn.commit() # 所有操作一起提交
except Exception as e:
conn.rollback() # 出错时回滚所有操作
print(f"事务回滚: {e}")
finally:
cursor.close()
conn.close()

一句话总结：Psycopg3 的核心是 cursor 和 connection——cursor 执行 SQL，connection 管理事务；永远用参数化查询防注入，永远记得 commit。

3. 连接池管理：性能与资源的关键

3.1 为什么需要连接池

每次请求都创建新连接开销巨大。连接池预先建立一组连接，请求复用，用完归还，避免频繁创建销毁连接带来的性能损耗。

# db/db_pool.py
import psycopg
from psycopg_pool import AsyncConnectionPool
from contextlib import contextmanager

# 同步连接池
from psycopg_pool import ConnectionPool

pool = ConnectionPool(
"postgresql://postgres:your_password@localhost:5432/testdb",
min_size=2, # 最小连接数
max_size=10, # 最大连接数
timeout=30, # 获取连接超时时间（秒）
)

# 使用连接
with pool.connection() as conn:
cursor = conn.cursor()
cursor.execute("SELECT * FROM users")
users = cursor.fetchall()

# 上下文管理器会自动归还连接到池中

3.2 连接池配置详解

pool = ConnectionPool(
conninfo="postgresql://postgres:your_password@localhost:5432/testdb",
min_size=2, # 池中最小连接数
max_size=10, # 池中最大连接数
timeout=30, # 获取连接超时（秒）
max_idle_time=600, # 空闲连接最大存活时间（秒）
reconnect_timeout=300, # 断开重连超时（秒）
)

3.3 完整连接池管理器

# db/db_pool.py
import psycopg
from psycopg_pool import ConnectionPool
from config import Config

class DatabasePool:
_instance = None
_pool = None

def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._pool = ConnectionPool(
conninfo=(
f"postgresql://{Config.DB_USER}:{Config.DB_PASSWORD}"
f"@{Config.DB_HOST}:{Config.DB_PORT}/{Config.DB_NAME}"
),
min_size=Config.DB_MIN_CONN,
max_size=Config.DB_MAX_CONN,
timeout=30,
)
return cls._instance

@property
def pool(self):
return self._pool

def get_connection(self):
return self._pool.getconn()

def release_connection(self, conn):
self._pool.putconn(conn)

def close_all(self):
if self._pool:
self._pool.close()

db_pool = DatabasePool()

3.4 上下文管理器封装

# db/db_pool.py 补充

@contextmanager
def get_db_connection():
"""获取数据库连接的上下文管理器"""
conn = db_pool.get_connection()
try:
yield conn
except Exception as e:
conn.rollback()
raise e
finally:
db_pool.release_connection(conn)

@contextmanager
def get_db_cursor():
"""获取数据库 cursor 的上下文管理器"""
with get_db_connection() as conn:
cursor = conn.cursor()
try:
yield cursor
conn.commit()
except Exception as e:
conn.rollback()
raise e
finally:
cursor.close()

一句话总结：连接池是 Web 应用的性能关键——通过复用连接减少开销，通过上下文管理器自动管理连接的生命周期。

4. Flask 路由设计：RESTful 接口规范

4.1 RESTful 路由设计原则

4.2 Flask 基础结构

# start_up.py
from flask import Flask
from config import Config

app = Flask(__name__)
app.config.from_object(Config)

# 注册蓝图
from routes.user_routes import user_bp

app.register_blueprint(user_bp, url_prefix='/api')

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

4.3 统一响应格式

# utils/response.py
from flask import jsonify

def success_response(data=None, message="操作成功", code=200):
"""统一成功响应"""
return jsonify({
"code": code,
"message": message,
"data": data
}), code

def error_response(message="操作失败", code=400, errors=None):
"""统一错误响应"""
response = {
"code": code,
"message": message,
}
if errors:
response["errors"] = errors
return jsonify(response), code

4.4 请求参数验证

# utils/validators.py
from functools import wraps
from flask import request
from utils.response import error_response

def validate_json(*required_fields):
"""验证请求 JSON 必填字段"""
def decorator(f):
@wraps(f)
def wrapper(*args, **kwargs):
if not request.is_json:
return error_response("请求必须是 JSON 格式", 400)
data = request.get_json()
missing = [field for field in required_fields if field not in data or not data[field]]
if missing:
return error_response(f"缺少必填字段: {', '.join(missing)}", 400)
return f(*args, **kwargs)
return wrapper
return decorator

def validate_pagination(f):
"""验证分页参数"""
@wraps(f)
def wrapper(*args, **kwargs):
page = request.args.get('page', 1, type=int)
page_size = request.args.get('page_size', 10, type=int)
if page < 1:
return error_response("page 必须大于等于 1", 400)
if page_size < 1 or page_size > 100:
return error_response("page_size 必须在 1-100 之间", 400)
kwargs['page'] = page
kwargs['page_size'] = page_size
return f(*args, **kwargs)
return wrapper

一句话总结：RESTful 路由设计的关键是 HTTP 方法语义正确、响应格式统一、参数验证前置——让 API 既规范又好维护。

5. 完整 CRUD 流程：用户管理实战

5.1 数据库初始化

# db/init_db.py
from config.db_pool import db_pool

def init_database():
"""初始化数据库表"""
with db_pool.connection() as conn:
cursor = conn.cursor()

# 创建 users 表
cursor.execute("""
CREATE TABLE IF NOT EXISTS users (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL,
email VARCHAR(255) UNIQUE NOT NULL,
age INTEGER,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")

# 创建索引
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_users_email ON users(email)
""")

conn.commit()
print("数据库初始化完成")

if __name__ == '__main__':
init_database()

5.2 用户模型

# models/user.py
from dataclasses import dataclass
from datetime import datetime
from typing import Optional, List

@dataclass
class User:
id: Optional[int]
name: str
email: str
age: Optional[int]
created_at: Optional[datetime] = None
updated_at: Optional[datetime] = None

def to_dict(self):
"""转换为字典"""
return {
"id": self.id,
"name": self.name,
"email": self.email,
"age": self.age,
"created_at": self.created_at.isoformat() if self.created_at else None,
"updated_at": self.updated_at.isoformat() if self.updated_at else None,
}

@classmethod
def from_row(cls, row):
"""从数据库行转换为 User 对象"""
return cls(
id=row[0],
name=row[1],
email=row[2],
age=row[3],
created_at=row[4],
updated_at=row[5],
)

5.3 用户路由（CRUD 完整实现）

# routes/user_routes.py
from flask import Blueprint, request
from config.db_pool import db_pool
from models.user import User
from utils.response import success_response, error_response
from utils.validators import validate_json, validate_pagination

user_bp = Blueprint('users', __name__)

# ———- Create ———-
@user_bp.route('/users', methods=['POST'])
@validate_json('name', 'email')
def create_user():
"""创建用户"""
data = request.get_json()

try:
with db_pool.connection() as conn:
cursor = conn.cursor()
cursor.execute(
"""
INSERT INTO users (name, email, age)
VALUES (%s, %s, %s)
RETURNING id, name, email, age, created_at, updated_at
""",
(data['name'], data['email'], data.get('age'))
)
row = cursor.fetchone()
conn.commit()

user = User.from_row(row)
return success_response(data=user.to_dict(), message="用户创建成功", code=201)

except psycopg.errors.UniqueViolation:
return error_response("邮箱已存在", 409)
except Exception as e:
return error_response(f"创建用户失败: {str(e)}", 500)

# ———- Read All (分页) ———-
@user_bp.route('/users', methods=['GET'])
@validate_pagination
def get_users(page, page_size):
"""获取用户列表（分页）"""
offset = (page – 1) * page_size

with db_pool.connection() as conn:
cursor = conn.cursor()

# 查询总数
cursor.execute("SELECT COUNT(*) FROM users")
total = cursor.fetchone()[0]

# 查询分页数据
cursor.execute(
"""
SELECT id, name, email, age, created_at, updated_at
FROM users
ORDER BY created_at DESC
LIMIT %s OFFSET %s
""",
(page_size, offset)
)
rows = cursor.fetchall()

users = [User.from_row(row).to_dict() for row in rows]

return success_response(data={
"users": users,
"pagination": {
"page": page,
"page_size": page_size,
"total": total,
"total_pages": (total + page_size – 1) // page_size
}
})

# ———- Read One ———-
@user_bp.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
"""获取单个用户"""
with db_pool.connection() as conn:
cursor = conn.cursor()
cursor.execute(
"SELECT id, name, email, age, created_at, updated_at FROM users WHERE id = %s",
(user_id,)
)
row = cursor.fetchone()

if not row:
return error_response("用户不存在", 404)

user = User.from_row(row)
return success_response(data=user.to_dict())

# ———- Update (PUT 全量更新) ———-
@user_bp.route('/users/<int:user_id>', methods=['PUT'])
@validate_json('name', 'email')
def update_user(user_id):
"""完整更新用户"""
data = request.get_json()

with db_pool.connection() as conn:
cursor = conn.cursor()

# 检查用户是否存在
cursor.execute("SELECT id FROM users WHERE id = %s", (user_id,))
if not cursor.fetchone():
return error_response("用户不存在", 404)

# 更新用户
cursor.execute(
"""
UPDATE users
SET name = %s, email = %s, age = %s, updated_at = CURRENT_TIMESTAMP
WHERE id = %s
RETURNING id, name, email, age, created_at, updated_at
""",
(data['name'], data['email'], data.get('age'), user_id)
)
row = cursor.fetchone()
conn.commit()

user = User.from_row(row)
return success_response(data=user.to_dict(), message="用户更新成功")

# ———- Update (PATCH 部分更新) ———-
@user_bp.route('/users/<int:user_id>', methods=['PATCH'])
def patch_user(user_id):
"""部分更新用户"""
if not request.is_json:
return error_response("请求必须是 JSON 格式", 400)

data = request.get_json()
if not data:
return error_response("没有要更新的字段", 400)

# 动态构建更新语句
allowed_fields = ['name', 'email', 'age']
update_fields = []
values = []

for field in allowed_fields:
if field in data:
update_fields.append(f"{field} = %s")
values.append(data[field])

if not update_fields:
return error_response("没有有效的更新字段", 400)

update_fields.append("updated_at = CURRENT_TIMESTAMP")
values.append(user_id)

with db_pool.connection() as conn:
cursor = conn.cursor()

# 检查用户是否存在
cursor.execute("SELECT id FROM users WHERE id = %s", (user_id,))
if not cursor.fetchone():
return error_response("用户不存在", 404)

sql = f"UPDATE users SET {', '.join(update_fields)} WHERE id = %s RETURNING id, name, email, age, created_at, updated_at"
cursor.execute(sql, values)
row = cursor.fetchone()
conn.commit()

user = User.from_row(row)
return success_response(data=user.to_dict(), message="用户更新成功")

# ———- Delete ———-
@user_bp.route('/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
"""删除用户"""
with db_pool.connection() as conn:
cursor = conn.cursor()

# 检查用户是否存在
cursor.execute("SELECT id FROM users WHERE id = %s", (user_id,))
if not cursor.fetchone():
return error_response("用户不存在", 404)

cursor.execute("DELETE FROM users WHERE id = %s", (user_id,))
conn.commit()

return success_response(message="用户删除成功")

5.4 PUT vs PATCH 的区别

5.5 常见 HTTP 状态码

一句话总结：CRUD 的核心是语义正确——Create 用 POST 返回 201，Read 用 GET 返回数据，Update 用 PUT/PATCH，Delete 用 DELETE 并返回 204；参数验证前置，异常处理兜底。

6. 总结

通过这次梳理，我们完成了一个完整的 Flask + Psycopg3 + PostgreSQL 的 CRUD 流程：

• Psycopg3 提供了现代的 PostgreSQL 连接方式，参数化查询防注入、RETURNING 子句获取新记录
• 连接池 是 Web 应用性能的关键，避免频繁创建销毁连接的开销
• Flask 蓝图 让路由模块化，RESTful 设计让接口语义清晰
• 统一响应格式 和 参数验证 是 API 可维护性的保障

如果用一句话总结：

构建一个可靠的 Web API，核心就三件事：数据库连接稳定（连接池）、SQL 执行安全（参数化查询）、接口设计规范（RESTful + 统一响应）。

当你能够独立完成从数据库设计到接口开发的完整流程时，Python Web 开发才算是真正入门了。

如果我的内容对你有帮助，请点赞，评论，收藏。创作不易，你们的支持就是我坚持下去的动力！