pgsqldatatool 1.0.3

Add your description here

使用教程

安装

pip install pgsqldatatool
pip install --upgrade pgsqldatatool
pip install --upgrade pgsqldatatool==版本号

工具列表

# 清洗数据的工具
from pgsqldatatool import data_clean as dc

# 数据库连接池工具
from pgsqldatatool import PoolSingleton

# 将数据写入pgsql的工具
from pgsqldatatool import write_pg

# 将查询的数据库结果转换为polars DataFrame的工具
from pgsqldatatool import records_to_df

变量环境示例

# pgsql
PG_HOST = '127.0.0.1'
PG_PORT = 5432
PG_USER = 'postgres'
PG_PASSWORD = '1dd66mima'
PG_DB = 'dbname'
PG_MIN_SIZE = 10
PG_MAX_SIZE = 100

date_clean（数据清洗）

# 基础清洗 （去掉列名两端空白字符、去掉数据两段空白字符、删除空行）
lf_basic_clean(df)

# 根据字典重命名列
lf_rename_cols

# 移除重复行
lf_remove_dup_rows

# 删除指定列
lf_remove_cols

# 移除数字千分号
lf_remove_per_mille

# 移除数字百分号
lf_remove_percent

# 添加时间列
lf_add_time

# 删除完全为空的列 -- 这一步不是惰性计算，可能会降低性能
df_drop_empty_cols

示例

from pgsqldatatool import data_clean as dc
df = pl.read_excel(r"D:\manji\Downloads\判断中国域名.xlsx")
df = dc.lf_basic_clean(df)
df = dc.lf_remove_cols(df,["域名22"])
df = dc.lf_remove_dup_rows(df)
df = dc.df_drop_empty_cols(df)
df = dc.lf_remove_dup_rows(df)
df = df.collect()
print(df)

PoolSingleton（连接池单例模式）

归还连接（Release）：把连接还给池子，让别的任务继续用。（async with 已经帮你自动做了） 关闭连接池（Close）：彻底断开与数据库的所有连接，销毁这个池子。这通常只在整个程序/服务准备退出时才需要做。

from pgsqldatatool import PoolSingleton

# 示例1：使用异步连接池
async def main_test_1():

    # 调用方式1，
    async with  PoolSingleton.acquire() as conn:
        records = await conn.fetch(""" SELECT * FROM public."test20260211" """)
        print(records)

    # 销毁整个连接池
    await PoolSingleton.close()

# 示例2：使用静态方法
async def main_test_2():
    # 调用方式2：使用静态方法
    records2 = await PoolSingleton.fetch(""" SELECT * FROM public."test20260211" """)
    print(records2)

    # 调用方式3：使用静态方法
    records3 = await PoolSingleton.fetchrow(""" SELECT * FROM public."test20260211" """)
    print(records3)

    # 调用方式4：使用静态方法
    records4 = await PoolSingleton.execute(""" SELECT * FROM public."test20260211" """)
    print(records4)
    
    
if __name__ == "__main__":
    asyncio.run(main_test_1())

创建异步数据库链接

推荐方案：asyncpg（性能最好，原生 asyncio）

适合：FastAPI / aiohttp / asyncio 项目

pip install asyncpg

import asyncpg
import asyncio

async def main():
    # 创建连接
    conn = await asyncpg.connect(
        host="localhost",
        port=5432,
        user="postgres",
        password="password",
        database="testdb"
    )

    # 查询
    row = await conn.fetchrow("SELECT NOW()")
    print(row)

    # 参数化查询
    rows = await conn.fetch(
        "SELECT * FROM users WHERE age > $1",
        18
    )

    await conn.close()

asyncio.run(main())

使用连接池（生产必选 ✅）

import asyncpg
import asyncio

async def get_pool():
    return await asyncpg.create_pool(
        dsn="postgresql://postgres:password@localhost:5432/testdb",
        min_size=5,
        max_size=20
    )

async def main():
    pool = await get_pool()

    async with pool.acquire() as conn:
        result = await conn.fetch("SELECT * FROM users")

    await pool.close()

asyncio.run(main())

事务

async with conn.transaction():
    await conn.execute(
        "INSERT INTO users(name) VALUES($1)",
        "Alice"
    )

SQLAlchemy 2.0 + asyncpg（ORM 场景）

适合：需要 ORM、多数据库兼容

pip install sqlalchemy[asyncio] asyncpg

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy import select

engine = create_async_engine(
    "postgresql+asyncpg://postgres:password@localhost/testdb",
    pool_size=10,
    max_overflow=20
)

AsyncSessionLocal = sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

async def query_users():
    async with AsyncSessionLocal() as session:
        result = await session.execute(select(User))
        return result.scalars().all()