Python PostgreSQL实战：psycopg2深度解析-深圳市維司達科技有限公司

Python PostgreSQL实战：psycopg2深度解析

引言

在Python开发中，PostgreSQL是构建企业级数据库应用的核心技术。作为一名从Rust转向Python的后端开发者，我深刻体会到psycopg2在PostgreSQL操作方面的优势。psycopg2是Python生态中最流行的PostgreSQL客户端库，提供了完整的功能和良好的性能。

psycopg2核心概念

什么是psycopg2

psycopg2是PostgreSQL的Python客户端，具有以下特点：

高性能：优化的C扩展实现
完整功能：支持所有PostgreSQL特性
事务支持：支持ACID事务
批量操作：支持批量插入和更新
服务器端游标：支持大结果集

架构设计

┌─────────────────────────────────────────────────────────────┐ │ psycopg2 架构 │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │ 客户端API │───▶│ 连接层 │───▶│ PostgreSQL │ │ │ │ (Python) │ │ (psycopg2) │ │ 服务器 │ │ │ └──────────────┘ └──────────────┘ └──────────────┘ │ │ │ │ │ │ ▼ ▼ │ │ ┌──────────────────────────────────────────────────────┐ │ │ │ 协议层 │ │ │ └──────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────┘

环境搭建与基础配置

安装依赖

pip install psycopg2-binary

基本连接

import psycopg2 conn = psycopg2.connect( host="localhost", database="mydatabase", user="myuser", password="mypassword" ) cur = conn.cursor() cur.execute(''' CREATE TABLE IF NOT EXISTS users ( id SERIAL PRIMARY KEY, name VARCHAR(100) NOT NULL, email VARCHAR(100) NOT NULL UNIQUE ) ''') conn.commit() conn.close()

插入数据

import psycopg2 conn = psycopg2.connect( host="localhost", database="mydatabase", user="myuser", password="mypassword" ) cur = conn.cursor() user = ('张三', 'zhangsan@example.com') cur.execute('INSERT INTO users (name, email) VALUES (%s, %s)', user) conn.commit() conn.close()

高级特性实战

查询数据

import psycopg2 conn = psycopg2.connect( host="localhost", database="mydatabase", user="myuser", password="mypassword" ) cur = conn.cursor() cur.execute('SELECT * FROM users') users = cur.fetchall() for user in users: print(user) cur.execute('SELECT * FROM users WHERE id = %s', (1,)) user = cur.fetchone() print(user) conn.close()

更新数据

import psycopg2 conn = psycopg2.connect( host="localhost", database="mydatabase", user="myuser", password="mypassword" ) cur = conn.cursor() cur.execute('UPDATE users SET name = %s WHERE id = %s', ('李四', 1)) conn.commit() conn.close()

删除数据

import psycopg2 conn = psycopg2.connect( host="localhost", database="mydatabase", user="myuser", password="mypassword" ) cur = conn.cursor() cur.execute('DELETE FROM users WHERE id = %s', (1,)) conn.commit() conn.close()

实际业务场景

场景一：用户管理

import psycopg2 from psycopg2 import OperationalError class UserManager: def __init__(self, host, database, user, password): self.host = host self.database = database self.user = user self.password = password def _get_connection(self): return psycopg2.connect( host=self.host, database=self.database, user=self.user, password=self.password ) def create_user(self, name, email): conn = self._get_connection() cur = conn.cursor() try: cur.execute('INSERT INTO users (name, email) VALUES (%s, %s) RETURNING id', (name, email)) user_id = cur.fetchone()[0] conn.commit() return user_id finally: conn.close() def get_user(self, user_id): conn = self._get_connection() cur = conn.cursor() try: cur.execute('SELECT * FROM users WHERE id = %s', (user_id,)) return cur.fetchone() finally: conn.close() def get_all_users(self): conn = self._get_connection() cur = conn.cursor() try: cur.execute('SELECT * FROM users') return cur.fetchall() finally: conn.close()

场景二：批量插入

import psycopg2 def batch_insert_users(users): conn = psycopg2.connect( host="localhost", database="mydatabase", user="myuser", password="mypassword" ) cur = conn.cursor() try: cur.executemany( 'INSERT INTO users (name, email) VALUES (%s, %s)', users ) conn.commit() return True except Exception as e: conn.rollback() raise e finally: conn.close() users = [ ('张三', 'zhangsan@example.com'), ('李四', 'lisi@example.com'), ('王五', 'wangwu@example.com') ] batch_insert_users(users)

性能优化

使用服务器端游标

import psycopg2 from psycopg2 import extras conn = psycopg2.connect( host="localhost", database="mydatabase", user="myuser", password="mypassword" ) cur = conn.cursor('server_side_cursor') cur.itersize = 1000 cur.execute('SELECT * FROM large_table') for row in cur: print(row) conn.close()

使用COPY命令

import psycopg2 from io import StringIO conn = psycopg2.connect( host="localhost", database="mydatabase", user="myuser", password="mypassword" ) cur = conn.cursor() data = StringIO() data.write('张三\tzhangsan@example.com\n') data.write('李四\tlisi@example.com\n') data.seek(0) cur.copy_from(data, 'users', columns=('name', 'email')) conn.commit() conn.close()