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>@roman: Nice challenge you have there.</p>

<p>I have being in a similar scenario before so here is my <em>2 cents</em>: unless this consumer only <em>"read"</em> and <em>"write"</em> the message, without do any real proccessing of it, you could <em>re-design</em> this consumer as a consumer/producer that will <em>consume</em> the message, it will process the message and then will put result in another queue, that queue (processed messages for say) could be read by 1..N non-pooled asynchronous processes that would have open the DB connection in it's own entire life-cycle.</p>

<p>I can extend my answer, but I don't know if this approach fits for your needs, if so, I can give you more detail about the extended design.</p>

python sqlalchemy rabbitmq python-multiprocessing python-asyncio