pgmanage 1.0.0

pgsql数据库工具

使用说明

本程序包用到的库：psycopg3 官方文档：https://www.psycopg.org/psycopg3/docs/advanced/pool.html#other-ways-to-create-a-pool

1. 下载安装

2. 导入包模块

3.使用示例

• 示例1

import Dbmanager

dbm = Dbmanager()
with dbm.pool.connection() as conn:
    with conn.cursor() as cur:
        cur.execute("SELECT 1")
        data = cur.fetchone()
        print(data)
# 使用完成后会自动关闭连接池

• 示例2

import Pgtool

# 设置日志
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

def main():

    # 生成10万行的测试数据
    data_to_bulk_insert = []
    for i in range(100000):
        data_to_bulk_insert.append({'id': f'{i}', 'name': f'name{i}88','name2': f'name{i}88', 'age': i % 100, 'email': f'email{i}@example.com','fasle': True})


    # 创建数据库管理器实例
    db_mgr = Pgtool()

    # 写入数据，主键冲突时更新所有非主键列 upsert_data
    rows_inserted = db_mgr.upsert_data(
        schema='data_test', 
        table='table_test', 
        data=data_to_bulk_insert,
        batch_size=10000,
        update=True,
        create=True,
        add_columns=True,
        max_workers=10,
    )
    logger.info(f"总共插入 {rows_inserted} 行")

    # 执行 SQL 查询
    query = "SELECT * FROM data_test.table_test LIMIT 10;"
    result = db_mgr.exec_query(query)
    for row in result:
        logger.info(row)
        
    # 删除数据
    query = "delete  FROM data_test.table_test;"
    result = db_mgr.exec_query_rowcount(query)
    logger.info(f"删除结果:{result}")

if __name__ == '__main__':
    main()