企业官网建设流程全解析

1. 从“连接”说起：为什么是Python + PostgreSQL？

如果你正在用Python做数据分析、后端开发，或者想把手头的Excel数据存到一个更靠谱的地方，那么“连接数据库”就是你绕不开的第一步。Python生态里能连的数据库不少，MySQL、SQLite都挺常见，但PostgreSQL（简称Postgres）有点不一样。它不只是个存数据的“仓库”，更像一个功能强大的“工具箱”，尤其适合处理复杂查询、地理空间数据，或者对数据一致性要求极高的场景。比如，你想分析用户行为轨迹，或者构建一个需要处理JSON、数组这类非结构化数据的应用，Postgres的扩展性就比传统关系型数据库强不少。

我最初接触Postgres也是因为一个数据分析项目，需要处理大量带有地理位置信息和时间序列的日志。用MySQL写一些复杂的窗口函数和地理函数时总觉得有点“拧巴”，换到Postgres后，很多原生支持的特性让代码清爽了很多。所以，今天这篇内容，我就从一个实际使用者的角度，聊聊怎么用Python稳稳当当地连上Postgres，以及在这个过程中，有哪些新手容易踩、老手也可能疏忽的“坑”。

2. 环境准备：选对“桥梁”和“码头”

在动手写代码之前，有两件事必须搞定：一是你的Python环境里得有连接Postgres的“桥梁”——也就是驱动库；二是你得有一个正在运行的Postgres“码头”——也就是数据库服务实例。这两者缺一不可。

2.1 驱动库选型：psycopg2 还是 asyncpg？

Python连接Postgres，主流选择有两个：psycopg2（及其现代版本psycopg3）和asyncpg。怎么选？看你的应用场景。

psycopg2/psycopg3： 稳健全面的“多面手”这是最经典、使用最广泛的驱动。psycopg2成熟稳定，几乎支持所有Postgres特性。而psycopg3是它的现代化重构版本，在保持API兼容性的同时，性能更好，内存占用更低，并且原生支持Python的异步上下文管理器。对于绝大多数同步Web应用（如Django、Flask）、数据脚本和ETL任务，选它准没错。

安装命令很简单：

# 安装 psycopg2（二进制包，无需系统依赖，推荐） pip install psycopg2-binary # 或者安装最新的 psycopg3 pip install psycopg

这里有个细节：psycopg2-binary预编译了C扩展，开箱即用。如果你在Windows或macOS上快速开始，用这个。但如果是在Linux生产环境，为了获得最佳性能和兼容性，通常建议安装psycopg2（非binary），它会从源码编译，但需要你先安装系统级的Postgres开发库（如libpq-dev）。

asyncpg： 高性能的“闪电侠”如果你的应用是基于asyncio的异步框架（如FastAPI、Sanic），并且对数据库查询的吞吐量有极致要求，那么asyncpg是更好的选择。它直接从底层协议实现，避免了psycopg2的一些开销，在大量并发小查询的场景下，性能提升非常明显。但它对Postgres某些高级特性（如监听/通知）的支持不如psycopg2全面。

安装：

pip install asyncpg

我的选择建议：

• 新手入门、通用项目、Django/Flask应用：无脑选psycopg2-binary或psycopg。生态最好，资料最多，踩坑了也容易找到答案。
• 高性能异步API、微服务：认真考虑asyncpg。
• 本文后续的示例将以psycopg2为主进行讲解，因为它的模式最具代表性，理解了它，再去看asyncpg也会很容易。

2.2 数据库实例：本地、远程还是Docker？

有了驱动，还得有数据库服务。通常有三种方式获取一个Postgres实例：

1. 本地安装：在你的开发机上直接安装Postgres。好处是网络延迟为零，完全可控。缺点是安装和配置稍显繁琐，不同操作系统步骤不同。对于想深入学习Postgres内部机制的朋友，推荐这种方式。
2. 云数据库服务：如AWS RDS、Google Cloud SQL、阿里云RDS等。这是生产环境的主流选择，省去了运维的麻烦，自带高可用和备份。对于开发测试，它们通常也提供免费或低价的入门套餐。
3. Docker运行：这是目前开发环境最流行、最干净的方式。一条命令就能拉起一个Postgres容器，不用污染主机环境，不用了随时删除，非常方便。

这里重点说一下Docker方式，因为它能极大简化环境准备。确保你的机器上安装了Docker，然后执行：

docker run --name my-postgres \ -e POSTGRES_PASSWORD=mysecretpassword \ -e POSTGRES_USER=myuser \ -e POSTGRES_DB=mydatabase \ -p 5432:5432 \ -d postgres:latest

这条命令做了几件事：

• --name my-postgres：给容器起个名字，方便管理。
• -e环境变量：分别设置了默认的超级用户密码、用户名和数据库名。务必修改mysecretpassword为一个强密码！
• -p 5432:5432：将容器内的Postgres默认端口（5432）映射到宿主机的5432端口。
• -d：后台运行。
• postgres:latest：使用官方的最新镜像。

运行后，一个包含完整Postgres服务的容器就启动了。你可以通过docker ps查看运行状态，用docker logs my-postgres查看日志。

注意：Docker方式的数据默认存储在容器内部，容器删除数据也会丢失。如果需要持久化，需要添加-v /your/host/path:/var/lib/postgresql/data参数，将数据目录挂载到宿主机。

3. 建立连接：参数、上下文与异常处理

环境就绪，现在可以写Python代码了。连接数据库的核心是创建一个“连接对象”（Connection），它代表了一个到数据库服务器的网络会话。

3.1 基础连接：拼凑连接字符串

使用psycopg2连接，最基本的方式是使用connect()函数并传入一个连接字符串（DSN）。

import psycopg2 # 最基本的连接字符串格式 conn = psycopg2.connect( host="localhost", # 数据库主机地址，如果是Docker本地运行就是localhost port="5432", # 端口，默认5432 database="mydatabase", # 你要连接的数据库名 user="myuser", # 用户名 password="mysecretpassword" # 密码 )

这是最直白的写法，每个参数一目了然。但在实际项目中，我们很少把密码等敏感信息硬编码在代码里。更常见的做法是使用一个完整的DSN字符串，或者从环境变量、配置文件中读取。

使用DSN字符串：

dsn = "host=localhost port=5432 dbname=mydatabase user=myuser password=mysecretpassword" conn = psycopg2.connect(dsn)

或者，如果你的连接参数都放在一个字典里，可以这样：

params = { 'host': 'localhost', 'port': '5432', 'dbname': 'mydatabase', 'user': 'myuser', 'password': 'mysecretpassword' } conn = psycopg2.connect(**params) # 字典解包传入

3.2 生产环境实践：连接池与配置管理

直接创建连接对象在简单脚本里没问题，但在Web服务器或长期运行的应用中，频繁地打开和关闭数据库连接会带来巨大的性能开销。这时就需要连接池。

连接池会预先创建一定数量的连接放在“池子”里，当应用需要连接时，从池中取用一个空闲的连接，用完后归还，而不是真正关闭。psycopg2本身不提供连接池，但可以通过psycopg2.pool模块或第三方库（如SQLAlchemy的引擎）来实现。

一个简单的ThreadedConnectionPool示例：

from psycopg2 import pool # 创建连接池，最小2个连接，最大10个 connection_pool = pool.ThreadedConnectionPool( minconn=2, maxconn=10, host='localhost', database='mydatabase', user='myuser', password='mysecretpassword' ) # 从池中获取一个连接 conn = connection_pool.getconn() try: # ... 使用conn执行操作 cur = conn.cursor() cur.execute("SELECT NOW();") print(cur.fetchone()) finally: # 使用完毕后，一定要将连接放回池中，而不是关闭 connection_pool.putconn(conn) # 应用关闭时，关闭整个连接池 connection_pool.closeall()

配置管理：安全地存储凭据永远不要将数据库密码提交到版本控制系统（如Git）！正确的做法是使用环境变量或专门的配置文件（如.env文件），并通过python-dotenv等库来加载。

1. 在项目根目录创建.env文件：

   DB_HOST=localhost DB_PORT=5432 DB_NAME=mydatabase DB_USER=myuser DB_PASSWORD=supersecretpassword

2. 在.gitignore文件中添加.env，确保它不会被提交。
3. 在Python代码中读取：

   import os from dotenv import load_dotenv import psycopg2 load_dotenv() # 加载 .env 文件中的环境变量 conn = psycopg2.connect( host=os.getenv('DB_HOST'), port=os.getenv('DB_PORT'), database=os.getenv('DB_NAME'), user=os.getenv('DB_USER'), password=os.getenv('DB_PASSWORD') )

3.3 使用上下文管理器：确保资源被正确释放

Python的with语句（上下文管理器）是管理资源（如文件、网络连接、数据库连接）的利器。它可以确保即使在发生异常的情况下，资源也能被正确关闭。psycopg2的连接和游标对象都支持上下文管理器。

最佳实践写法：

import psycopg2 # 连接字符串（实际应从环境变量读取） dsn = "host=localhost dbname=mydatabase user=myuser password=mysecretpassword" try: # 使用 with 语句管理连接 with psycopg2.connect(dsn) as conn: # 自动提交模式设置（稍后详解） conn.autocommit = False # 使用 with 语句管理游标 with conn.cursor() as cur: # 执行SQL语句 cur.execute("INSERT INTO users (name, email) VALUES (%s, %s)", ('张三', 'zhangsan@example.com')) cur.execute("SELECT * FROM users WHERE name = %s", ('张三',)) # 获取结果 rows = cur.fetchall() for row in rows: print(row) # 如果没有异常，提交事务 conn.commit() except psycopg2.Error as e: # 发生任何数据库错误，连接上下文管理器会自动回滚事务并关闭连接 # 游标上下文管理器也会自动关闭游标 print(f"数据库操作出错: {e}") # 这里可以记录日志或进行其他错误处理

这段代码的精妙之处在于：

• with psycopg2.connect(...) as conn:：无论块内代码是否发生异常，当退出with块时，连接都会自动关闭。
• with conn.cursor() as cur:：游标也会被自动关闭。
• 如果在with块内发生异常，连接上下文管理器会自动回滚当前事务，然后关闭连接。这防止了数据处于不一致的状态。

3.4 连接参数详解与常见错误排查

连接时可能遇到各种错误，理解连接参数和常见错误信息能帮你快速定位问题。

关键连接参数：

• connect_timeout：尝试连接的超时时间（秒），默认是None（无限等待）。在网络不稳定的环境，可以设置为5或10。
• sslmode：SSL连接模式。连接云数据库（如RDS）时，通常需要设置为'require'或'verify-ca'。本地开发可以设为'disable'。
• client_encoding：设置客户端编码，确保与数据库编码一致，避免中文乱码。可以设置为'UTF8'。

常见连接错误与排查：

1. psycopg2.OperationalError: could not connect to server: Connection refused

   • 原因：Postgres服务没启动，或者监听地址/端口不对。
   • 排查：
     • 检查服务状态：sudo systemctl status postgresql(Linux) 或docker ps(查看容器是否运行)。
     • 检查端口是否被占用：netstat -tlnp | grep 5432。
     • 检查Postgres配置文件postgresql.conf中的listen_addresses是否为'*'或'localhost'，以及pg_hba.conf中的认证规则是否允许你的客户端IP和用户连接。

2. psycopg2.OperationalError: FATAL: password authentication failed for user "myuser"

   • 原因：用户名或密码错误。
   • 排查：仔细核对密码。对于Docker容器，密码是启动时通过POSTGRES_PASSWORD设置的。也可以尝试用psql命令行工具直接连接验证。

3. psycopg2.OperationalError: timeout expired

   • 原因：网络问题或服务器负载过高，在connect_timeout内未完成连接。
   • 排查：增加connect_timeout值，检查网络连通性（ping、telnet端口）。

4. 执行查询：游标、参数化与结果处理

连接建立后，所有与数据库的交互都通过游标（Cursor）对象进行。你可以把游标想象成一个在数据库结果集中移动的“指针”，它负责执行SQL语句并获取结果。

4.1 基础查询与参数化传递

执行一个简单的查询：

with conn.cursor() as cur: # 执行查询 cur.execute("SELECT version();") # 获取单条结果 db_version = cur.fetchone() print(f"PostgreSQL版本: {db_version[0]}")

fetchone()返回结果集的下一行，作为一个元组。如果查询返回多列，可以通过索引访问，如row[0],row[1]。

至关重要的参数化查询这是数据库编程中安全性的基石。永远不要使用字符串拼接的方式将变量传入SQL语句！

错误示范（极易导致SQL注入攻击）：

user_input = "张三'; DROP TABLE users; --" sql = f"SELECT * FROM users WHERE name = '{user_input}'" # 危险！ cur.execute(sql)

恶意用户输入可以轻易破坏你的数据库。

正确示范（使用占位符%s）：

user_name = "张三" with conn.cursor() as cur: # 使用 %s 作为占位符，第二个参数传入一个元组 cur.execute("SELECT * FROM users WHERE name = %s", (user_name,)) # 或者传入一个列表 cur.execute("SELECT * FROM users WHERE name = %s AND age > %s", [user_name, 18])

psycopg2会自动处理参数的类型转换和引号转义，确保输入被安全地处理，从根本上杜绝SQL注入。%s是psycopg2的占位符，无论参数是什么数据类型（字符串、整数、日期），都使用%s。

4.2 处理多种结果集

根据不同的SQL语句，你需要使用不同的方法来获取结果。

• fetchone()：获取下一行。常用于你知道只返回一行结果时，或者在循环中逐行处理大数据集。

  cur.execute("SELECT COUNT(*) FROM users;") count = cur.fetchone()[0] # 获取单个值

• fetchall()：获取所有（剩余）行，返回一个由元组组成的列表。注意：如果查询结果非常大，这可能会耗尽内存。仅在你确信结果集较小时使用。

  cur.execute("SELECT id, name FROM users LIMIT 10;") all_users = cur.fetchall() for user in all_users: print(f"ID: {user[0]}, Name: {user[1]}")

• fetchmany(size)：获取指定大小的行数。这是处理大型结果集的最佳方式，它允许你分批处理数据，平衡内存和I/O。

  cur.execute("SELECT * FROM large_table;") while True: rows = cur.fetchmany(100) # 每次获取100行 if not rows: break for row in rows: process_row(row) # 处理每一批数据

4.3 插入、更新与删除：处理事务

对于修改数据的语句（INSERT, UPDATE, DELETE），execute()方法同样适用。但这里涉及一个关键概念：事务。

默认情况下，psycopg2连接处于自动提交模式关闭的状态（autocommit=False）。这意味着你执行的多个SQL语句会被组合成一个事务，只有显式调用conn.commit()后，更改才会永久保存到数据库。如果发生错误，你可以调用conn.rollback()回滚所有未提交的更改。

一个完整的事务示例：

try: with conn.cursor() as cur: # 插入一条新用户记录 cur.execute( "INSERT INTO users (name, email) VALUES (%s, %s) RETURNING id;", ('李四', 'lisi@example.com') ) # 使用 RETURNING 子句获取刚插入的ID new_user_id = cur.fetchone()[0] print(f"新用户ID: {new_user_id}") # 为新用户创建一个账户记录 cur.execute( "INSERT INTO accounts (user_id, balance) VALUES (%s, %s);", (new_user_id, 100.00) ) # 所有操作成功，提交事务 conn.commit() print("事务提交成功，数据已保存。") except psycopg2.Error as e: # 发生任何错误，回滚事务 conn.rollback() print(f"操作失败，已回滚: {e}")

这个例子展示了“原子性”：要么两个INSERT都成功，要么都失败，不会出现用户创建了但没有账户的情况。RETURNING是Postgres的一个强大特性，可以在插入或更新后立即返回指定的列值，非常有用。

批量操作：executemany()当需要插入大量数据时，逐条执行execute()效率很低。应该使用executemany()。

users_data = [ ('王五', 'wangwu@example.com'), ('赵六', 'zhaoliu@example.com'), ('孙七', 'sunqi@example.com'), ] with conn.cursor() as cur: # 注意：这里的占位符仍然是 %s，但第二个参数是一个包含多个元组的列表 cur.executemany( "INSERT INTO users (name, email) VALUES (%s, %s);", users_data ) conn.commit()

executemany()在内部会进行优化，通常比循环调用execute()快得多。但对于海量数据（数十万以上），还有更高效的方法，如使用COPY命令，这将在后面提到。

5. 高级特性与性能优化

掌握了基本操作后，我们可以看看如何利用Postgres和psycopg2的一些高级特性来提升应用的健壮性和性能。

5.1 使用服务器端游标（命名游标）

当查询结果集非常大（例如数百万行）时，使用fetchall()或fetchmany()在客户端分批获取，仍然需要服务器一次性准备好所有结果，可能对服务器内存造成压力。服务器端游标（Server-side cursor）或命名游标（Named cursor）允许你在服务器端维持一个游标，客户端可以像翻阅一本书一样，按需“流式”获取结果。

# 创建连接时指定不自动提交，因为命名游标需要在事务内使用 conn.autocommit = False with conn.cursor(name='my_large_cursor') as cur: # 给游标一个名字 cur.itersize = 1000 # 每次从服务器传输的批大小 cur.execute("SELECT * FROM gigantic_table;") # 使用 for 循环迭代，每次从服务器获取 itersize 行 for row in cur: process_row(row) # 在处理过程中，可以随时 break，未传输的数据不会拉到客户端 conn.commit()

使用命名游标时，for row in cur:是一个生成器，它按itersize的大小分批从服务器拉取数据，极大地降低了客户端和服务器的瞬时内存压力。

5.2 利用 COPY 命令进行高速数据导入导出

这是Postgres的“杀手锏”之一，用于极高速地批量导入或导出数据。psycopg2通过copy_from()和copy_to()方法提供了支持。

从文件或类文件对象导入数据：假设你有一个CSV文件data.csv：

1,Alice,alice@example.com 2,Bob,bob@example.com

with conn.cursor() as cur: # 假设表 users 有 id, name, email 三列 with open('data.csv', 'r') as f: # copy_from 需要文件对象和表名 # 默认分隔符是制表符，CSV文件需指定 delimiter=',' cur.copy_from(f, 'users', sep=',') conn.commit()

将查询结果导出到文件：

with conn.cursor() as cur: with open('output.csv', 'w') as f: cur.copy_to(f, 'users', sep=',')

COPY命令比循环INSERT或executemany()快一个数量级以上，是数据迁移或初始化时必备的工具。copy_from也支持从StringIO等内存对象读取，方便直接处理程序中的数据结构。

5.3 处理复杂数据类型：JSON、数组与自定义类型

Postgres支持丰富的原生数据类型，如JSON/JSONB、数组等。psycopg2能很好地处理它们。

JSON/JSONB：

import json data = {'name': '张三', 'age': 30, 'tags': ['tech', 'music']} json_data = json.dumps(data, ensure_ascii=False) # 序列化为JSON字符串 with conn.cursor() as cur: # 插入JSONB数据 cur.execute( "INSERT INTO user_profiles (user_id, profile) VALUES (%s, %s::jsonb);", (1, json_data) ) # 查询并解析 cur.execute("SELECT profile FROM user_profiles WHERE user_id = %s;", (1,)) result = cur.fetchone()[0] # 返回的是一个Python字典（如果列类型是jsonb） print(result['tags']) # 输出: ['tech', 'music']

注意，对于jsonb类型，psycopg2会自动将其转换为Python字典或列表。对于json类型，返回的是字符串。

数组：

with conn.cursor() as cur: # 插入数组 cur.execute( "INSERT INTO products (name, tags) VALUES (%s, %s);", ('笔记本电脑', ['电子', '电脑', '便携']) ) # 查询包含特定标签的产品 cur.execute("SELECT name FROM products WHERE %s = ANY(tags);", ('电脑',))

Postgres数组在Python中对应的是list。ANY()操作符用于检查数组是否包含某个元素。

5.4 连接健康检查与重连机制

在长时间运行的应用中，数据库连接可能会因为网络波动、服务器重启等原因中断。一个健壮的程序需要具备连接健康检查和自动重连的能力。

一个简单的实现思路是，在执行关键查询前，先执行一个轻量级的测试查询（如SELECT 1;）。如果失败，则关闭旧连接，建立新连接。更复杂的方案可以结合连接池和重试装饰器。

import time from psycopg2 import OperationalError def get_healthy_connection(original_conn, dsn): """检查连接是否健康，如果不健康则重建""" try: with original_conn.cursor() as cur: cur.execute("SELECT 1;") return original_conn # 连接健康，直接返回 except OperationalError: print("连接中断，尝试重连...") try: new_conn = psycopg2.connect(dsn) print("重连成功。") return new_conn except OperationalError as e: print(f"重连失败: {e}") raise e # 在长时间循环中 dsn = "host=localhost dbname=mydatabase user=myuser password=mysecretpassword" conn = psycopg2.connect(dsn) while True: try: conn = get_healthy_connection(conn, dsn) with conn.cursor() as cur: cur.execute("SELECT ...") # 你的业务查询 # ... 处理结果 time.sleep(60) # 每分钟执行一次 except Exception as e: print(f"业务执行失败: {e}") time.sleep(5) # 失败后等待一段时间再重试

6. 实战：构建一个简单的数据访问层

理论说再多，不如一个实际例子。我们来构建一个简单的用户管理模块，将上面的知识点串联起来。这个模块会包含连接管理、基本的CRUD操作和错误处理。

首先，我们创建一个配置文件config.py来管理数据库连接参数：

# config.py import os from dotenv import load_dotenv load_dotenv() DB_CONFIG = { 'host': os.getenv('DB_HOST', 'localhost'), 'port': os.getenv('DB_PORT', '5432'), 'database': os.getenv('DB_NAME', 'mydatabase'), 'user': os.getenv('DB_USER', 'myuser'), 'password': os.getenv('DB_PASSWORD', ''), }

接着，创建一个数据库连接和操作类database.py：

# database.py import psycopg2 from psycopg2 import pool, sql from psycopg2.extras import RealDictCursor import logging from config import DB_CONFIG logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class Database: _connection_pool = None @classmethod def initialize(cls): """初始化连接池""" try: cls._connection_pool = pool.ThreadedConnectionPool( minconn=1, maxconn=10, **DB_CONFIG ) logger.info("数据库连接池初始化成功。") except Exception as e: logger.error(f"连接池初始化失败: {e}") raise @classmethod def get_connection(cls): """从池中获取一个连接""" if cls._connection_pool is None: cls.initialize() return cls._connection_pool.getconn() @classmethod def return_connection(cls, connection): """将连接归还给池""" if cls._connection_pool: cls._connection_pool.putconn(connection) @classmethod def close_all_connections(cls): """关闭所有连接""" if cls._connection_pool: cls._connection_pool.closeall() logger.info("所有数据库连接已关闭。") @staticmethod def get_dict_cursor(connection): """获取一个返回字典格式结果的游标""" return connection.cursor(cursor_factory=RealDictCursor) class UserDAO: """用户数据访问对象""" @staticmethod def create_user(name, email): """创建用户，返回新用户的ID""" conn = Database.get_connection() try: with conn: with conn.cursor() as cur: # 使用 RETURNING 获取自增ID cur.execute( """ INSERT INTO users (name, email, created_at) VALUES (%s, %s, NOW()) RETURNING id; """, (name, email) ) new_id = cur.fetchone()[0] logger.info(f"用户创建成功，ID: {new_id}") return new_id except psycopg2.IntegrityError as e: # 捕获唯一约束冲突（如重复邮箱） logger.warning(f"创建用户失败，可能邮箱已存在: {e}") return None except psycopg2.Error as e: logger.error(f"数据库操作失败: {e}") return None finally: Database.return_connection(conn) @staticmethod def get_user_by_id(user_id): """根据ID获取用户信息（返回字典）""" conn = Database.get_connection() try: with conn: with Database.get_dict_cursor(conn) as cur: cur.execute( "SELECT id, name, email, created_at FROM users WHERE id = %s;", (user_id,) ) user = cur.fetchone() # 返回一个字典或None return user except psycopg2.Error as e: logger.error(f"查询用户失败: {e}") return None finally: Database.return_connection(conn) @staticmethod def update_user_email(user_id, new_email): """更新用户邮箱""" conn = Database.get_connection() try: with conn: with conn.cursor() as cur: cur.execute( "UPDATE users SET email = %s WHERE id = %s;", (new_email, user_id) ) if cur.rowcount == 0: logger.warning(f"未找到ID为 {user_id} 的用户。") return False logger.info(f"用户 {user_id} 邮箱更新成功。") return True except psycopg2.IntegrityError as e: logger.warning(f"邮箱更新失败，可能新邮箱已存在: {e}") return False except psycopg2.Error as e: logger.error(f"更新操作失败: {e}") return False finally: Database.return_connection(conn) @staticmethod def search_users_by_name(name_pattern, limit=10): """根据姓名模糊查询用户""" conn = Database.get_connection() try: with conn: with Database.get_dict_cursor(conn) as cur: # 使用 ILIKE 进行不区分大小写的模糊匹配 cur.execute( """ SELECT id, name, email, created_at FROM users WHERE name ILIKE %s ORDER BY created_at DESC LIMIT %s; """, (f'%{name_pattern}%', limit) ) users = cur.fetchall() # 返回字典列表 return users except psycopg2.Error as e: logger.error(f"搜索用户失败: {e}") return [] finally: Database.return_connection(conn) # 应用启动时初始化连接池 Database.initialize()

最后，在一个主程序app.py中使用这个数据访问层：

# app.py from database import UserDAO if __name__ == "__main__": # 1. 创建用户 user_id = UserDAO.create_user("测试用户", "test@example.com") if user_id: print(f"创建的用户ID: {user_id}") # 2. 查询用户 user = UserDAO.get_user_by_id(user_id) if user: print(f"查询到的用户: {user}") # 3. 更新用户 success = UserDAO.update_user_email(user_id, "new_email@example.com") print(f"更新邮箱结果: {'成功' if success else '失败'}") # 4. 模糊搜索 users = UserDAO.search_users_by_name("测试", 5) print(f"搜索到 {len(users)} 个用户:") for u in users: print(f" - {u['name']} ({u['email']})")

这个实战例子涵盖了多个关键点：

1. 连接池管理：使用ThreadedConnectionPool管理连接，避免频繁开关连接。
2. 资源自动释放：大量使用with语句管理连接和游标，确保异常时也能正确清理。
3. 字典游标：使用RealDictCursor让查询结果以字典形式返回，通过列名访问数据比索引更清晰。
4. 错误处理：专门捕获psycopg2.IntegrityError（如唯一约束违反）和通用的psycopg2.Error，并进行不同级别的日志记录。
5. 事务控制：每个UserDAO的方法都在一个独立的with conn:块中运行，形成一个事务。方法成功执行则提交，发生异常则自动回滚。
6. 安全的参数化查询：所有SQL语句都使用%s占位符。
7. 日志记录：使用Python标准库的logging记录操作日志，便于调试和监控。

通过这样的分层设计，业务逻辑与数据库访问细节解耦，代码更清晰，也更易于测试和维护。你可以在此基础上，继续扩展更复杂的数据模型和查询逻辑。