15.2.PostgreSQL

import psycopg2

SQL 操作

create a connection

# Connect to an existing database
# key parameter format
conn = psycopg2.connect(host="localhost", port="5433",
                        user="postgres", password=password,
                        database="study")

# DATABASE_URL format
db_url = "postgresql://username:password@host:port/database_name"
conn = psycopg2.connect(db_url)

# another string format
string = "dbname=postgres user=postgres password=9464275 port=5433"
conn = psycopg2.connect(string)

open a cursor to execute SQL

# Open a cursor to perform database operations
cur = conn.cursor()

# Execute a command
cur.execute("SELECT version();")
# cur.executemany()

cur.fetchall()
# fetchone
# fetchmany

[('PostgreSQL 12.0, compiled by Visual C++ build 1914, 64-bit',)]

cur.execute("SELECT current_user;")
cur.fetchall()

[('postgres',)]

cur.execute("SELECT * FROM product;")
cur.fetchall()

[('0001', 'T恤', '衣服', 1000, 500, datetime.date(2009, 9, 20)),
 ('0002', '打孔器', '办公用品', 500, 320, datetime.date(2009, 9, 11)),
 ('0003', '运动T恤', '衣服', 4000, 2800, None),
 ('0004', '菜刀', '厨房用具', 3000, 2800, datetime.date(2009, 9, 20)),
 ('0005', '高压锅', '厨房用具', 6800, 5000, datetime.date(2009, 1, 15)),
 ('0006', '叉子', '厨房用具', 500, None, datetime.date(2009, 9, 20)),
 ('0007', '擦菜板', '厨房用具', 880, 790, datetime.date(2008, 4, 28)),
 ('0008', '圆珠笔', '办公用品', 100, None, datetime.date(2009, 11, 11)),
 ('0009', '平板电脑', '办公用品', 4000, 2500, datetime.date(2009, 11, 12)),
 ('0010', 'kindle', '办公用品', 500, 400, datetime.date(2009, 11, 15)),
 ('0011', 'ipad', '办公用品', 6500, 4000, datetime.date(2009, 12, 5))]

commit and close connection

# Make the changes to the database persistent
conn.commit()

# Close communication with the database
cur.close()
conn.close()

检查 postgresql 数据库是否可用

from psycopg2 import OperationalError as PsOperationalError


def check_postgresql_connection(host, username, password, database,
                                port="5432"):
    """
    Check for PostgreSQL database available.
    """
    db_url = f"postgresql://{username}:{password}@{host}:{port}/{database}"
    try:
        conn = psycopg2.connect(db_url)
        cur = conn.cursor()
        cur.execute("SELECT 1;")
        cur.fetchall()
        conn.commit()
        cur.close()
        conn.close()
        return "database available"
    except PsOperationalError as e:
        error_str = e.__str__()
        host_port_error_str_1 = \
            "could not connect to server: Connection refused"
        host_port_error_str_2 = \
            f'Is the server running on host "{host}" and accepting\n\t' \
            f'TCP/IP connections on port {port}?'
        authentication_error_str = \
            f'FATAL:  password authentication failed for user "{username}"'
        database_error_str = f'FATAL:  database "{database}" does not exist'

        if host_port_error_str_1 in error_str \
                and host_port_error_str_2 in error_str:
            return "Incorrect IP address or port"
        elif authentication_error_str in error_str:
            return "Incorrect username or password"
        elif database_error_str in error_str:
            return f"Database {database} does not exist"
        else:
            return "An error occurred while connecting to the database"
    except Exception as e:
        return "An error occurred while connecting to the database"