Connection problems with SQLAlchemy and multiple processes

by

Quoting “How do I use engines / connections / sessions with Python multiprocessing, or os.fork()?” with added emphasis:

The SQLAlchemy Engine object refers to a connection pool of existing database connections. So when this object is replicated to a child process, the goal is to ensure that no database connections are carried over.

and

However, for the case of a transaction-active Session or Connection being shared, there’s no automatic fix for this; an application needs to ensure a new child process only initiate new Connection objects and transactions, as well as ORM Session objects.

The issue stems from the forked child process inheriting the live global session, which is holding on to a Connection. When target calls init, it overwrites the global references to engine and session, thus decreasing their refcounts to 0 in the child, forcing them to finalize. If you for example one way or another create another reference to the inherited session in the child, you prevent it from being cleaned up – but don’t do that. After main has joined and returns to business as usual it is trying to use the now potentially finalized – or otherwise out of sync – connection. As to why this causes an error only after some amount of iterations I’m not sure.

The only way to handle this situation using globals the way you do is to

  1. Close all sessions
  2. Call engine.dispose()

before forking. This will prevent connections from leaking to the child. For example:

def main():
    global session
    init()
    try:
        dummy = Dummy(value=1)
        session.add(dummy)
        session.commit()
        dummy_id = dummy.id
        # Return the Connection to the pool
        session.close()
        # Dispose of it!
        engine.dispose()
        # ...or call your cleanup() function, which does the same
        p = multiprocessing.Process(target=target, args=(dummy_id,))
        p.start()
        p.join()
        # Start a new session
        session = Session()
        dummy = session.query(Dummy).get(dummy_id)
        assert dummy.value == 2
    finally:
        cleanup()

Your second example does not trigger finalization in the child, and so it only seems to work, though it might be as broken as the first, as it is still inheriting a copy of the session and its connection defined locally in main.

More Related Contents:

  1. How to write DataFrame to postgres table
  2. CS50: LIKE operator, variable substitution with % expansion
  3. SQLAlchemy create_all() does not create tables
  4. Executing multiple statements with Postgresql via SQLAlchemy does not persist changes
  5. How to create an SQL View with SQLAlchemy?
  6. Generate sql with subquery as a column in select statement using SQLAlchemy
  7. Updates to JSON field don’t persist to DB
  8. Complex foreign key constraint in SQLAlchemy
  9. How to upsert pandas DataFrame to PostgreSQL table?
  10. SQLAlchemy multiple foreign keys in one mapped class to the same primary key
  11. sqlalchemy.exc.NoSuchModuleError: Can’t load plugin: sqlalchemy.dialects:postgres
  12. SQLAlchemy – performing a bulk upsert (if exists, update, else insert) in postgresql
  13. Strange SQLAlchemy error message: TypeError: ‘dict’ object does not support indexing
  14. Creating partial unique index with sqlalchemy on Postgres
  15. SQLAlchemy func.count on boolean column
  16. How to set connection timeout in SQLAlchemy
  17. SQLAlchemy: filtering on values stored in nested list of the JSONB field
  18. How to prevent UUID primary key for new SQLAlchemy objects being created with the same value
  19. psycopg2: insert multiple rows with one query
  20. Sqlite / SQLAlchemy: how to enforce Foreign Keys?
  21. multiprocessing.Pool: What’s the difference between map_async and imap?
  22. Log output of multiprocessing.Process
  23. Why does Python’s multiprocessing module import __main__ when starting a new process on Windows?
  24. sqlalchemy unique across multiple columns
  25. Using the multiprocessing module for cluster computing
  26. How to get rows which match a list of 3-tuples conditions with SQLAlchemy
  27. SQLAlchemy Joinedload filter column
  28. SQLAlchemy: get Model from table name. This may imply appending some function to a metaclass constructor as far as I can see
  29. Python multiprocessing memory usage
  30. Psycopg2, Postgresql, Python: Fastest way to bulk-insert

Leave a Comment