本文记录在Python编程中操作PostgreSQL数据库的基本方法与实现代码，包括连接数据库、创建数据表、插入/更新/删除数据表中的记录、函数调用、存储过程调用及事务处理等。

获取数据库

首先把连接数据的信息写入 database.ini 文本文件中，内容如下：

[postgresql]

host=localhost

database=dbname

user=postgres

password=123456

然后定义一个读取database.ini文件的函数，代码如下：

import psycopg2
from configparser import ConfigParser

def load_config(filename='database.ini', section='postgresql'):
    parser = ConfigParser()
    parser.read(filename)

    # 获取数据库的设置参数
    config = {}
    if parser.has_section(section):
        params = parser.items(section)
        for param in params:
            config[param[0]] = param[1]
    else:
        raise Exception(' 在 {0} 文件中没有发现参数 {1}'.format(filename,section))

    return config

if __name__ == '__main__':
     config = load_config()
     print(config)

以上代码运行结果如下：

{'host': 'localhost', 'database': 'dbname', 'user': 'postgres', 'password': '123456'}

建立数据库连接

import psycopg2
from config import load_config
def connect(config):    
    try:       
        with psycopg2.connect(**config) as conn:
            print('已连接到 PostgreSQL server.')
            return conn
    except (Exception. psycopg2.DatabaseError ) as error:
        print(error)

if __name__ == '__main__':
    config = load_config()
    connect(config) 

创建数据表

import psycopg2
from config import load_config

def create_tables():
      commands = (
        """
        CREATE TABLE vendors (
            vendor_id SERIAL PRIMARY KEY,
            vendor_name VARCHAR(255) NOT NULL
        )
        """,
        """ CREATE TABLE parts (
                part_id SERIAL PRIMARY KEY,
                part_name VARCHAR(255) NOT NULL
                )
        """,
        """
        CREATE TABLE part_drawings (
                part_id INTEGER PRIMARY KEY,
                file_extension VARCHAR(5) NOT NULL,
                drawing_data BYTEA NOT NULL,
                FOREIGN KEY (part_id)
                REFERENCES parts (part_id)
                ON UPDATE CASCADE ON DELETE CASCADE
        )
        """,
        """
        CREATE TABLE vendor_parts (
                vendor_id INTEGER NOT NULL,
                part_id INTEGER NOT NULL,
                PRIMARY KEY (vendor_id , part_id),
                FOREIGN KEY (vendor_id)
                    REFERENCES vendors (vendor_id)
                    ON UPDATE CASCADE ON DELETE CASCADE,
                FOREIGN KEY (part_id)
                    REFERENCES parts (part_id)
                    ON UPDATE CASCADE ON DELETE CASCADE
        )
        """)
    try:
        config = load_config()
        with psycopg2.connect(**config) as conn:
            with conn.cursor() as cur:
                # execute the CREATE TABLE statement
                for command in commands:
                    cur.execute(command)
    except (Exception, psycopg2.DatabaseError) as error:
        print(error)

if __name__ == '__main__':
     create_tables()    

插入1条或多条记录到数据表

def insert_vendor(vendor_name):   
    sql = """INSERT INTO vendors(vendor_name)
             VALUES(%s) RETURNING vendor_id;"""
    
    vendor_id = None
    config = load_config()

    try:
        with  psycopg2.connect(**config) as conn:
            with  conn.cursor() as cur:
                # execute the INSERT statement
                cur.execute(sql, (vendor_name,))

                # get the generated id back                
                rows = cur.fetchone()
                if rows:
                    vendor_id = rows[0]
            
            conn.commit()
    except (Exception, psycopg2.DatabaseError) as error:
        print(error)    
    finally:
        return vendor_id


def insert_many_vendors(vendor_list):
    sql = "INSERT INTO vendors(vendor_name) VALUES(%s) RETURNING *"
    config = load_config()
    try:
        with  psycopg2.connect(**config) as conn:
            with  conn.cursor() as cur:
                # execute the INSERT statement
                cur.executemany(sql, vendor_list)

            # commit the changes to the database
            conn.commit()
    except (Exception, psycopg2.DatabaseError) as error:
        print(error) 


if __name__ == '__main__':
     insert_vendor("拓邦电子")

     insert_many_vendors([
         ('中兴通讯',),
         ('洪都航空',),
         ('招商银行',),
         ('九阳股份',)
     ])

函数调用

# import psycopg2
# from config import load_config

def get_parts(vendor_id):    
    parts = []
    # read database configuration
    params = load_config()
    try:
        # 连接数据库
        with  psycopg2.connect(**params) as conn:
            with conn.cursor() as cur:
                # create a cursor object for execution
                cur = conn.cursor()
                cur.callproc('get_parts_by_vendor', (vendor_id,))
                
                # 处理要插入的数据
                row = cur.fetchone()
                while row is not None:
                    parts.append(row)
                    row = cur.fetchone()

    except (Exception, psycopg2.DatabaseError) as error:
        print(error)
    finally:
        return parts

if __name__ == '__main__':
     parts = get_parts(1)
     print(parts)     

更新数据表中的记录

# import psycopg2
# from config import load_config

def update_vendor(vendor_id, vendor_name):
    updated_row_count = 0

    sql = """ UPDATE vendors
                SET vendor_name = %s
                WHERE vendor_id = %s"""
    
    config = load_config()
    
    try:
        with  psycopg2.connect(**config) as conn:
            with  conn.cursor() as cur:
                
                # 执行 UPDATE 语句
                cur.execute(sql, (vendor_name, vendor_id))
                updated_row_count = cur.rowcount

            # 提交更新数据到数据库
            conn.commit()
    except (Exception, psycopg2.DatabaseError) as error:
        print(error)    
    finally:
        return updated_row_count

if __name__ == '__main__':
     update_vendor(1, "拓邦股份")

调用存储过程

# import psycopg2
# from config import load_config

# 增加一个新的零部件
def add_part(part_name, vendor_name):
   
    # 读取数据库参数
    params = load_config()
    
    try:
        # 连接数据库
        with psycopg2.connect(**params) as conn:
            with conn.cursor() as cur:
                # 调用一个存储过过程
                cur.execute('CALL add_new_part(%s,%s)', (part_name, vendor_name))

            # 提交数据变化到数据库
            conn.commit()
    except (Exception, psycopg2.DatabaseError) as error:
        print(error)


if __name__ == '__main__':
     add_part('液晶板', '京东方')

事务处理

# import psycopg2
# from config import load_config

def add_part(part_name, vendor_list):
    # 给part数据表插入一条新记录
    insert_part = "INSERT INTO parts(part_name) VALUES(%s) RETURNING part_id;"

    # 给 vendor_parts 数据表插入一条新记录
    assign_vendor = "INSERT INTO vendor_parts(vendor_id,part_id) VALUES(%s,%s)"

    conn = None
    config = load_config()

    try:
        with psycopg2.connect(**config) as conn:
            with conn.cursor() as cur:
                # 增加一个新的零部件
                cur.execute(insert_part, (part_name,))

                # 获取该零部件的 id
                row = cur.fetchone()
                if row:
                    part_id = row[0]
                else:
                    raise Exception('没有该零件的 id')
                
                # 零部件与供应商对应，插入新记录
                for vendor_id in vendor_list:
                    cur.execute(assign_vendor, (vendor_id, part_id))

                # 提交事务
                conn.commit()
    except (Exception, psycopg2.DatabaseError) as error:
        if conn:
            conn.rollback()

        print(error)

if __name__ == '__main__':    
     # add_part('扬声器', (3, 4))
     # add_part('示波器', (5, 6))
     # add_part('天线', (6, 7))
     # add_part('按钮', (1, 5))
     # add_part('调制解调就', (1, 5))

     add_part('功率放大器', (99,))            

（本文完）