Metadata-Version: 2.3
Name: pgsqldatatool
Version: 1.0.3
Summary: Add your description here
Author: manji
Author-email: manji <pnsm@qq.com>
Requires-Dist: asyncpg>=0.31.0
Requires-Dist: polars>=1.41.2
Requires-Dist: python-dotenv>=1.2.2
Requires-Dist: sqlalchemy>=2.0.51
Requires-Dist: tzdata>=2026.2
Description-Content-Type: text/markdown

## 使用教程

### 安装
```python
pip install pgsqldatatool
pip install --upgrade pgsqldatatool
pip install --upgrade pgsqldatatool==版本号
```
### 工具列表

```python
# 清洗数据的工具
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
```
### 变量环境示例
```python
# 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（数据清洗）

``` python
# 基础清洗 （去掉列名两端空白字符、去掉数据两段空白字符、删除空行）
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
```
示例

```python
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）：彻底断开与数据库的所有连接，销毁这个池子。这通常只在整个程序/服务准备退出时才需要做。
```python
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()

```


```python

# 示例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 项目
```python
pip install asyncpg
```

```python
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())
```

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

```python
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())
```

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


### SQLAlchemy 2.0 + asyncpg（ORM 场景）
适合：需要 ORM、多数据库兼容

```python
pip install sqlalchemy[asyncio] asyncpg
```

```python
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()
```

