How to handle SQLAlchemy Connections in ProcessPool?
<p>I have a reactor that fetches messages from a RabbitMQ broker and triggers worker methods to process these messages in a process pool, something like this:</p>

<p><a href="http://i.stack.imgur.com/eKbAK.png"><img src="http://i.stack.imgur.com/eKbAK.png" alt="Reactor"></a></p>

<p>This is implemented using python <code>asyncio</code>, <code>loop.run_in_executor()</code> and <code>concurrent.futures.ProcessPoolExecutor</code>.</p>

<p>Now I want to access the database in the worker methods using SQLAlchemy. Mostly the processing will be very straightforward and quick CRUD operations.</p>

<p>The reactor will process 10-50 messages per second in the beginning, so it is not acceptable to open a new database connection for every request. Rather I would like to maintain one persistent connection per process.</p>

<p>My questions are: How can I do this? Can I just store them in a global variable? Will the SQA connection pool handle this for me? How to clean up when the reactor stops?</p>

<p><strong>[Update]</strong></p>

<ul>
<li>The database is MySQL with InnoDB.</li>
</ul>

<p><strong>Why choosing this pattern with a process pool?</strong></p>

<p>The current implementation uses a different pattern where each consumer runs in its own thread. Somehow this does not work very well. There are already about 200 consumers each running in their own thread, and the system is growing quickly. To scale better, the idea was to separate concerns and to consume messages in an I/O loop and delegate the processing to a pool. Of course, the performance of the whole system is mainly I/O bound. However, CPU is an issue when processing large result sets.</p>

<p>The other reason was "ease of use." While the connection handling and consumption of messages is implemented asynchronously, the code in the worker can be synchronous and simple.   </p>

<p>Soon it became evident that accessing remote systems through persistent network connections from within the worker are an issue. This is what the CommunicationChannels are for: Inside the worker, I can grant requests to the message bus through these channels.</p>

<p>One of my current ideas is to handle DB access in a similar way: Pass statements through a queue to the event loop where they are sent to the DB. However, I have no idea how to do this with SQLAlchemy. 
Where would be the entry point? 
Objects need to be <code>pickled</code> when they are passed through a queue. How do I get such an object from an SQA query?
The communication with the database has to work asynchronously in order not to block the event loop. Can I use e.g. aiomysql as a database driver for SQA?  </p>

<p>Your requirement of <strong>one database connection per process-pool process</strong> can be easily satisfied if some care is taken on how you instantiate the <code>session</code>, assuming you are working with the orm, in the worker processes.</p>

<p>A simple solution would be to have a global <a href="http://docs.sqlalchemy.org/en/latest/orm/contextual.html" rel="nofollow">session</a> which you reuse across requests:</p>

<pre><code># db.py
engine = create_engine("connection_uri", pool_size=1, max_overflow=0)
DBSession = scoped_session(sessionmaker(bind=engine)) 
</code></pre>

<p>And on the worker task:</p>

<pre><code># task.py
from db import engine, DBSession
def task():
    DBSession.begin() # each task will get its own transaction over the global connection
    ...
    DBSession.query(...)
    ...
    DBSession.close() # cleanup on task end
</code></pre>

<p>Arguments <code>pool_size</code> and <code>max_overflow</code> <a href="http://docs.sqlalchemy.org/en/latest/core/pooling.html#connection-pool-configuration" rel="nofollow">customize</a> the default <a href="http://docs.sqlalchemy.org/en/latest/core/pooling.html#sqlalchemy.pool.QueuePool" rel="nofollow">QueuePool</a> used by create_engine.<code>pool_size</code> will make sure your process only keeps 1 connection alive per process in the process pool.</p>

<p>If you want it to reconnect you can use <code>DBSession.remove()</code> which will remove the session from the registry and will make it reconnect at the next DBSession usage. You can also use the <code>recycle</code> argument of <a href="http://docs.sqlalchemy.org/en/latest/core/pooling.html#sqlalchemy.pool.Pool" rel="nofollow">Pool</a> to make the connection reconnect after the specified amount of time.</p>

<p>During development/debbuging you can use <a href="http://docs.sqlalchemy.org/en/latest/core/pooling.html#sqlalchemy.pool.AssertionPool" rel="nofollow">AssertionPool</a> which will raise an exception if more than one connection is checked-out from the pool, see <a href="http://docs.sqlalchemy.org/en/latest/core/pooling.html#switching-pool-implementations" rel="nofollow">switching pool implementations</a> on how to do that.</p>

python sqlalchemy rabbitmq python-multiprocessing python-asyncio