ENH: Added support for `AsyncEngine` and `AsyncConnection` when using pandas SQL functions/methods
[`pandas.read_sql`](https://pandas.pydata.org/pandas-docs/dev/reference/api/pandas.read_sql.html#pandas.read_sql), [`pandas.read_sql_query`](https://pandas.pydata.org/pandas-docs/dev/reference/api/pandas.read_sql_query.html#pandas.read_sql_query), [`pandas.read_sql_table`](https://pandas.pydata.org/pandas-docs/dev/reference/api/pandas.read_sql_table.html#pandas.read_sql_table), [`DataFrame.to_sql`](https://pandas.pydata.org/pandas-docs/dev/reference/api/pandas.DataFrame.to_sql.html#pandas.DataFrame.to_sql), and [`Series.to_sql`](https://pandas.pydata.org/pandas-docs/dev/reference/api/pandas.Series.to_sql.html#pandas.Series.to_sql) can now accept an [`AsyncEngine`](https://docs.sqlalchemy.org/en/latest/orm/extensions/asyncio.html#sqlalchemy.ext.asyncio.AsyncEngine) or an [`AsyncConnection`](https://docs.sqlalchemy.org/en/latest/orm/extensions/asyncio.html#sqlalchemy.ext.asyncio.AsyncConnection) as a valid argument to `con`.

Example Usage:

```py
import asyncio
from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine('postgresql+asyncpg://anakin:skywalker@localhost/test')

async def sql_operations():
  # Accessing a table with the AsyncEngine
  umbara = await pd.read_sql('umbara', engine)
  # Can alternatively be written as: await pd.read_sql_table('umbara', engine)

  # Using an SQL query with the AsyncConnection
  async with engine.connect() as conn:
     seasons = await pd.read_sql_query('SELECT season6, season5, season7 FROM clonewars', conn)
  
  favs = pd.DataFrame({
     'FirstName': ['Mace', 'Obi-Wan', 'Ahsoka'],
     'LastName': ['Windu', 'Kenobi', 'Tano']
  })
  
  # Writing data with the AsyncEngine
  await favs.to_sql('jedi', engine)

asyncio.run(sql_operations())
```

# Check list
- [x] tests added / passed (found a way to do this efficiently and relatively easily, will be pushing soon!)
- [X] Ensure all linting tests pass, see [here](https://pandas.pydata.org/pandas-docs/dev/development/contributing.html#code-standards) for how to run them
- [X] whatsnew entry

# Issue with unit tests
So unfortunately, as far as I can tell, creating unit tests will require that all the current unit tests for the SQL functions/methods will need to be rewritten, since every time the SQL functions/methods are invoked they need to be awaited.  One kind of unique idea I have is to read the source code of the existing test functions and add the word `await` whenever one of the SQL functions/methods is called and the use `exec` function to create the new `asyncio` version of the function.  Or perhaps there is a better approach I am not familiar with to avoid having to completely rewrite the unit tests for these new use cases.  I'm open to any ideas!

Thank you for considering my proposal! ♥
Hello @Ahsoka! Thanks for updating this PR. We checked the lines you've touched for [PEP 8](https://www.python.org/dev/peps/pep-0008) issues, and found:



* In the file [`pandas/io/sql.py`](https://github.com/pandas-dev/pandas/blob/25b70f3e6436fe78a1ca796607951c590e141349/pandas/io/sql.py):

> [Line 1218:17](https://github.com/pandas-dev/pandas/blob/25b70f3e6436fe78a1ca796607951c590e141349/pandas/io/sql.py#L1218): [E225](https://duckduckgo.com/?q=pep8%20E225) missing whitespace around operator
> [Line 2020:89](https://github.com/pandas-dev/pandas/blob/25b70f3e6436fe78a1ca796607951c590e141349/pandas/io/sql.py#L2020): [E501](https://duckduckgo.com/?q=pep8%20E501) line too long (98 > 88 characters)

* In the file [`pandas/tests/io/conftest.py`](https://github.com/pandas-dev/pandas/blob/25b70f3e6436fe78a1ca796607951c590e141349/pandas/tests/io/conftest.py):

> [Line 14:1](https://github.com/pandas-dev/pandas/blob/25b70f3e6436fe78a1ca796607951c590e141349/pandas/tests/io/conftest.py#L14): [E302](https://duckduckgo.com/?q=pep8%20E302) expected 2 blank lines, found 1

* In the file [`pandas/tests/io/test_sql.py`](https://github.com/pandas-dev/pandas/blob/25b70f3e6436fe78a1ca796607951c590e141349/pandas/tests/io/test_sql.py):

> [Line 1727:89](https://github.com/pandas-dev/pandas/blob/25b70f3e6436fe78a1ca796607951c590e141349/pandas/tests/io/test_sql.py#L1727): [E501](https://duckduckgo.com/?q=pep8%20E501) line too long (99 > 88 characters)
> [Line 1985:89](https://github.com/pandas-dev/pandas/blob/25b70f3e6436fe78a1ca796607951c590e141349/pandas/tests/io/test_sql.py#L1985): [E501](https://duckduckgo.com/?q=pep8%20E501) line too long (125 > 88 characters)
> [Line 2361:5](https://github.com/pandas-dev/pandas/blob/25b70f3e6436fe78a1ca796607951c590e141349/pandas/tests/io/test_sql.py#L2361): [E303](https://duckduckgo.com/?q=pep8%20E303) too many blank lines (2)
> [Line 2381:89](https://github.com/pandas-dev/pandas/blob/25b70f3e6436fe78a1ca796607951c590e141349/pandas/tests/io/test_sql.py#L2381): [E501](https://duckduckgo.com/?q=pep8%20E501) line too long (115 > 88 characters)
> [Line 2416:70](https://github.com/pandas-dev/pandas/blob/25b70f3e6436fe78a1ca796607951c590e141349/pandas/tests/io/test_sql.py#L2416): [E226](https://duckduckgo.com/?q=pep8%20E226) missing whitespace around arithmetic operator
> [Line 2422:89](https://github.com/pandas-dev/pandas/blob/25b70f3e6436fe78a1ca796607951c590e141349/pandas/tests/io/test_sql.py#L2422): [E501](https://duckduckgo.com/?q=pep8%20E501) line too long (137 > 88 characters)
> [Line 2427:89](https://github.com/pandas-dev/pandas/blob/25b70f3e6436fe78a1ca796607951c590e141349/pandas/tests/io/test_sql.py#L2427): [E501](https://duckduckgo.com/?q=pep8%20E501) line too long (116 > 88 characters)
> [Line 2429:89](https://github.com/pandas-dev/pandas/blob/25b70f3e6436fe78a1ca796607951c590e141349/pandas/tests/io/test_sql.py#L2429): [E501](https://duckduckgo.com/?q=pep8%20E501) line too long (100 > 88 characters)
> [Line 2586:89](https://github.com/pandas-dev/pandas/blob/25b70f3e6436fe78a1ca796607951c590e141349/pandas/tests/io/test_sql.py#L2586): [E501](https://duckduckgo.com/?q=pep8%20E501) line too long (92 > 88 characters)
> [Line 2593:89](https://github.com/pandas-dev/pandas/blob/25b70f3e6436fe78a1ca796607951c590e141349/pandas/tests/io/test_sql.py#L2593): [E501](https://duckduckgo.com/?q=pep8%20E501) line too long (93 > 88 characters)
> [Line 2606:89](https://github.com/pandas-dev/pandas/blob/25b70f3e6436fe78a1ca796607951c590e141349/pandas/tests/io/test_sql.py#L2606): [E501](https://duckduckgo.com/?q=pep8%20E501) line too long (95 > 88 characters)
> [Line 2637:89](https://github.com/pandas-dev/pandas/blob/25b70f3e6436fe78a1ca796607951c590e141349/pandas/tests/io/test_sql.py#L2637): [E501](https://duckduckgo.com/?q=pep8%20E501) line too long (91 > 88 characters)
> [Line 2789:89](https://github.com/pandas-dev/pandas/blob/25b70f3e6436fe78a1ca796607951c590e141349/pandas/tests/io/test_sql.py#L2789): [E501](https://duckduckgo.com/?q=pep8%20E501) line too long (92 > 88 characters)
> [Line 2810:1](https://github.com/pandas-dev/pandas/blob/25b70f3e6436fe78a1ca796607951c590e141349/pandas/tests/io/test_sql.py#L2810): [E302](https://duckduckgo.com/?q=pep8%20E302) expected 2 blank lines, found 1
> [Line 2811:89](https://github.com/pandas-dev/pandas/blob/25b70f3e6436fe78a1ca796607951c590e141349/pandas/tests/io/test_sql.py#L2811): [E501](https://duckduckgo.com/?q=pep8%20E501) line too long (99 > 88 characters)



##### Comment last updated at 2021-04-25 02:47:02 UTC > does this change the existing api at all?
> 
> absolutely needs tests

No this doesn't affect the existing API at all.  The sync SQLAlchemy engine is used the exact same way. All the existing arguments will work the same whether you are using a sync engine or an async engine.  The only difference is that you will need to await the function/method whenever you are using the async engine.

tl;dr This is **not a breaking change**.

I will keep brainstorming on how to efficiently create unit tests as ideally I would like to avoid having to rewrite the unit tests for this change.  Any ideas are appreciated.  Thank you! > haven't really looked at all. is there a pre-cursor PR you could do to refactor _first_ to setup what is needed . you are adding a lot here .

I could but the refactoring does not really make much sense on it's own.  It only makes sense in the context of this PR.  Also the refactor does not change the existing implementation.  It simply moves some aspects of the existing code into functions/methods to avoid having to copy and paste code into the new code.  Lmk if you're still interested in a separate PR for the refactoring first. @github-actions pre-commit > does this change the existing api at all?
> 
> absolutely needs tests

Hello @jreback

After much hard work I have implemented ALL unit tests related to this change, with a 100% pass rate, for SQLite.  I will be implementing PostgreSQL shortly. At the moment these are the only two databases that SQLAlchemy supports (with asyncio) so I will be unable to implement MySQL or any another database.

Also the syntax of the unit tests must be written in a certain way for them to automatically be converted to async.  This is why certain portions of the code in the unit tests are not PEP8 compliant. Thanks for the proposal and pull request, but unfortunately this PR has gone stale.

Additionally, this pull request, while valuable, is unfortunately too large to review in full. As mentioned in https://github.com/pandas-dev/pandas/pull/40654#pullrequestreview-623319887, it would be more reviewable to introduce smaller PRs first e.g. reorg code, set up internals, set this feature up for 1 method at a time. Closing for now, but feel free to introduce smaller PRs related to this feature. > Thanks for the proposal and pull request, but unfortunately this PR has gone stale.
> 
> Additionally, this pull request, while valuable, is unfortunately too large to review in full. As mentioned in [#40654 (review)](https://github.com/pandas-dev/pandas/pull/40654#pullrequestreview-623319887), it would be more reviewable to introduce smaller PRs first e.g. reorg code, set up internals, set this feature up for 1 method at a time. Closing for now, but feel free to introduce smaller PRs related to this feature.

Hey there, thanks for the feedback.  I will see what I can do revive this PR, however, I am quite busy with school at the moment and I am not sure how exactly to best split this up.  If I find some free time perhaps I will pick up this project but for now I must resign from this.

If anyone is interested in a potentially less ideal work-around the best solution at the moment is the following:
```py
from sqlalchemy.ext.asyncio import create_async_engine

import asyncio
import pandas

engine = create_async_engine('sqlite+aiosqlite:///:memory:')

async def sql_operations():
  favs = pandas.DataFrame({
     'FirstName': ['Mace', 'Obi-Wan', 'Ahsoka'],
     'LastName': ['Windu', 'Kenobi', 'Tano']
  })

  # NOTE: Read more about `run_sync` here:
  # https://docs.sqlalchemy.org/en/latest/orm/extensions/asyncio.html#running-synchronous-methods-and-functions-under-asyncio

  # Writing data with the AsyncEngine
  async with engine.begin() as conn:
    to_sql = lambda conn, name, *args, **kwargs: favs.to_sql(name, conn, *args, **kwargs)
    await conn.run_sync(to_sql, 'jedi')

  # Reading using AsyncEngine
  async with engine.connect() as conn:
    read_sql = lambda conn, name, *args, **kwargs: pandas.read_sql(name, conn, *args, **kwargs)
    jedi = await conn.run_sync(read_sql, 'jedi')

  print(jedi)

asyncio.run(sql_operations())
``` Perhaps someone could add this work-around to the docs as temporary less ideal solution until a proper solution is implemented. Hello @Ahsoka

Thank you for sharing your code.

I tried your code and it works well :).
But It only works when 'name' is table like 'jedi'.
It doesn't work with SQL string like 'select * from jedi'

How can I use SQL query string with your code?




> > Thanks for the proposal and pull request, but unfortunately this PR has gone stale.
> > Additionally, this pull request, while valuable, is unfortunately too large to review in full. As mentioned in [#40654 (review)](https://github.com/pandas-dev/pandas/pull/40654#pullrequestreview-623319887), it would be more reviewable to introduce smaller PRs first e.g. reorg code, set up internals, set this feature up for 1 method at a time. Closing for now, but feel free to introduce smaller PRs related to this feature.
> 
> Hey there, thanks for the feedback. I will see what I can do revive this PR, however, I am quite busy with school at the moment and I am not sure how exactly to best split this up. If I find some free time perhaps I will pick up this project but for now I must resign from this.
> 
> If anyone is interested in a potentially less ideal work-around the best solution at the moment is the following:
> 
> ```python
> from sqlalchemy.ext.asyncio import create_async_engine
> 
> import asyncio
> import pandas
> 
> engine = create_async_engine('sqlite+aiosqlite:///:memory:')
> 
> async def sql_operations():
>   favs = pandas.DataFrame({
>      'FirstName': ['Mace', 'Obi-Wan', 'Ahsoka'],
>      'LastName': ['Windu', 'Kenobi', 'Tano']
>   })
> 
>   # NOTE: Read more about `run_sync` here:
>   # https://docs.sqlalchemy.org/en/latest/orm/extensions/asyncio.html#running-synchronous-methods-and-functions-under-asyncio
> 
>   # Writing data with the AsyncEngine
>   async with engine.begin() as conn:
>     to_sql = lambda conn, name, *args, **kwargs: favs.to_sql(name, conn, *args, **kwargs)
>     await conn.run_sync(to_sql, 'jedi')
> 
>   # Reading using AsyncEngine
>   async with engine.connect() as conn:
>     read_sql = lambda conn, name, *args, **kwargs: pandas.read_sql(name, conn, *args, **kwargs)
>     jedi = await conn.run_sync(read_sql, 'jedi')
> 
>   print(jedi)
> 
> asyncio.run(sql_operations())
> ```

 > Hello @Ahsoka
>
> Thank you for sharing your code.
> 
> I tried your code and it works well :).
> But It only works when 'name' is table like 'jedi'.
> It doesn't work with SQL string like 'select * from jedi'
>
> How can I use SQL query string with your code?

Hello @sanghunka!

It is possible to do this.  In newer versions of SQLAlchemy, you cannot use a `str` object to execute queries.  You must wrap a `str` object in a `text` object first before the query can be executed.  See the code below for reference:
```py
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy import text

import asyncio
import pandas

engine = create_async_engine('sqlite+aiosqlite:///:memory:')

async def sql_operations():
    favs = pandas.DataFrame({
      'FirstName': ['Mace', 'Obi-Wan', 'Ahsoka'],
      'LastName': ['Windu', 'Kenobi', 'Tano']
    })

    # NOTE: Read more about `run_sync` here:
    # https://docs.sqlalchemy.org/en/latest/orm/extensions/asyncio.html#running-synchronous-methods-and-functions-under-asyncio

    # Writing data with the AsyncEngine
    async with engine.begin() as conn:
        to_sql = lambda conn, name, *args, **kwargs: favs.to_sql(name, conn, *args, **kwargs)
        await conn.run_sync(to_sql, 'jedi')

    # Reading using AsyncEngine
    async with engine.connect() as conn:
        read_sql = lambda conn, name, *args, **kwargs: pandas.read_sql(name, conn, *args, **kwargs)
        query = text("SELECT FirstName FROM jedi") # <- NOTE: You must wrap the str object with a text object
        jedi = await conn.run_sync(read_sql, query)

    print(jedi)

asyncio.run(sql_operations())
```

Do note, that you can also use other SQLAlchemy operators in this parameter.  If you have a [declarative mapping](https://docs.sqlalchemy.org/en/latest/orm/mapping_styles.html#declarative-mapping) created, then you can do something like:
```py
from sqlalchemy import select, Column, Integer, String
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm.decl_api import registry
from dataclasses import dataclass, field

import asyncio
import pandas

mapper = registry()


# Create the declarative mapping
@mapper.mapped
@dataclass
class Jedi:
    __tablename__ = 'jedi'
    __sa_dataclass_metadata_key__ = 'sa'

    # NOTE: Default primary keyname is index
    index: int = field(metadata={'sa': Column(Integer, primary_key=True)})
    first_name: str = field(default=None, metadata={'sa': Column('FirstName', String)})
    last_name: str = field(default=None, metadata={'sa': Column('LastName', String)})

engine = create_async_engine('sqlite+aiosqlite:///:memory:')

async def sql_operations():
    favs = pandas.DataFrame({
      'FirstName': ['Mace', 'Obi-Wan', 'Ahsoka'],
      'LastName': ['Windu', 'Kenobi', 'Tano']
    })

    # NOTE: Read more about `run_sync` here:
    # https://docs.sqlalchemy.org/en/latest/orm/extensions/asyncio.html#running-synchronous-methods-and-functions-under-asyncio

    # Writing data with the AsyncEngine
    async with engine.begin() as conn:
        to_sql = lambda conn, name, *args, **kwargs: favs.to_sql(name, conn, *args, **kwargs)
        await conn.run_sync(to_sql, 'jedi')

    # Reading using AsyncEngine
    async with engine.connect() as conn:
        read_sql = lambda conn, name, *args, **kwargs: pandas.read_sql(name, conn, *args, **kwargs)
        query = select(Jedi) # <- NOTE: Use select statement on our declarative mapping, does the same thing as "SELECT * FROM jedi"
      jedi = await conn.run_sync(read_sql, query)

    print(jedi)

asyncio.run(sql_operations())

``` I feel sad that this feature is not integrated. > I feel sad that this feature is not integrated.

Me too 😔 +1 Thank you so much @Ahsoka It is really helpful😇
I think this should have merged because Async/Await is becoming more and more popular.

It's sad that I can't develop this by myself... cuz I'm not good at python.
If anyone is going to try this again, please let me know. I'd be happy to help.