前言

psycopg2是Python中最常用的用于连接PostgreSQL数据库的库。它提供了一个功能齐全的API来执行SQL语句，可以处理事务、获取查询结果、执行批量数据插入等操作。以下是关于如何使用psycopg2连接PostgreSQL的核心知识点和示例。

安装

要使用psycopg2，首先需要确保安装。可以使用以下命令通过pip安装：

pip install psycopg2

# 对于一些环境可能需要安装psycopg2-binary，以避免编译相关问题：
pip install psycopg2-binary

基本用法

psycopg2的基本用法包括连接到数据库、执行SQL查询、处理结果，以及进行事务管理。

示例代码

import psycopg2
from psycopg2 import sql

# 数据库连接参数
config = {
    'dbname': 'your_database',
    'user': 'your_username',
    'password': 'your_password',
    'host': 'localhost',
    'port': '5432'  # 默认端口
}

try:
    # 建立连接
    connection = psycopg2.connect(**config)
    
    # 创建游标对象
    cursor = connection.cursor()
    
    # 执行SQL查询
    cursor.execute("SELECT * FROM your_table WHERE status = %s", ('active',))
    
    # 获取结果
    results = cursor.fetchall()
    for row in results:
        print(row)
    
    # 提交事务（对于读操作可省略）
    connection.commit()

except psycopg2.Error as error:
    print("Error occurred:", error)

finally:
    # 关闭游标和连接
    if cursor:
        cursor.close()
    if connection:
        connection.close()

高级功能

上下文管理器：

使用with语句管理连接和游标更为方便和安全：

with psycopg2.connect(**config) as conn:
    with conn.cursor() as cur:
        cur.execute("SELECT * FROM your_table")
        results = cur.fetchall()
        print(results)

这样会自动处理事务提交，并在异常时自动回滚，同时明确地处理资源释放（关闭连接和游标）。

动态SQL构建：

使用psycopg2.sql模块构建动态SQL查询：

query = sql.SQL("SELECT * FROM {} WHERE {} = %s").format(
    sql.Identifier('your_table'),
    sql.Identifier('status')
)
cursor.execute(query, ('active',))

 

实际开发

#!/usr/bin/python
# -*- coding: UTF-8 -*-

import sys
import os
import logging
import psycopg2
import psycopg2.extras

script_path=os.path.dirname(os.path.realpath(__file__))
LOG_FORMAT = "%(asctime)s - %(levelname)s - %(pathname)s - %(funcName)s - %(lineno)d -msg: %(message)s"
logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
logging.info("script_path: {}".format(script_path))
logging.info('postgresql_connector init done ...')


class PGConnector(object):
    """
    postgresql api
    """

    def __init__(self, host, port, user, passwd, db=None, dictcurosr=True):
        """
        constroctor
        """
        self.host = host
        self.port = port
        self.user = user
        self.password = passwd
        self.db_name = db
        self.connect(dictcurosr)

        logging.info("connect to postgresql db %s:%d %s" % (host, port, db))
    
    def __del__(self):
        """
        del
        """
        if self.conn:
            self.conn.close()

    def connect(self, dictcurosr):
        """
        connect
        """
        self.conn = psycopg2.connect(database=self.db_name, 
                                        user=self.user,
                                        password=self.password, 
                                        host=self.host, 
                                        port=self.port, 
                                        keepalives=1, # 保持连接
                                        keepalives_idle=130, # 空闲时，每130秒保持连接连通
                                        keepalives_interval=10, # 没得到回应时，等待10秒重新尝试保持连通
                                        keepalives_count=15 # 尝试最多15次重新保持连通
                                        )
        if dictcurosr:
            self.cursor = self.conn.cursor(cursor_factory=psycopg2.extras.NamedTupleCursor)
            logging.info('PGConnector use DictCursor')
        else:
            self.cursor = self.conn.cursor()

    def sqlRead(self, sql):
        """
        read data
        """
        try:
            c = self.cursor.execute(sql)
            data = self.cursor.fetchall()
            return data
        except Exception as e:
            logging.error('sqlRead err ...')
            raise e

    def sqlWrite(self, sql):
        """
        write data
        """
        try:
            if isinstance(sql, list) or isinstance(sql, tuple):
                for item in sql:
                    self.cursor.execute(item)
            else:
                self.cursor.execute(sql)
            self.conn.commit()
        except Exception as e:
            self.conn.rollback()
            logging.error("execl sql: %s failed(%s). rollback!" % (sql, e))
            raise e
        return True

总结

• psycopg2是一个强大且灵活的库，用于与PostgreSQL数据库交互。
• 通过参数化查询和使用Python上下文管理器，可以编写更安全和高效的数据库代码。
• 对于更复杂的需求，比如批量操作和复杂事务管理，psycopg2提供了丰富的功能来支持。