QST: How can I use df.to_sql() with SQLAlchemy AsyncSession/AsyncEgine

[lamhktommy]

Research

• I have searched the [pandas] tag on StackOverflow for similar questions.

• I have asked my usage related question on StackOverflow.

Link to question on StackOverflow

https://stackoverflow.com/questions/75566585/how-can-i-use-df-to-sql-with-sqlalchemy-asyncsession-asyncegine

Question about pandas

df.to_sql("temp_data", engine)

Expecting similar things where engine is AsyncEngine

[phi-friday]

pandas support for sqlachemy is implemented based on a fairly old version. Therefore, it is not completely compatible with recently added features. However, you can try the following ways:

1. use to_sql

from __future__ import annotations

from asyncio import current_task, run
from contextlib import asynccontextmanager
from typing import TYPE_CHECKING

import pandas as pd
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_scoped_session,
    create_async_engine,
)
from sqlalchemy.ext.automap import AutomapBase, automap_base
from sqlalchemy.orm import sessionmaker

if TYPE_CHECKING:
    from collections.abc import AsyncGenerator

automap: AutomapBase = automap_base()
engine = create_async_engine(...)
maker = sessionmaker(engine, class_=AsyncSession)
scope = async_scoped_session(maker, current_task)

chunk_size: int = ...
tablename: str = ...
data = pd.DataFrame(...)


@asynccontextmanager
async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with scope() as session:
        yield session


async def main():
    async with get_session() as session:
        conn = await session.connection()
        await conn.run_sync(
            lambda sync_conn: data.to_sql(
                tablename,
                con=sync_conn,
                ...,
            ),
        )
        await session.commit()


if __name__ == "__main__":
    run(main())

2. use execute

from __future__ import annotations

from asyncio import current_task, run
from contextlib import asynccontextmanager
from typing import TYPE_CHECKING

import numpy as np
import pandas as pd
from sqlalchemy import insert
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_scoped_session,
    create_async_engine,
)
from sqlalchemy.ext.automap import AutomapBase, automap_base
from sqlalchemy.orm import sessionmaker

if TYPE_CHECKING:
    from collections.abc import AsyncGenerator

automap: AutomapBase = automap_base()
engine = create_async_engine(...)
maker = sessionmaker(engine, class_=AsyncSession)
scope = async_scoped_session(maker, current_task)

chunk_size: int = ...
tablename: str = ...
data = pd.DataFrame(...)


@asynccontextmanager
async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with scope() as session:
        yield session


async def main():
    async with get_session() as session:
        conn = await session.connection()
        await conn.run_sync(lambda sync_conn: automap.prepare(autoload_with=sync_conn))
        mapper = getattr(automap.classes, tablename)
        stmt = insert(mapper)

        for _, group in data.groupby(np.arange(data.shape[0], dtype=int) // chunk_size):
            await session.execute(stmt, group.to_dict("records"))
        await session.commit()


if __name__ == "__main__":
    run(main())

Using to_sql is a familiar method, but sqlalchemy>2 is not available due to a bug #51015.

[lamhktommy]

@phi-friday thanks. It works for me

[ryanmerolle]

Would this still be the case with the updates to the project dependencies over the past year? Thanks again for all the work here!