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>An approach that has served me really well is to use a webserver to handle and scale the process pool. flask-sqlalchemy even in its default state will keep a connection pool and not close each connection on each request response cycle. </p>

<p>The asyncio executor can just call url end points to execute your functions. The added benefit is that because all processes doing the work are behind a url you can trivially scale your worker pool across mutliple machines, adding more processes via gunicorn or one of the other many methods to scale a simple wsgi server. Plus you get all the fault tolerant goodness. </p>

<p>The downside is that you might be passing more information across the network. However as you say the problem is CPU bound and you will probably be passing much more data to and from the database. </p>

python sqlalchemy rabbitmq python-multiprocessing python-asyncio