华为云 Python RDS For PostgreSQL 演示

1. 说明

本项目将演示通过psycopg2对PostgreSQL进行操作

1.1 环境准备

• python3.9.2
• Pycharm 2022+ / Visual Studio 2022+
• PostgreSql 13.0+

1.2 必备PyPI包

• psycopg2==2.9.3

1.3 数据库选型

• RDS For PostgreSQL(推荐)
  更多详情，请参考 RDS For PostgreSQL 成长地图

1.4 关键点

• src\postgresql.py

class PostgresqlConnect:

    def __new__(cls, *args, **kwargs):
        # 获取配置文件路径
        dir_path = os.path.split(os.path.split(__file__)[0])[0]
        config_path = os.path.join(dir_path, r"config\config.json")
        with open(config_path, "r") as config_file:
            # 反序列化配置文件
            config = json.load(config_file)
        pgsql_address = config.get("pgsql", None)
        if not pgsql_address:
            raise Exception("连接地址不存在或为空")
        return psycopg2.connect(**pgsql_address)

• test\postgresql_test.py 对PostgreSQL进行增删改查测试

import unittest

from RDS4PostgreSQL.src.postgresql import PostgresqlConnect


class PostgresqlTest(unittest.TestCase):
    conn = None
    cursor = None

    @classmethod
    def setUpClass(cls) -> None:
        cls.conn = PostgresqlConnect()
        cls.cursor = cls.conn.cursor()
        create_sql = "create table tmp_blogs (id varchar(20), user_id varchar(20), name varchar(20))"
        cls.cursor.execute(create_sql)
        cls.conn.commit()

    def test_0_insert(self):
        """测试插入数据"""
        print("插入数据")
        insert_sql = "insert into tmp_blogs values('test_id', 'test_user_id', 'test_name')"
        self.cursor.execute(insert_sql)
        self.conn.commit()

        self.cursor.execute("select name from tmp_blogs")
        name = self.cursor.fetchall()[0][0]
        self.assertEqual(name, "test_name")

    def test_1_update(self):
        """测试更新数据"""
        print("更新数据")
        update_sql = "update tmp_blogs set name = 'update_name' where id = 'test_id'"
        self.cursor.execute(update_sql)
        self.conn.commit()

        self.cursor.execute("select name from tmp_blogs")
        name = self.cursor.fetchall()[0][0]
        self.assertEqual(name, "update_name")

    def test_2_delete(self):
        """测试删除数据"""
        print("删除数据")
        delete_sql = "delete from tmp_blogs where id = 'test_id'"
        self.cursor.execute(delete_sql)
        self.conn.commit()

        self.cursor.execute("select name from tmp_blogs")
        name = self.cursor.fetchall()
        self.assertListEqual(name, [])

    @classmethod
    def tearDownClass(cls) -> None:
        drop_sql = "drop table if exists tmp_blogs"
        cls.cursor.execute(drop_sql)
        cls.conn.commit()
        cls.cursor.close()
        cls.conn.close()


if __name__ == '__main__':
    unittest.main()

• config\config.json

{
  "pgsql": {
        "host": "ip_address",
        "port": "port",
        "user": "username",
        "password": "password",
        "database": "db_name"
  }
}

ip_address: 数据库IP地址
port: 数据库端口 username: 用户名
password: 用户密码
db_name: 数据库名