Python 与数据库的交互是每个软件开发者绕不过去的一道坎。正好，遇有需求要用到 PostgreSQL，所以就它了。废话不多说，走起。

1. 先简单介绍一下 PostgreSQL

众所周知，PostgreSQL 是一个功能强大的开源对象-关系型数据库管理系统（ORDBMS），以其高可靠性、扩展性和对 SQL 标准的严格遵循而闻名。它支持广泛的功能，适用于各种应用场景，从小型项目到大型企业级系统。

1.1. PostgreSQL 的核心特点

ACID 兼容性
支持事务的原子性、一致性、隔离性和持久性，确保数据的安全性和可靠性。

扩展性强
允许用户自定义数据类型、函数、操作符和索引方法，满足特定业务需求。

多版本并发控制（MVCC）
支持高并发操作，允许多个事务同时进行而不会相互阻塞。

丰富的索引支持
支持多种索引类型，包括 B-tree、Hash、GiST、SP-GiST、GIN 和 BRIN，优化查询性能。

JSON 和 JSONB 支持
提供对 JSON 数据的原生支持，JSONB 类型还支持高效的查询和索引。

地理空间数据处理
通过 PostGIS 扩展支持地理空间数据，适用于地理信息系统（GIS）应用。

全文搜索
内置全文搜索功能，支持高效的文本检索。

高可用性和复制
支持流复制、逻辑复制和同步复制，确保数据的高可用性和灾难恢复能力。

安全性
提供强大的安全功能，包括 SSL 加密、角色权限管理和数据加密。

跨平台支持
可以在 Linux、Windows、macOS 等多种操作系统上运行。

1.2.PostgreSQL 的优势

• 开源免费：PostgreSQL 是完全开源的，拥有活跃的社区支持和丰富的文档资源。
• 功能全面：支持复杂查询、存储过程、触发器、视图等高级功能。
• 高性能：通过优化查询计划和并发控制机制，能够高效处理大规模数据。
• 灵活性和可扩展性：支持多种编程语言（如 PL/pgSQL、PL/Python、PL/Perl 等）和扩展插件。
• 社区支持：拥有庞大的开发者社区，问题解决速度快，生态丰富。

1.3.PostgreSQL 的常见应用场景

1.3.1. Web 应用程序
适用于需要复杂事务处理和高效查询的 Web 应用。

1.3.2. 数据仓库和商业智能
支持大规模数据存储和分析，适合数据仓库和 BI 系统。

1.3.3. 地理信息系统（GIS）
通过 PostGIS 扩展，广泛应用于地理空间数据管理。

1.3.4. 科学研究
用于管理和分析科学实验中的大量数据。

1.3.5. 金融系统
因其 ACID 特性和高可靠性，常用于金融交易和账务系统。

1.4.如何开始使用 PostgreSQL

1.4.1. 安装 PostgreSQL
可以从 PostgreSQL 官方网站 下载并安装适合你操作系统的版本。

1.4.2. 使用命令行工具 psql
psql 是 PostgreSQL 的交互式命令行工具，用于执行 SQL 命令和管理数据库。

1.4.3. 使用图形化管理工具
推荐使用 pgAdmin（PostgreSQL 官方图形化管理工具）或其他第三方工具（如 DBeaver、DataGrip）。

1.5.常用命令示例

连接到数据库：

psql -U 用户名 -d 数据库名 -h 主机地址 -p 端口

创建数据库：

CREATE DATABASE 数据库名;

创建表：

CREATE TABLE 表名 (

id SERIAL PRIMARY KEY,

姓名 VARCHAR(100) NOT NULL,

年龄 INT

);

插入数据：

INSERT INTO 表名 (姓名, 年龄) VALUES ('张三', 25);

查询数据：

SELECT * FROM 表名 WHERE 年龄 > 20;

更新数据：

UPDATE 表名 SET 年龄 = 26 WHERE 姓名 = '张三';

删除数据：

DELETE FROM 表名 WHERE 姓名 = '张三';

1.6. 做一下小结

PostgreSQL 是一个功能强大且灵活的数据库系统，适用于各种规模和类型的应用。无论是开发小型应用还是构建大型企业级系统，PostgreSQL 都能提供可靠的数据管理解决方案。其开源特性、丰富的功能和活跃的社区支持，使其成为开发者和企业的首选数据库之一。

2. 下面看看 Python与PostgreSQL 的交互

以下是使用 Python 连接和操作 PostgreSQL 的详细步骤，包括安装驱动、基本操作和示例代码：

2.1. 安装 Python 的 PostgreSQL 驱动

Python 中常用的 PostgreSQL 驱动库是 psycopg2（或轻量版的 psycopg2-binary）。
也可以通过 SQLAlchemy（ORM 工具）操作 PostgreSQL。

安装驱动：

pip install psycopg2-binary # 简化版，适合快速开发

# 或

pip install sqlalchemy # ORM 工具（可选）

注意，在安装 psycopg2 或 psycopg2-binary 时，遇到 “ Error: pg_config executable not found.” 的错误，并且你确定已经成功安装了 PostgreSQL ，问题出在哪里？OK，不饶弯子，可能需要配置一下你的环境变量，简单的处理方式，以 macOS为例，在~/.bash_profile 或者 ~/.profile 文件中，添加一行

export PATH="/Library/PostgreSQL/15/bin/:$PATH"

重启终端环境，或者终端下执行，

source ~/.bash_profile

如此肯定可以搞定。

2.2. 连接 PostgreSQL 数据库

使用 psycopg2 连接数据库的基本流程：

示例代码：

import psycopg2

# 连接参数

conn_params = {

"host": "localhost", # 数据库主机地址

"port": 5432, # 默认端口 5432

"database": "mydatabase", # 数据库名

"user": "postgres", # 用户名

"password": "your_password" # 密码

}

try:

# 建立连接

conn = psycopg2.connect(**conn_params)

print("连接成功！")

# 创建游标（用于执行 SQL 命令）

cursor = conn.cursor()

# 执行 SQL 操作（见后续步骤）

except psycopg2.Error as e:

print(f"数据库连接失败: {e}")

finally:

# 关闭连接

if 'conn' in locals():

cursor.close()

conn.close()

print("连接已关闭。")

2.3. 基本操作示例

(1) 创建表

cursor.execute("""

CREATE TABLE IF NOT EXISTS users (

id SERIAL PRIMARY KEY,

name VARCHAR(100) NOT NULL,

email VARCHAR(100) UNIQUE,

age INT

);

""")

conn.commit() # 提交事务

print("表创建成功！")

(2) 插入数据

# 单条插入

cursor.execute("""

INSERT INTO users (name, email, age)

VALUES (%s, %s, %s);

""", ("张三", "zhangsan@example.com", 25))

# 批量插入

data = [

("李四", "lisi@example.com", 30),

("王五", "wangwu@example.com", 28)

]

cursor.executemany("""

INSERT INTO users (name, email, age)

VALUES (%s, %s, %s);

""", data)

conn.commit() # 提交事务

print("数据插入成功！")

(3) 查询数据

cursor.execute("SELECT * FROM users WHERE age > %s;", (26,))

rows = cursor.fetchall() # 获取所有结果

print("查询结果：")

for row in rows:

print(f"ID: {row[0]}, 姓名: {row[1]}, 年龄: {row[3]}")

(4) 更新数据

cursor.execute("""

UPDATE users

SET age = %s

WHERE name = %s;

""", (26, "张三"))

conn.commit()

print("数据更新成功！")

(5) 删除数据

cursor.execute("DELETE FROM users WHERE name = %s;", ("王五",))

conn.commit()

print("数据删除成功！")

2.4. 使用上下文管理器（推荐）

使用 with 语句自动管理连接和游标，避免资源泄漏：

try:

with psycopg2.connect(**conn_params) as conn:

with conn.cursor() as cursor:

cursor.execute("SELECT version();")

print("PostgreSQL 版本:", cursor.fetchone())

except psycopg2.Error as e:

print(f"操作失败: {e}")

2.5. 使用 SQLAlchemy（ORM 方式）

通过 ORM 操作数据库更面向对象：

示例代码：

from sqlalchemy import create_engine, Column, Integer, String

from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy.orm import sessionmaker

# 定义数据库连接

DATABASE_URL = "postgresql://postgres:your_password@localhost:5432/mydatabase"

engine = create_engine(DATABASE_URL)

Base = declarative_base()

# 定义数据模型

class User(Base):

__tablename__ = 'users'

id = Column(Integer, primary_key=True)

name = Column(String(100))

email = Column(String(100))

age = Column(Integer)

# 创建表（如果不存在）

Base.metadata.create_all(engine)

# 创建会话

Session = sessionmaker(bind=engine)

session = Session()

# 插入数据

new_user = User(name="赵六", email="zhaoliu@example.com", age=22)

session.add(new_user)

session.commit()

# 查询数据

users = session.query(User).filter(User.age > 20).all()

for user in users:

print(f"ID: {user.id}, 姓名: {user.name}")

session.close()

2.6. 常见问题解决

2.6.1. 连接失败

• 检查 PostgreSQL 服务是否启动。
• 确认 pg_hba.conf 文件允许远程连接（如果需要）。
• 确保用户名和密码正确。

2.6.2. 中文乱码

在连接参数中添加字符集：

conn_params = {

...,

"client_encoding": "UTF8"

}

2.6.3. 事务处理

• 执行写操作后必须 conn.commit()。
• 出错时用 conn.rollback() 回滚事务。

2.7. 做下一小结

• 直接操作：使用 psycopg2 适合需要精细控制 SQL 的场景。
• ORM 操作：使用 SQLAlchemy 适合面向对象开发，减少 SQL 编写。
• 始终注意 关闭连接 和 异常处理，避免资源泄漏。

参考文档：

psycopg2 官方文档（
https://www.psycopg.org/docs/）

SQLAlchemy 官方文档（
https://www.sqlalchemy.org）

3. python 在实际生产开发环境里使用PostgreSQL

在实际的软件生产环境中，Python 与 PostgreSQL 的结合需要遵循一系列最佳实践，以确保系统的 高性能、高可用性、安全性和可维护性。以下是关键注意事项和实用方案：

3.1. 连接管理与连接池

生产环境中频繁创建和销毁数据库连接会导致性能瓶颈，需使用连接池管理。

方案：

• 使用 psycopg2.pool 或第三方库（如 sqlalchemy 的连接池）：

from psycopg2 import pool

# 初始化连接池（根据负载调整大小）

connection_pool = pool.SimpleConnectionPool(

minconn=1,

maxconn=10,

host="localhost",

database="mydb",

user="postgres",

password="your_password"

)

# 从池中获取连接

conn = connection_pool.getconn()

cursor = conn.cursor()

cursor.execute("SELECT * FROM users;")

# 使用后归还连接

connection_pool.putconn(conn)

• SQLAlchemy 自动管理连接池：

from sqlalchemy import create_engine

# 配置连接池参数

engine = create_engine(

"postgresql://user:password@host/dbname",

pool_size=10,

max_overflow=5,

pool_timeout=30

)

3.2. 性能优化

(1) 查询优化

• 避免 SELECT *：仅查询需要的字段。
• 使用索引：通过 EXPLAIN ANALYZE 分析慢查询，添加合适的索引。
• 批量操作：使用 executemany 或批量插入工具（如 COPY 命令）。

(2) 异步支持

• 使用异步驱动（如 asyncpg 或 SQLAlchemy + async 模式）提升高并发性能：

import asyncpg

async def query_data():

conn = await asyncpg.connect("postgresql://user:password@host/dbname")

rows = await conn.fetch("SELECT * FROM users;")

await conn.close()

return rows

3.3. 安全措施

(1) 敏感信息管理

• 避免硬编码凭据：使用环境变量或密钥管理服务（如 AWS Secrets Manager）：

import os

from dotenv import load_dotenv

load_dotenv() # 从 .env 文件加载配置

DB_PASSWORD = os.getenv("DB_PASSWORD")

(2) 防 SQL 注入

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

# 正确做法

cursor.execute("SELECT * FROM users WHERE name = %s;", (user_input,))

(3) 网络与权限

限制数据库 IP 白名单，使用 SSL 加密连接：

conn = psycopg2.connect(

...,

sslmode="require"

)

3.4. 事务与错误处理

(1) 原子性事务

• 使用事务确保数据一致性，结合 try/except 处理异常：

try:

with conn.cursor() as cursor:

cursor.execute("INSERT INTO orders (...) VALUES (...);")

cursor.execute("UPDATE inventory SET stock = stock - 1;")

conn.commit() # 提交事务

except Exception as e:

conn.rollback() # 回滚事务

logger.error(f"事务失败: {e}")

(2) 重试机制

• 实现数据库操作的 重试逻辑（如网络闪断）：

from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(3))

def safe_query():

cursor.execute(...)

3.5. 维护与监控

(1) 数据库迁移

使用 Alembic（SQLAlchemy 的迁移工具）管理表结构变更：

# 初始化 Alembic

alembic init migrations

# 生成迁移脚本

alembic revision --autogenerate -m "Add user table"

# 执行迁移

alembic upgrade head

(2) 监控与日志

• 监控指标：使用 Prometheus + Grafana 监控 PostgreSQL 的 QPS、连接数、慢查询。
• 日志记录：记录所有数据库操作的日志（通过 Python 的 logging 模块）。

3.6. 高可用与灾备

• 主从复制：配置 PostgreSQL 流复制（Streaming Replication）实现读写分离。
• 备份策略：定期使用 pg_dump 或 WAL 归档备份数据。
• 云托管服务：考虑使用 AWS RDS、Azure Database for PostgreSQL 等托管服务，自动处理高可用和备份。

3.7. 使用 ORM 的最佳实践

• 合理使用 ORM：避免过度复杂的联表查询，必要时直接编写 SQL。
• 惰性加载与预加载：优化查询性能（如 SQLAlchemy 的 joinedload）：

from sqlalchemy.orm import joinedload

users = session.query(User).options(joinedload(User.orders)).all()

3.8. 测试与 CI/CD

• 单元测试：使用测试数据库（如 pytest + pytest-postgresql）隔离测试环境。
• 集成测试：在 CI/CD 流程中自动化运行数据库相关测试。

#yaml文件

# GitHub Actions 示例

- name: Run PostgreSQL Tests

services:

postgres:

image: postgres:14

env:

POSTGRES_PASSWORD: postgres

steps:

- run: pytest tests/

3.9.做一下小结

在生产环境中使用 Python 操作 PostgreSQL，需重点关注：

连接池管理防止资源耗尽。

性能优化（索引、异步、批量操作）。

安全性（防注入、SSL、密钥管理）。

事务与错误恢复确保数据一致性。

监控与自动化运维提升稳定性。

遵循这些实践，可以构建高效、可靠的企业级应用。推荐结合具体场景选择工具（如 ORM 或原生 SQL），并定期进行性能调优和安全审计。

到此为止，希望小伙伴们多实践，少空想。

爱学习的小伙伴，关注不迷路哟～