1. This site uses cookies. By continuing to use this site, you are agreeing to our use of cookies. Learn More.
  2. Anuncie Aqui
    Anuncie aqui você Também: fdantas@4each.com.br

[Python] Why does SQLAlchemy 2.x fail with Redshift due to standard_conforming_strings, and...

Discussão em 'Python' iniciado por Stack, Setembro 14, 2025.

  1. Stack

    Stack Membro Participativo

    We are integrating Amazon Redshift with SQLAlchemy and found an incompatibility when moving from SQLAlchemy 1.4.x to 2.x.

    Problem

    When using SQLAlchemy 2.x, our Redshift connections fail with:


    sqlalchemy.exc.ProgrammingError: (psycopg.errors.UndefinedObject) unrecognized configuration parameter "standard_conforming_strings"
    ​

    It looks like SQLAlchemy 2.x tries to check the standard_conforming_strings parameter at connection time.

    What we tried:

    • SQLAlchemy 1.4.49 → Works fine with Redshift.
    • SQLAlchemy 2.x with postgresql+psycopg2:// → Fails.
    • SQLAlchemy 2.x with redshift+redshift_connector:// → Still fails unless we downgrade back to 1.4.x

    Question

    • Is there a way to use SQLAlchemy 2.x with Redshift despite the lack of standard_conforming_strings? I want to use latest
    • Can I connect with to redshift postgresql with SQLAlchemy latest version and using psycopg (postgresql+psycopg) or (postgresql+psycopg)
    • Can I enable standard_conforming_strings on AWS RDS side so my code would work? I don't want to bypass it

    code:

    from sqlalchemy import create_engine, event, text

    # Database credentials and connection details
    engine = create_engine(
    f"postgresql+psycopg2://{db_user}:{db_password}@{db_host}:{db_port}/{db_name}"
    )

    # Set standard_conforming_strings=on if supported (PostgreSQL only, not Redshift)
    @event.listens_for(engine, "connect")
    def set_standard_conforming_strings(dbapi_connection, connection_record):
    try:
    cursor = dbapi_connection.cursor()
    cursor.execute("SET standard_conforming_strings = on;")
    cursor.close()
    except Exception:
    # Ignore if not supported (e.g., Redshift)
    pass

    # List of tables to query
    all_tables = []

    # Query row counts
    with engine.connect() as conn:
    for table in all_tables:
    try:
    query = text(f"SELECT COUNT(*) FROM {table}")
    result = conn.execute(query)
    count = result.scalar()
    print(f"Row count for {table}: {count}")
    except Exception as e:
    print(f"Error querying {table}: {e}")

    Continue reading...

Compartilhe esta Página