Add connectables with default pool and add test for if_exists=append.

Author: Diadochokinetic

pandas/tests/io/test_sql.py

@@ -312,7 +312,6 @@ def create_and_load_types_postgresql(conn, types_data: list[dict]):
 
 def create_and_load_types(conn, types_data: list[dict], dialect: str):
     from sqlalchemy import insert
-    from sqlalchemy.engine import Engine
 
     types = types_table_metadata(dialect)
 
@@ -338,7 +337,6 @@ def create_and_load_postgres_datetz(conn):
         Table,
         insert,
     )
-    from sqlalchemy.engine import Engine
 
     metadata = MetaData()
     datetz = Table("datetz", metadata, Column("DateColWithTz", DateTime(timezone=True)))
@@ -620,6 +618,22 @@ def mysql_pymysql_engine():
     engine.dispose()
 
 
+@pytest.fixture
+def mysql_pymysql_engine_default_pool():
+    sqlalchemy = pytest.importorskip("sqlalchemy")
+    pymysql = pytest.importorskip("pymysql")
+    engine = sqlalchemy.create_engine(
+        "mysql+pymysql://root@localhost:3306/pandas",
+        connect_args={"client_flag": pymysql.constants.CLIENT.MULTI_STATEMENTS},
+    )
+    yield engine
+    for view in get_all_views(engine):
+        drop_view(view, engine)
+    for tbl in get_all_tables(engine):
+        drop_table(tbl, engine)
+    engine.dispose()
+
+
 @pytest.fixture
 def mysql_pymysql_engine_iris(mysql_pymysql_engine, iris_path):
     create_and_load_iris(mysql_pymysql_engine, iris_path)
@@ -639,6 +653,12 @@ def mysql_pymysql_conn(mysql_pymysql_engine):
         yield conn
 
 
+@pytest.fixture
+def mysql_pymysql_conn_default_pool(mysql_pymysql_engine_default_pool):
+    with mysql_pymysql_engine_default_pool.connect() as conn:
+        yield conn
+
+
 @pytest.fixture
 def mysql_pymysql_conn_iris(mysql_pymysql_engine_iris):
     with mysql_pymysql_engine_iris.connect() as conn:
@@ -667,6 +687,21 @@ def postgresql_psycopg2_engine():
     engine.dispose()
 
 
+@pytest.fixture
+def postgresql_psycopg2_engine_default_pool():
+    sqlalchemy = pytest.importorskip("sqlalchemy")
+    pytest.importorskip("psycopg2")
+    engine = sqlalchemy.create_engine(
+        "postgresql+psycopg2://postgres:postgres@localhost:5432/pandas",
+    )
+    yield engine
+    for view in get_all_views(engine):
+        drop_view(view, engine)
+    for tbl in get_all_tables(engine):
+        drop_table(tbl, engine)
+    engine.dispose()
+
+
 @pytest.fixture
 def postgresql_psycopg2_engine_iris(postgresql_psycopg2_engine, iris_path):
     create_and_load_iris(postgresql_psycopg2_engine, iris_path)
@@ -686,6 +721,12 @@ def postgresql_psycopg2_conn(postgresql_psycopg2_engine):
         yield conn
 
 
+@pytest.fixture
+def postgresql_psycopg2_conn_default_pool(postgresql_psycopg2_engine_default_pool):
+    with postgresql_psycopg2_engine_default_pool.connect() as conn:
+        yield conn
+
+
 @pytest.fixture
 def postgresql_adbc_conn():
     pytest.importorskip("adbc_driver_postgresql")
@@ -768,12 +809,30 @@ def sqlite_engine(sqlite_str):
     engine.dispose()
 
 
+@pytest.fixture
+def sqlite_engine_default_pool(sqlite_str):
+    sqlalchemy = pytest.importorskip("sqlalchemy")
+    engine = sqlalchemy.create_engine(sqlite_str)
+    yield engine
+    for view in get_all_views(engine):
+        drop_view(view, engine)
+    for tbl in get_all_tables(engine):
+        drop_table(tbl, engine)
+    engine.dispose()
+
+
 @pytest.fixture
 def sqlite_conn(sqlite_engine):
     with sqlite_engine.connect() as conn:
         yield conn
 
 
+@pytest.fixture
+def sqlite_conn_default_pool(sqlite_engine_default_pool):
+    with sqlite_engine_default_pool.connect() as conn:
+        yield conn
+
+
 @pytest.fixture
 def sqlite_str_iris(sqlite_str, iris_path):
     sqlalchemy = pytest.importorskip("sqlalchemy")
@@ -900,6 +959,11 @@ def sqlite_buildin_types(sqlite_buildin, types_data):
     pytest.param("mysql_pymysql_conn", marks=pytest.mark.db),
 ]
 
+mysql_connectable_default_pool = [
+    pytest.param("mysql_pymysql_engine_default_pool", marks=pytest.mark.db),
+    pytest.param("mysql_pymysql_conn_default_pool", marks=pytest.mark.db),
+]
+
 mysql_connectable_iris = [
     pytest.param("mysql_pymysql_engine_iris", marks=pytest.mark.db),
     pytest.param("mysql_pymysql_conn_iris", marks=pytest.mark.db),
@@ -915,6 +979,11 @@ def sqlite_buildin_types(sqlite_buildin, types_data):
     pytest.param("postgresql_psycopg2_conn", marks=pytest.mark.db),
 ]
 
+postgresql_connectable_default_pool = [
+    pytest.param("postgresql_psycopg2_engine_default_pool", marks=pytest.mark.db),
+    pytest.param("postgresql_psycopg2_conn_default_pool", marks=pytest.mark.db),
+]
+
 postgresql_connectable_iris = [
     pytest.param("postgresql_psycopg2_engine_iris", marks=pytest.mark.db),
     pytest.param("postgresql_psycopg2_conn_iris", marks=pytest.mark.db),
@@ -931,6 +1000,11 @@ def sqlite_buildin_types(sqlite_buildin, types_data):
     "sqlite_str",
 ]
 
+sqlite_connectable_default_pool = [
+    "sqlite_engine_default_pool",
+    "sqlite_conn_default_pool",
+]
+
 sqlite_connectable_iris = [
     "sqlite_engine_iris",
     "sqlite_conn_iris",
@@ -945,6 +1019,12 @@ def sqlite_buildin_types(sqlite_buildin, types_data):
 
 sqlalchemy_connectable = mysql_connectable + postgresql_connectable + sqlite_connectable
 
+sqlalchemy_connectable_default_pool = (
+    mysql_connectable_default_pool
+    + postgresql_connectable_default_pool
+    + sqlite_connectable_default_pool
+)
+
 sqlalchemy_connectable_iris = (
     mysql_connectable_iris + postgresql_connectable_iris + sqlite_connectable_iris
 )
@@ -1135,7 +1215,6 @@ def test_read_iris_query_expression_with_parameter(conn, request):
     from sqlalchemy import (
         MetaData,
         Table,
-        create_engine,
         select,
     )
 
@@ -1250,7 +1329,6 @@ def test_read_procedure(conn, request):
     # GH 7324
     # Although it is more an api test, it is added to the
     # mysql tests as sqlite does not have stored procedures
-    from sqlalchemy.engine import Engine
 
     df = DataFrame({"a": [1, 2, 3], "b": [0.1, 0.2, 0.3]})
     df.to_sql(name="test_frame", con=conn, index=False)
@@ -1324,7 +1402,6 @@ def test_insertion_method_on_conflict_do_nothing(conn, request):
     conn = request.getfixturevalue(conn)
 
     from sqlalchemy.dialects.postgresql import insert
-    from sqlalchemy.engine import Engine
     from sqlalchemy.sql import text
 
     def insert_on_conflict(table, conn, keys, data_iter):
@@ -1406,7 +1483,6 @@ def test_insertion_method_on_conflict_update(conn, request):
     conn = request.getfixturevalue(conn)
 
     from sqlalchemy.dialects.mysql import insert
-    from sqlalchemy.engine import Engine
     from sqlalchemy.sql import text
 
     def insert_on_conflict(table, conn, keys, data_iter):
@@ -1458,7 +1534,6 @@ def test_read_view_postgres(conn, request):
     # GH 52969
     conn = request.getfixturevalue(conn)
 
-    from sqlalchemy.engine import Engine
     from sqlalchemy.sql import text
 
     table_name = f"group_{uuid.uuid4().hex}"
@@ -2389,7 +2464,6 @@ def test_read_sql_delegate(conn, request):
 
 def test_not_reflect_all_tables(sqlite_conn):
     conn = sqlite_conn
-    from sqlalchemy.engine import Engine
 
     # create invalid table
     query_list = [
@@ -3199,8 +3273,6 @@ def test_get_schema_create_table(conn, request, test_frame3):
 
     conn = request.getfixturevalue(conn)
 
-    from sqlalchemy.engine import Engine
-
     tbl = "test_get_schema_create_table"
     create_sql = sql.get_schema(test_frame3, tbl, con=conn)
     blank_test_df = test_frame3.iloc[:0]
@@ -3353,7 +3425,6 @@ def test_connectable_issue_example(conn, request):
 
     # This tests the example raised in issue
     # https://github.com/pandas-dev/pandas/issues/10104
-    from sqlalchemy.engine import Engine
 
     def test_select(connection):
         query = "SELECT test_foo_data FROM test_foo_data"
@@ -4356,7 +4427,7 @@ def test_xsqlite_if_exists(sqlite_buildin):
     drop_table(table_name, sqlite_buildin)
 
 
-@pytest.mark.parametrize("conn", sqlalchemy_connectable)
+@pytest.mark.parametrize("conn", sqlalchemy_connectable_default_pool)
 def test_exists_temporary_table(conn, test_frame1, request):
     conn = request.getfixturevalue(conn)
 
@@ -4375,56 +4446,38 @@ def test_exists_temporary_table(conn, test_frame1, request):
     assert True if table.exists() else False
 
 
-@pytest.mark.parametrize("conn", sqlalchemy_connectable)
+@pytest.mark.parametrize("conn", sqlalchemy_connectable_default_pool)
 def test_to_sql_temporary_table_replace(conn, test_frame1, request):
     conn = request.getfixturevalue(conn)
 
-    # some DBMS only allow temporary tables to exist within a connection, therefore
-    # we can only test for a connection and not all types of connectables.
-    if isinstance(conn, Engine):
-        con = conn.connect()
-    elif isinstance(conn, str):
-        con = create_engine(conn).connect()
-    else:
-        con = conn
-
     test_frame1.to_sql(
         name="test_frame1",
-        con=con,
+        con=conn,
         if_exists="fail",
         index=False,
         prefixes=["TEMPORARY"],
     )
 
     test_frame1.to_sql(
         name="test_frame1",
-        con=con,
+        con=conn,
         if_exists="replace",
         index=False,
         prefixes=["TEMPORARY"],
     )
 
-    df_test = pd.read_sql("SELECT * FROM test_frame1", con)
+    df_test = pd.read_sql("SELECT * FROM test_frame1", conn)
 
     assert_frame_equal(test_frame1, df_test)
 
 
-@pytest.mark.parametrize("conn", sqlalchemy_connectable)
+@pytest.mark.parametrize("conn", sqlalchemy_connectable_default_pool)
 def test_to_sql_temporary_table_fail(conn, test_frame1, request):
     conn = request.getfixturevalue(conn)
 
-    # some DBMS only allow temporary tables to exist within a connection, therefore
-    # we can only test for a connection and not all types of connectables.
-    if isinstance(conn, Engine):
-        con = conn.connect()
-    elif isinstance(conn, str):
-        con = create_engine(conn).connect()
-    else:
-        con = conn
-
     test_frame1.to_sql(
         name="test_frame1",
-        con=con,
+        con=conn,
         if_exists="fail",
         index=False,
         prefixes=["TEMPORARY"],
@@ -4433,8 +4486,37 @@ def test_to_sql_temporary_table_fail(conn, test_frame1, request):
     with pytest.raises(ValueError, match=r"Table 'test_frame1' already exists."):
         test_frame1.to_sql(
             name="test_frame1",
-            con=con,
+            con=conn,
             if_exists="fail",
             index=False,
             prefixes=["TEMPORARY"],
         )
+
+
+@pytest.mark.parametrize("conn", sqlalchemy_connectable_default_pool)
+def test_to_sql_temporary_table_append(conn, test_frame1, request):
+    conn = request.getfixturevalue(conn)
+
+    test_frame1.to_sql(
+        name="test_frame1",
+        con=conn,
+        if_exists="fail",
+        index=False,
+        prefixes=["TEMPORARY"],
+    )
+
+    test_frame1.to_sql(
+        name="test_frame1",
+        con=conn,
+        if_exists="append",
+        index=False,
+        prefixes=["TEMPORARY"],
+    )
+
+    df_test = pd.read_sql("SELECT * FROM test_frame1", conn)
+
+    df_true = concat([test_frame1, test_frame1], axis=0, ignore_index=True).reset_index(
+        drop=True
+    )
+
+    assert_frame_equal(df_true, df_test)