pandas_operate.py

import pandas as pd
import asyncio

# 初始化 DataFrame
df = pd.DataFrame(columns=['id', 'name', 'value'])

# 模拟异步增删改查操作
async def add_row(name: str, value: float):
    global df
    new_id = len(df) + 1
    new_row = pd.DataFrame({'id': [new_id], 'name': [name], 'value': [value]})
    df = pd.concat([df, new_row], ignore_index=True)

async def delete_row(row_id: int):
    global df
    df = df[df['id'] != row_id].reset_index(drop=True)

async def update_row(row_id: int, name: str, value: float):
    global df
    df.loc[df['id'] == row_id, ['name', 'value']] = [name, value]

async def read_rows():
    global df
    return df

# 示例使用
async def main():
    await add_row("Item1", 10.5)
    await add_row("Item2", 20.5)
    print("After adding rows:\n", await read_rows())
    
    await update_row(1, "UpdatedItem1", 15.0)
    print("After updating row 1:\n", await read_rows())
    
    await delete_row(2)
    print("After deleting row 2:\n", await read_rows())

# 运行示例
asyncio.run(main())

pip.sh

pip install sqlmodel pandas aiomysql sqlalchemy

sql_model.py

from sqlmodel import SQLModel, Field
from typing import Optional

class Item(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    value: float

sql_operate.py

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from sqlmodel import select
import pandas as pd
import asyncio

DATABASE_URL = "mysql+aiomysql://user:password@localhost/testdb"

engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

async def init_db():
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

async def create_item(name: str, value: float):
    async with async_session() as session:
        item = Item(name=name, value=value)
        session.add(item)
        await session.commit()

async def read_items():
    async with async_session() as session:
        result = await session.execute(select(Item))
        return result.scalars().all()

async def update_item(item_id: int, name: str, value: float):
    async with async_session() as session:
        item = await session.get(Item, item_id)
        if item:
            item.name = name
            item.value = value
            await session.commit()

async def delete_item(item_id: int):
    async with async_session() as session:
        item = await session.get(Item, item_id)
        if item:
            await session.delete(item)
            await session.commit()

async def mysql_to_pandas():
    items = await read_items()
    df = pd.DataFrame([item.dict() for item in items])
    return df

async def pandas_to_mysql(df: pd.DataFrame):
    async with async_session() as session:
        for _, row in df.iterrows():
            item = Item(name=row['name'], value=row['value'])
            session.add(item)
        await session.commit()

async def main():
    await init_db()
    
    # 创建数据
    await create_item("Item1", 10.5)
    
    # 读取数据并打印
    items = await read_items()
    print("Items:", items)
    
    # 更新数据
    await update_item(1, "UpdatedItem", 20.5)
    
    # 删除数据
    await delete_item(1)
    
    # MySQL 到 Pandas
    df = await mysql_to_pandas()
    print("DataFrame:", df)
    
    # Pandas 到 MySQL
    data = {'name': ['NewItem1', 'NewItem2'], 'value': [30.0, 40.0]}
    df = pd.DataFrame(data)
    await pandas_to_mysql(df)

# 运行示例
asyncio.run(main())