I'm getting a Duplicate Prepared Statement Error when using SQLAlchemy with asyncpg and pgbouncer

General
1

Error Details

{
  "status": "healthy",
  "database": "disconnected",
  "details": {
    "status": "disconnected",
    "error": "(sqlalchemy.dialects.postgresql.asyncpg.ProgrammingError) <class 'asyncpg.exceptions.DuplicatePreparedStatementError'>: prepared statement \"__asyncpg_stmt_1__\" already exists\nHINT: \nNOTE: pgbouncer with pool_mode set to \"transaction\" or\n\"statement\" does not support prepared statements properly.\nYou have two options:\n\n* if you are using pgbouncer for connection pooling to a\n single server, switch to the connection pool functionality\n provided by asyncpg, it is a much better option for this\n purpose;\n\n* if you have no option of avoiding the use of pgbouncer,\n then you can set statement_cache_size to 0 when creating\n the asyncpg connection object.\n\n[SQL: select pg_catalog.version()]\n(Background on this error at: https://sqlalche.me/e/20/f405)"
  }
}

Environment

  • Database: PostgreSQL (Supabase)
  • Connection Pooler: pgbouncer (transaction mode)
  • Python: 3.11
  • SQLAlchemy: 2.x (async)
  • Driver: asyncpg
  • Deployment: Heroku

Current Configuration

Here’s my current database configuration:

def get_ssl_args():
    """Get SSL arguments based on the database URL."""
    try:
        parsed_url = urlparse(settings.DATABASE_URL)
        
        if "supabase" in settings.DATABASE_URL:
            return {
                "ssl": "require",
                "server_settings": {
                    "application_name": "finance_advisor_agent",
                    "statement_timeout": "60000",
                    "idle_in_transaction_session_timeout": "60000",
                    "client_min_messages": "warning"
                },
                "statement_cache_size": 0,  # Disable prepared statements
            }
        # ... other configurations
    except Exception as e:
        logger.error(f"Error parsing database URL: {str(e)}")
        raise

engine = create_async_engine(
    settings.DATABASE_URL,
    echo=False,
    poolclass=AsyncAdaptedQueuePool,
    pool_size=5,
    max_overflow=10,
    pool_timeout=30,
    pool_recycle=1800,
    connect_args=get_ssl_args(),
    pool_pre_ping=True,
    execution_options={
        "compiled_cache": None,
        "isolation_level": "READ COMMITTED"
    },
    pool_reset_on_return='commit'
)

What I’ve Tried

  1. Set statement_cache_size=0 in connect_args to disable prepared statements
  2. Disabled compiled cache with "compiled_cache": None
  3. Enabled pool_pre_ping for connection health checks
  4. Set pool recycling to prevent stale connections

Questions

  1. Is my configuration correct for disabling prepared statements with pgbouncer?
  2. How can I fix this error?

The error still occurs occasionally even with statement_cache_size=0. Any insights on what might be causing this or additional configuration needed would be greatly appreciated.

Additional Context

  • I’m only facing this error after deploying my FastAPI to Heroku but it’s working locally.
  • Using Supabase’s built-in pgbouncer with default transaction pooling mode
2

Understanding the Issue

PgBouncer, when operating in transaction or statement pooling modes, does not support prepared statements properly. This is because:[pgbouncer.org+5stackoverflow.com+5magicstack.github.io+5]

  • Prepared statements are tied to specific database connections.
  • In transaction pooling mode, PgBouncer reuses connections across different transactions, leading to potential conflicts when a prepared statement is already cached for a previous transaction.[pganalyze.com]

This results in errors like:

prepared statement "__asyncpg_stmt_1__" already exists

Recommended Solutions

1. Disable Prepared Statements

Since you’re already setting statement_cache_size=0 in your connect_args, this should theoretically disable prepared statements. However, ensure that:

  • You’re not explicitly preparing statements elsewhere in your code using Connection.prepare().
  • Your ORM or database library isn’t implicitly using prepared statements.[stackoverflow.com+2magicstack.github.io+2supabase.com+2]

If you’re using SQLAlchemy, for instance, ensure that prepared_statement_cache_size is set to 0.

2. Switch PgBouncer to Session Pooling Mode

If disabling prepared statements doesn’t resolve the issue, consider switching PgBouncer’s pooling mode to session. In session mode:[magicstack.github.io+2stackoverflow.com+2magicstack.github.io+2]

  • Connections are dedicated to a single client session for the duration of the session.
  • Prepared statements are preserved across transactions within the same session.pgbouncer.org

To switch to session mode, modify your PgBouncer configuration (pgbouncer.ini):

[pgbouncer]
pool_mode = session

Alternatively, set the environment variable:

POOL_MODE=session

After making this change, restart PgBouncer to apply the new pooling mode.

3. Use Asyncpg’s Connection Pooling

Instead of relying on PgBouncer for connection pooling, consider using asyncpg’s built-in connection pooling. This approach is more compatible with asynchronous applications and can help avoid the issues associated with PgBouncer’s transaction pooling mode.

To implement asyncpg’s connection pooling:

import asyncpg
from asyncpg import create_pool

async def init():
    pool = await create_pool(
        dsn=settings.DATABASE_URL,
        min_size=5,
        max_size=10,
        max_queries=50000,
        max_inactive_connection_lifetime=300.0,
    )

This method manages connections more effectively in asynchronous environments.

Additional Considerations

  • Ensure Compatibility: Verify that all components of your stack (e.g., SQLAlchemy, asyncpg, PgBouncer) are compatible with each other, especially regarding prepared statement handling.
  • Monitor Performance: After implementing these changes, monitor your application’s performance and error logs to ensure that the issue is resolved and that no new issues have emerged.