| 3493 | |
| 3494 | @pytest.mark.parametrize("conn", sqlalchemy_connectable) |
| 3495 | def test_temporary_table(conn, request): |
| 3496 | if conn == "sqlite_str": |
| 3497 | pytest.skip("test does not work with str connection") |
| 3498 | |
| 3499 | conn = request.getfixturevalue(conn) |
| 3500 | |
| 3501 | from sqlalchemy import ( |
| 3502 | Column, |
| 3503 | Integer, |
| 3504 | Unicode, |
| 3505 | select, |
| 3506 | ) |
| 3507 | from sqlalchemy.orm import ( |
| 3508 | Session, |
| 3509 | declarative_base, |
| 3510 | ) |
| 3511 | |
| 3512 | test_data = "Hello, World!" |
| 3513 | expected = DataFrame({"spam": [test_data]}) |
| 3514 | Base = declarative_base() |
| 3515 | |
| 3516 | class Temporary(Base): |
| 3517 | __tablename__ = "temp_test" |
| 3518 | __table_args__ = {"prefixes": ["TEMPORARY"]} |
| 3519 | id = Column(Integer, primary_key=True) |
| 3520 | spam = Column(Unicode(30), nullable=False) |
| 3521 | |
| 3522 | with Session(conn) as session: |
| 3523 | with session.begin(): |
| 3524 | conn = session.connection() |
| 3525 | Temporary.__table__.create(conn) |
| 3526 | session.add(Temporary(spam=test_data)) |
| 3527 | session.flush() |
| 3528 | df = sql.read_sql_query(sql=select(Temporary.spam), con=conn) |
| 3529 | tm.assert_frame_equal(df, expected) |
| 3530 | |
| 3531 | |
| 3532 | @pytest.mark.parametrize("conn", all_connectable) |