Make pandas/io/sql.py work with sqlalchemy 2.0
- [x] closes #40686
- [ ] [Tests added and passed](https://pandas.pydata.org/pandas-docs/dev/development/contributing_codebase.html#writing-tests) if fixing a bug or adding a new feature
- [x] All [code checks passed](https://pandas.pydata.org/pandas-docs/dev/development/contributing_codebase.html#pre-commit).
- [ ] Added [type annotations](https://pandas.pydata.org/pandas-docs/dev/development/contributing_codebase.html#type-hints) to new arguments/methods/functions.
- [x] Added an entry in the latest `doc/source/whatsnew/vX.X.X.rst` file if fixing a bug or adding a new feature.

I have run the following with sqlalchemy 2.0.0 to confirm compatilibility:
* `pytest pandas/tests/io/test_sql.py`
* `mypy pandas\io\sql.py pandas\tests\io\test_sql.py --follow-imports=silent`
> I'm sorry I'm not so familiar with the SQLAlchemy changes, but do you mind explaining why we should filter `RemovedIn20Warnings` instead of updating the code? I assume we want to be compatible with both 1.4 and 2.0 which are not compatible. But feels like it'd make more sense to for now use `if` statements to support both versions.

My approach is to update pandas/io/sql.py so that it can support both versions.  I also had to update pandas/tests/io/test_sql.py, because pandas users will need run DataFrame.to_sql within a transaction if they pass a sqlalchemy Connection under sqlalchemy 2.0.  This is how I've approached the transition to sqlalchemy 2.0 in my own work: by continuing to use 1.4 while ensuring that my code would still work under 2.0.

The RemovedIn20Warning provides a way of testing that the code will work under sqlalchemy 2.0, while the tests run under sqlalchemy 1.4.  Another approach is to pass future=True to sqlalchemy.create_engine in order to create 2.0-style engines for testing.  I hadn't thought of expanding the tests to run under both 1.4 and 2.0 sqlalchemy engines and connections, but that might be a better approach to testing the code during the transition period, since it does not require the SQLALCHEMY_WARN_20 environment variable to be set prior to the start of the tests. Thanks a lot for all the information @cdcadman, that's useful. There is something I still don't fully understand. Let me explain in detail, and please correct me if I'm wrong, I can surely be missing something. Let me use this example:

 ```python
# 1.4
session.query(User).get(42)

# 2.0
session.get(User, 42)
```

I see two different cases.

**Case A: 2.0 syntax also works in 1.4**

This case is easy, we should simply replace one by the other, right?

**Case B: 2.0 syntax doesn't work on 1.4**

Then, while we want to keep compatibility with 1.4, we should have something like (code won't work, just to illustrate):

```python
if sqlalchemy.__version__ < '2.0':
    session.query(User).get(42)
else:
    session.get(User, 42)
```

When we test a pandas function, if we have all the code written this way, once we finished all the changes, we shouldn't have any warning being raised, so we wouldn't need to filter any warning in our code.


If we agree on this (please let me know if I missed anything), then the next question is how we make these changes. If I understand correctly, there is this flag you mention that makes alchemy start raising warnings about deprecations. I guess it can be activated locally, address the warnings, and then open PRs just with the fixes. Or we can activate it in our CI, so the warnings are visible in our tests. Either way is probably fine.

If the above is correct, then I don't understand why we need to filter warnings in our code. Do you mind explaining what I'm missing please? @datapythonista I agree with everything you wrote.  I don't think we should censor `RemovedIn20Warning`.  The only warnings filter I put into any code is this code in pandas/tests/io/test_sql.py:
```
if SQLALCHEMY_INSTALLED:
    # This only matters if the environment variable SQLALCHEMY_WARN_20 is set to 1.
    pytestmark = pytest.mark.filterwarnings("error::sqlalchemy.exc.RemovedIn20Warning")
```
This causes tests to fail if they raise RemovedIn20Warning.

Your comment about checking sqlalchemy versions reminded me that it is possible within sqlalchemy version 1.4 to obtain a 2.0-style sqlalchemy engine by passing `future=True` to `sqlalchemy.create_engine`.  Any situation which raises `RemovedIn20Warning` should fail if it runs with a 2.0-style engine.  Testing with 2.0-style engines is better than testing based on the `RemovedIn20Warning`.  For example, when I tried running the tests after adding `future=True` to `create_engine`, one of the tests froze, and to fix it I had to use the connection as a context manager.

So I'd like to redo this PR in the next few days.  I don't expect pandas/io/sql.py to change, but I will do something different with the test file:
* Don't change test code to eliminate `RemovedIn20Warning`, so that I can ensure the existing tests still pass with 1.x-style sqlalchemy engines (`future=False`).  That will provide evidence that downstream users' code won't break based on the edits I make to pandas/io/sql.py.
* Add tests with 2.0-style engines.  I can think of 3 different ways to do this:
  * Use `pytest.mark.parametrize("future", [True, False])` on existing tests and classes.  I will probably do this.
  * Create duplicate fixtures and classes for sqlalchemy 2.0 tests.
  * Create a duplicate test file test_sql_future.py. Update: pandas/io/sql.py will have to change significantly from how I currently have it in this PR. @datapythonista I modified this PR's code, title, and original comment, and it is again ready for review.  Since I am now testing 2.0-style sqlalchemy connectables instead of checking for warnings, I think this can close #40686.  Since I did make a lot of changes, would you prefer that I start a new PR instead? Is it known when SQLAlchemy 2.0 will be released?

If it comes before pandas 2.0 (late 2022/early 2023), maybe we can just bump the minimum version of SQLAlchemy to 2.0 and adopt the new syntax. I'm not in love in duplicating all the testing & having to support 2 versions of an optional dependency with different syntax.

xref: https://github.com/pandas-dev/pandas/issues/44823
 @mroeschke I haven't seen a release date for sqlalchemy 2.0.  Instead of duplicating all the tests, I could make the `future` argument of `create_engine` depend on either a global constant in test_sql.py or an environment variable.  The `future` argument will be supported in sqlalchemy 2.0 and required to be `True`.  I could also put this PR on hold until the beta release of sqlalchemy 2.0 comes out. The 2.0 beta is out.

https://github.com/sqlalchemy/sqlalchemy/issues/8631  I'm planning to make some changes to this PR.  Firstly, I noticed that pandas.io.sql.execute is documented, right above this line: https://pandas.pydata.org/docs/user_guide/io.html?highlight=sql%20execute#engine-connection-examples .  As it stands, my PR would make this return a context manager instead of a Results Iterable, and I don't think I need to make this change, so I will change it back.

I plan to make `SQLDatabase` accept only a SQLAlchemy `Connection` and not an `Engine`. I would change `pandasSQL_builder` into a generating function, decorated by `contextlib.contextmanager`, so that it can dispose of the `Engine` that is created if the connectable is a string.  A new argument, `need_txn` will be set to `True` by `to_sql`, and otherwise be `False`.  An advantage of this approach is that I can begin a transaction if the connectable is a `Connection` which is not already in a transaction.

```
@contextmanager
def pandasSQL_builder(
    con,
    schema: str | None = None,
    need_txn: bool = False,
) -> Iterator[SQLDatabase] | Iterator[SQLiteDatabase]:
    """
    Convenience function to return the correct PandasSQL subclass based on the
    provided parameters.  Also creates a sqlalchemy connection and transaction
    if necessary.
    """
    import sqlite3
    import warnings

    if isinstance(con, sqlite3.Connection) or con is None:
        yield SQLiteDatabase(con)
    else:
        sqlalchemy = import_optional_dependency("sqlalchemy", errors="ignore")

        if sqlalchemy is not None and isinstance(con, (str, sqlalchemy.engine.Connectable)):
            with _sqlalchemy_con(con, need_txn) as con:
                yield SQLDatabase(con, schema=schema)
        elif isinstance(con, str) and sqlalchemy is None:
            raise ImportError("Using URI string without sqlalchemy installed.")
        else:

            warnings.warn(
                "pandas only supports SQLAlchemy connectable (engine/connection) or "
                "database string URI or sqlite3 DBAPI2 connection. "
                "Other DBAPI2 objects are not tested. Please consider using SQLAlchemy.",
                UserWarning,
                stacklevel=find_stack_level(),
            )
            yield SQLiteDatabase(con)


@contextmanager
def _sqlalchemy_con(connectable, need_txn: bool):
    """Create a sqlalchemy connection and a transaction if necessary."""
    import sqlalchemy

    if isinstance(connectable, str):
        engine = sqlalchemy.create_engine(connectable)
        try:
            with engine.connect() as con:
                if need_txn:
                    with con.begin():
                        yield con
                else:
                    yield con
        finally:
            engine.dispose()
    elif isinstance(connectable, sqlalchemy.engine.Engine):
        with connectable.connect() as con:
            if need_txn:
                with con.begin():
                    yield con
            else:
                yield con
    else:
        if need_txn and not connectable.in_transaction():
            with connectable.begin():
                yield connectable
        else:
            yield connectable
```

In `test_sql.py`, I will take out the test classes which pass Engines to `SQLDatabase`, which will reduce the number of tests.

As I was looking over the tests, I noticed this interesting behavior related to transactions.  I like having the ability to rollback a `DataFrame.to_sql` call to help maintain data integrity, but if a `sqlite3.Connection` is used, pandas commits the transaction.  If a `sqlalchemy.engine.Connection` is used, then pandas does not commit the transaction.  Maybe this is fine, because someone who wants the sqlalchemy behavior with a sqlite database can just make a sqlalchemy connection to the database.  Here is an example script:

```
import sqlite3
from pandas import DataFrame
from sqlalchemy import create_engine

with sqlite3.connect(":memory:") as con:
    con.execute("create table test (A integer, B integer)")
    row_count = con.execute("insert into test values (2, 4), (5, 10)").rowcount
    if row_count > 1:
        con.rollback()
    print(con.execute("select count(*) from test").fetchall()[0][0]) # prints 0

with sqlite3.connect(":memory:") as con:
    con.execute("create table test (A integer, B integer)")
    row_count = DataFrame({'A': [2, 5], 'B': [4, 10]}).to_sql('test', con, if_exists='append', index=False)
    if row_count > 1:
        con.rollback() # does nothing, because pandas already committed the transaction.
    print(con.execute("select count(*) from test").fetchall()[0][0]) # prints 2
    
with create_engine("sqlite:///:memory:").connect() as con:
    with con.begin():
        con.exec_driver_sql("create table test (A integer, B integer)")
    try:
        with con.begin():
            row_count = DataFrame({'A': [2, 5], 'B': [4, 10]}).to_sql('test', con, if_exists='append', index=False)
            assert row_count < 2
    except AssertionError:
        pass
    print(con.execute("select count(*) from test").fetchall()[0][0]) # prints 0
``` @datapythonista @mroeschke Based on your feedback, I took out all the test duplication, and instead ran the tests with sqlalchemy 2.0.0b2 installed to ensure that this can close #40686.  This PR will allow pandas to work with sqlalchemy 1.4.16 (the documented minimum version) and higher, even after sqlalchemy 2.0 is released.  I found a note here (written 10/13/2022) on the timing of sqlalchemy 2.0: https://www.sqlalchemy.org/blog/2022/10/13/sqlalchemy-2.0.0b1-released/
>we will likely move from beta releases into release candidates as well, anticipating a 2.0 final release after some months.
 Thanks for your work on this @cdcadman. IMO I would still be partial on just supporting sqlalchemy 2.0 syntax/tests/min version when it becomes available. @mroeschke As a pandas/sqlalchemy user, it would be really helpful to get these changes into pandas sooner, so that I can get my code ready for sqlalchemy 2.0 sooner.  This PR is making the kind of changes envisioned in the first paragraph of the sqlalchemy 2.0 migration document: https://docs.sqlalchemy.org/en/14/changelog/migration_20.html#overview .

>The SQLAlchemy 2.0 transition presents itself in the SQLAlchemy 1.4 release as a series of steps that allow an application of any size or complexity to be migrated to SQLAlchemy 2.0 using a gradual, iterative process. Lessons learned from the Python 2 to Python 3 transition have inspired a system that intends to as great a degree as possible to not require any “breaking” changes, or any change that would need to be made universally or not at all.

It's understandable if you are concerned that these changes will break existing code running on sqlalchemy 1.4.  I made three types of changes to pandas/tests/io/test_sql.py:

- Changed a test involving a PostgreSQL timezone-aware data type which doesn't exist in Sqlite or MySQL.  This test was failing for me in the main branch, so I fixed it before making other changes.  It xfails against Sqlite and MySQL.
- I changed the private API within `pandas.io.sql`, and that broke many tests that were testing the private API.
- I changed some tests so that they can run in sqlalchemy 2.0, even though they ran fine in sqlalchemy 1.4.

Let me know if there's anything I can do to help get this merged sooner. Thanks @cdcadman. I misunderstood that sqlalchemy 1.4 already has 2.0 functionality, so we can adopt 2.0 syntax/functionality while pandas min version is 1.4.

>  I made three types of changes to pandas/tests/io/test_sql.py:

Could you split these into 3 separate PRs? It's difficult for me to determine which changes correspond with a certain objective. Generally, 1 PR targeting 1 type of change is easier for review. @mroeschke , I split this into two commits.  The tests pass after the first commit with sqlalchemy 1.4.44, but I had to modify `test_sql.py` to make it work with sqlalchemy 2.0.0b3. Unfortunately, I introduced a bug in #49531 .  Now iteration through the result of `read_sql_query` with a `chunksize` argument can fail, since the connection already closed.  I will create a new PR to fix that. Here is my new PR.  I will update this one after that is complete: #49967  @mroeschke, I found another bug in addition to the one I intend to resolve with #49967.  The second bug is with `pandas.io.sql.execute`.  If the caller passes a query that returns a resultset, then the connection might close before the caller can fetch the results.  I'm having trouble finding a good workaround.  Would it be ok to stop allowing either a sqlalchemy `Engine` or `str` to be passed to this method?  It is not a well-documented method.  The only mention I found of it is above this line: https://pandas.pydata.org/docs/user_guide/io.html?highlight=sql%20execute#engine-connection-examples. Another potential solution to the `pandas.io.sql.execute` bug is to use `freeze` to fetch all the results before cleaning up.  This is inefficient, and I will update the usage docs to point that out, but it will allow downstream code to continue working if it does not pull large amounts of data using `pandas.io.sql.execute`.  It will also require updating the minimum version of sqlalchemy to 1.4.45, which is not yet released: https://github.com/sqlalchemy/sqlalchemy/discussions/8962 Status update: I think this is ready, but it has two PRs in front of it (#50177 and #49967) and I'm still not getting the CI checks to all pass. There is one failing test unrelated to the changes I am making.  I welcome any suggestions:
```
pandas/tests/io/parser/common/test_file_buffer_url.py::test_file_descriptor_leak[c_high] - AssertionError: ([], [pconn(fd=17, family=<AddressFamily.AF_INET: 2>, type=<SocketKind.SOCK_STREAM: 1>, laddr=addr(ip='10.1.0.145', port=36288), raddr=addr(ip='140.82.112.4', port=443), status='ESTABLISHED')])
``` Quick question: Can you split this somehow (hard to review) or is this all interconnected? You can ignore the failure, this happens sometimes @phofl , the first commit is split off in a separate PR: #49967 . I could make a new PR for the second commit, as it is a separate item: Make PandasSQL.execute arguments more precise.  Some of my changes were to get mypy to work with the sqlalchemy 2.0 beta version - for example removing the import of `to_instance` from `sqlalchemy.types`. Ah got you, thx. Not necessary. Could you activate the warning here though? To check that ci passes. Set the env variable as in my pr I had turned off SQLALCHEMY_WARN_20.  I will turn it back on and debug these warnings. @phofl , the tests are now passing with SQLALCHEMY_WARN_20 set to 1. They also pass under sqlalchemy version 2.0.0rc1. SQLAlchemy 2.0 is now released. :-) Is this already included in a release, I just created this issue #51061 and would be very happy for a fix. @moritzmeister , I was hoping it could be included in pandas 1.5.3, but the decision was not to include it: https://github.com/pandas-dev/pandas/issues/49857#issuecomment-1374889050 .  For your issue, I am surrounding the query string with `sqlalchemy.text`. Thanks for the link @cdcadman, then I suppose it will end up in the 2.0 release. Yeah I also fell back to wrapping the text. I am suprised that the sqlalchemy version was not pinned to <2.0.0 on 1.5.3 in the meantime.