I am trying to read and write to a db in the following way:

with open("argument_file.txt", "r") as f:
    arguments = f.read().split("\n")
engine = create_engine('postgresql://...', pool_size=20, max_overflow=0)
with futures.ProcessPoolExecutor(max_workers=20, initializer=lambda : engine.dispose(close=False)) as e:
    for result in e.map(main, arguments):

def main(argument):
    metadata = s.MetaData()
    events = Table('events', metadata, autoload_with=engine)
    query = select(events).where(...)  
    with engine.connect() as connection:
        ResultProxy = connection.execute(query)
        df = pd.DataFrame(ResultProxy.fetchall())
        df.columns = ResultProxy.keys()
    transformed_df = f(df)
    transformed_df.to_sql("table_name", engine, if_exists="append", index=False)
    return f"Argument {argument} processed"

(f is a function that maps a pd.DataFrame to a pd.DataFrame.) I am running into an error sqlalchemy.exc.OperationalError: (psycopg2.OperationalError) FATAL: sorry, too many clients already. When checking SELECT * FROM pg_stat_activity; I am seeing many old idle processes (the most I have seen is 96, which could explain that there is an issue with a pool size of 20 and a maximum of 115 connections in postgresql).

I suspect that there is an issue with closing old connections. But I think my code closes (either via context manager or via pd built-in code). What am I missing? (If this way to read/write is suboptimal in other ways, I'd also be very happy to hear that!)

  • You must think wrongly. There has to be a connection leak in your code. Aug 5 at 1:57


Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Browse other questions tagged or ask your own question.