qidao123.com技术社区-IT企服评测·应用市场

标题: Python 连接和操作 PostgreSQL 数据库的详解 [打印本页]
作者: 海哥    时间: 2024-11-12 11:51
标题: Python 连接和操作 PostgreSQL 数据库的详解
引言

在当今信息化的期间，数据库已成为存储和管理数据的关键技能。PostgreSQL 是一种开源的对象关系型数据库管理系统（ORDBMS），以其强盛的功能和稳定性而广受欢迎。Python 作为一种高级编程语言，因其简洁易读的语法和丰富的库支持，成为了数据处置惩罚和数据库操作的理想选择。本文将详细先容如何使用 Python 连接和操作 PostgreSQL 数据库，包括环境搭建、连接数据库、执行 SQL 查询和更新操作，以及处置惩罚异常和事务管理等内容。

环境搭建

在开始之前，我们必要确保系统上已经安装了 PostgreSQL 数据库和 Python 环境。以下是安装步骤：
安装 PostgreSQL

在 Windows 上安装 PostgreSQL

在 Linux 上安装 PostgreSQL

  1. sudo apt-get update
  2. sudo apt-get install postgresql postgresql-contrib
复制代码
在 macOS 上安装 PostgreSQL

  1. brew install postgresql
复制代码
安装 Python 和相关库

确保系统上已经安装了 Python。然后使用 pip 安装 psycopg2 库，这是一个用于连接 PostgreSQL 数据库的 Python 扩展模块。
  1. pip install psycopg2
复制代码
连接数据库

连接数据库是进行数据库操作的第一步。以下是使用 Python 连接 PostgreSQL 数据库的基本步骤：
导入库

  1. import psycopg2
复制代码
创建连接

  1. try:
  2.     conn = psycopg2.connect(
  3.         host="localhost",
  4.         database="mydatabase",
  5.         user="myuser",
  6.         password="mypassword"
  7.     )
  8.     print("成功连接到数据库")
  9. except psycopg2.Error as e:
  10.     print(f"连接数据库失败: {e}")
复制代码
创建游标

游标用于执行 SQL 查询并获取结果。
  1. cur = conn.cursor()
复制代码
执行查询

  1. try:
  2.     cur.execute("SELECT version();")
  3.     db_version = cur.fetchone()
  4.     print(f"数据库版本: {db_version[0]}")
  5. except psycopg2.Error as e:
  6.     print(f"执行查询失败: {e}")
复制代码
关闭游标和连接

  1. cur.close()
  2. conn.close()
复制代码
执行 SQL 查询和更新操作

查询数据

查询数据是最常见的数据库操作之一。以下是一个简单的查询示例：
  1. try:
  2.     cur.execute("SELECT * FROM mytable;")
  3.     rows = cur.fetchall()
  4.     for row in rows:
  5.         print(row)
  6. except psycopg2.Error as e:
  7.     print(f"查询失败: {e}")
复制代码
插入数据

插入数据用于向数据库表中添加新记录。
  1. try:
  2.     cur.execute("INSERT INTO mytable (column1, column2) VALUES (%s, %s);", ("value1", "value2"))
  3.     conn.commit()
  4.     print("插入成功")
  5. except psycopg2.Error as e:
  6.     print(f"插入失败: {e}")
  7.     conn.rollback()
复制代码
更新数据

更新数据用于修改数据库表中的现有记录。
  1. try:
  2.     cur.execute("UPDATE mytable SET column1 = %s WHERE column2 = %s;", ("new_value1", "value2"))
  3.     conn.commit()
  4.     print("更新成功")
  5. except psycopg2.Error as e:
  6.     print(f"更新失败: {e}")
  7.     conn.rollback()
复制代码
删除数据

删除数据用于从数据库表中移除记录。
  1. try:
  2.     cur.execute("DELETE FROM mytable WHERE column1 = %s;", ("value1",))
  3.     conn.commit()
  4.     print("删除成功")
  5. except psycopg2.Error as e:
  6.     print(f"删除失败: {e}")
  7.     conn.rollback()
复制代码
处置惩罚异常

在数据库操作过程中，可能会碰到各种异常环境。为了确保程序的健壮性，我们必要捕获并处置惩罚这些异常。
捕获异常

  1. try:
  2.     # 数据库操作代码
  3. except psycopg2.Error as e:
  4.     print(f"数据库操作失败: {e}")
  5. finally:
  6.     if conn is not None:
  7.         conn.close()
复制代码
处置惩罚特定异常

有时我们必要处置惩罚特定范例的异常，比方连接异常或查询异常。
  1. try:
  2.     # 数据库操作代码
  3. except psycopg2.OperationalError as e:
  4.     print(f"连接或操作错误: {e}")
  5. except psycopg2.ProgrammingError as e:
  6.     print(f"SQL 语句错误: {e}")
复制代码
事务管理

事务是一组数据库操作，这些操作要么全部成功，要么全部失败。事务管理对于确保数据的同等性和完整性至关重要。
开启事务

  1. conn.autocommit = False
复制代码
提交事务

  1. try:
  2.     # 数据库操作代码
  3.     conn.commit()
  4.     print("事务提交成功")
  5. except psycopg2.Error as e:
  6.     conn.rollback()
  7.     print(f"事务提交失败: {e}")
复制代码
回滚事务

  1. try:
  2.     # 数据库操作代码
  3.     conn.commit()
  4. except psycopg2.Error as e:
  5.     conn.rollback()
  6.     print(f"事务回滚: {e}")
复制代码
使用上下文管理器

Python 的上下文管理器可以简化资源管理，特别是在处置惩罚数据库连接和游标时。
使用 with 语句管理连接

  1. try:
  2.     with psycopg2.connect(
  3.         host="localhost",
  4.         database="mydatabase",
  5.         user="myuser",
  6.         password="mypassword"
  7.     ) as conn:
  8.         with conn.cursor() as cur:
  9.             cur.execute("SELECT version();")
  10.             db_version = cur.fetchone()
  11.             print(f"数据库版本: {db_version[0]}")
  12. except psycopg2.Error as e:
  13.     print(f"连接或查询失败: {e}")
复制代码
使用 with 语句管理事务

  1. try:    with psycopg2.connect(        host="localhost",        database="mydatabase",        user="myuser",        password="mypassword"    ) as conn:        conn.autocommit = False
  2.         with conn.cursor() as cur:            cur.execute("INSERT INTO mytable (column1, column2) VALUES (%s, %s);", ("value1", "value2"))            conn.commit()            print("插入成功")except psycopg2.Error as e:    print(f"插入失败: {e}")
复制代码
高级功能

使用参数化查询

参数化查询可以有效防止 SQL 注入攻击，并提高查询性能。
  1. try:
  2.     with psycopg2.connect(
  3.         host="localhost",
  4.         database="mydatabase",
  5.         user="myuser",
  6.         password="mypassword"
  7.     ) as conn:
  8.         with conn.cursor() as cur:
  9.             cur.execute("SELECT * FROM mytable WHERE column1 = %s;", ("value1",))
  10.             rows = cur.fetchall()
  11.             for row in rows:
  12.                 print(row)
  13. except psycopg2.Error as e:
  14.     print(f"查询失败: {e}")
复制代码
使用批量操作

批量操作可以显著提高数据插入和更新的性能。
  1. try:
  2.     with psycopg2.connect(
  3.         host="localhost",
  4.         database="mydatabase",
  5.         user="myuser",
  6.         password="mypassword"
  7.     ) as conn:
  8.         with conn.cursor() as cur:
  9.             data = [("value1", "value2"), ("value3", "value4")]
  10.             cur.executemany("INSERT INTO mytable (column1, column2) VALUES (%s, %s);", data)
  11.             conn.commit()
  12.             print("批量插入成功")
  13. except psycopg2.Error as e:
  14.     print(f"批量插入失败: {e}")
复制代码
使用存储过程

存储过程是预编译的 SQL 代码块，可以在数据库中存储并重复调用。
  1. CREATE OR REPLACE FUNCTION get_user_by_id(user_id INT) RETURNS TABLE(id INT, name TEXT) AS $$
  2. BEGIN
  3.     RETURN QUERY SELECT id, name FROM users WHERE id = user_id;
  4. END;
  5. $$ LANGUAGE plpgsql;
复制代码
  1. try:
  2.     with psycopg2.connect(
  3.         host="localhost",
  4.         database="mydatabase",
  5.         user="myuser",
  6.         password="mypassword"
  7.     ) as conn:
  8.         with conn.cursor() as cur:
  9.             cur.callproc('get_user_by_id', [1])
  10.             rows = cur.fetchall()
  11.             for row in rows:
  12.                 print(row)
  13. except psycopg2.Error as e:
  14.     print(f"调用存储过程失败: {e}")
复制代码
性能优化

使用连接池

连接池可以减少连接数据库的开销，提高性能。
  1. from psycopg2 import pool
  3. try:
  4.     postgreSQL_pool = psycopg2.pool.SimpleConnectionPool(
  5.         1, 20,
  6.         host="localhost",
  7.         database="mydatabase",
  8.         user="myuser",
  9.         password="mypassword"
  10.     )
  11.     if postgreSQL_pool:
  12.         print("连接池创建成功")
  13. except psycopg2.Error as e:
  14.     print(f"连接池创建失败: {e}")
  16. # 获取连接
  17. conn = postgreSQL_pool.getconn()
  19. try:
  20.     with conn.cursor() as cur:
  21.         cur.execute("SELECT version();")
  22.         db_version = cur.fetchone()
  23.         print(f"数据库版本: {db_version[0]}")
  24. finally:
  25.     # 释放连接
  26.     postgreSQL_pool.putconn(conn)
复制代码
使用索引

索引可以显著提高查询性能，特别是在大数据集上。
  1. CREATE INDEX idx_column1 ON mytable(column1);
复制代码
使用批量提交

批量提交可以减少事务的开销，提高性能。
  1. try:    with psycopg2.connect(        host="localhost",        database="mydatabase",        user="myuser",        password="mypassword"    ) as conn:        conn.autocommit = False
  2.         with conn.cursor() as cur:            data = [("value1", "value2"), ("value3", "value4")]            for row in data:                cur.execute("INSERT INTO mytable (column1, column2) VALUES (%s, %s);", row)                if len(data) % 1000 == 0:                    conn.commit()                    print("批量提交成功")            conn.commit()            print("插入完成")except psycopg2.Error as e:    print(f"插入失败: {e}")    conn.rollback()
复制代码
案例分析

为了更好地明白如何使用 Python 连接和操作 PostgreSQL 数据库，我们将通过一个实际案例来进行演示。
案例配景

假设我们有一个简单的电子商务网站，必要管理用户信息和订单信息。我们将创建两个表：users 和 orders，并演示如何进行基本的增删改查操作。
创建表

  1. CREATE TABLE users (
  2.     id SERIAL PRIMARY KEY,
  3.     name TEXT NOT NULL,
  4.     email TEXT UNIQUE NOT NULL
  5. );
  7. CREATE TABLE orders (
  8.     id SERIAL PRIMARY KEY,
  9.     user_id INT NOT NULL,
  10.     amount DECIMAL(10, 2) NOT NULL,
  11.     FOREIGN KEY (user_id) REFERENCES users(id)
  12. );
复制代码
插入数据

  1. try:
  2.     with psycopg2.connect(
  3.         host="localhost",
  4.         database="mydatabase",
  5.         user="myuser",
  6.         password="mypassword"
  7.     ) as conn:
  8.         with conn.cursor() as cur:
  9.             users_data = [
  10.                 ("Alice", "alice@example.com"),
  11.                 ("Bob", "bob@example.com")
  12.             ]
  13.             cur.executemany("INSERT INTO users (name, email) VALUES (%s, %s);", users_data)
  14.             conn.commit()
  15.             print("用户数据插入成功")
  17.             orders_data = [
  18.                 (1, 100.00),
  19.                 (2, 200.00)
  20.             ]
  21.             cur.executemany("INSERT INTO orders (user_id, amount) VALUES (%s, %s);", orders_data)
  22.             conn.commit()
  23.             print("订单数据插入成功")
  24. except psycopg2.Error as e:
  25.     print(f"数据插入失败: {e}")
复制代码
查询数据

  1. try:
  2.     with psycopg2.connect(
  3.         host="localhost",
  4.         database="mydatabase",
  5.         user="myuser",
  6.         password="mypassword"
  7.     ) as conn:
  8.         with conn.cursor() as cur:
  9.             cur.execute("SELECT * FROM users;")
  10.             users = cur.fetchall()
  11.             print("用户数据:")
  12.             for user in users:
  13.                 print(user)
  15.             cur.execute("SELECT * FROM orders;")
  16.             orders = cur.fetchall()
  17.             print("订单数据:")
  18.             for order in orders:
  19.                 print(order)
  20. except psycopg2.Error as e:
  21.     print(f"数据查询失败: {e}")
复制代码
更新数据

  1. try:
  2.     with psycopg2.connect(
  3.         host="localhost",
  4.         database="mydatabase",
  5.         user="myuser",
  6.         password="mypassword"
  7.     ) as conn:
  8.         with conn.cursor() as cur:
  9.             cur.execute("UPDATE users SET email = %s WHERE name = %s;", ("alice_new@example.com", "Alice"))
  10.             conn.commit()
  11.             print("用户数据更新成功")
  12. except psycopg2.Error as e:
  13.     print(f"数据更新失败: {e}")
复制代码
删除数据

  1. try:
  2.     with psycopg2.connect(
  3.         host="localhost",
  4.         database="mydatabase",
  5.         user="myuser",
  6.         password="mypassword"
  7.     ) as conn:
  8.         with conn.cursor() as cur:
  9.             cur.execute("DELETE FROM orders WHERE user_id = %s;", (1,))
  10.             conn.commit()
  11.             print("订单数据删除成功")
  12. except psycopg2.Error as e:
  13.     print(f"数据删除失败: {e}")
复制代码
结论

通过本文的详细先容，我们学习了如何使用 Python 连接和操作 PostgreSQL 数据库。从环境搭建到高级功能的使用，再到性能优化和实际案例的分析，我们涵盖了数据库操作的各个方面。希望本文能为新手朋侪提供有代价的参考和引导，资助大家在 Python 和 PostgreSQL 的世界中探索更多的可能性。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。



欢迎光临 qidao123.com技术社区-IT企服评测·应用市场 (https://dis.qidao123.com/) Powered by Discuz! X3.4