Concurrent Database Connections

DocTable makes it easy to establish concurrent database connections from different processes. DocTable objects can be copied as-is from one process to another, except that you must call .reopen_engine() to initialize in process thread. This removes now stale database connections (which are not meant to traverse processes) from the engine connection pool.

You may also want to use a large timeout using the timeout argument of the doctable constructor (provided in seconds).

import sqlalchemy
from multiprocessing import Process
import os
import random
import string
import dataclasses
import time
import sys
sys.path.append('..')
import doctable

import datetime
@doctable.schema
class SimpleRow:
    __slots__ = []
    id: int = doctable.IDCol()
    updated: datetime.datetime = doctable.AddedCol()
    process: str = doctable.Col()
    number: int = doctable.Col()

tmp = doctable.TempFolder('exdb')
db = doctable.DocTable(schema=SimpleRow, target=tmp.joinpath('tmp_concurrent.db'), new_db=True, timeout=60)

def thread_func(numbers, db):
    process_id = ''.join(random.choices(string.ascii_uppercase, k=2))
    print(f'starting process {process_id}')
    db.reopen_engine() # create all new connections
    for num in numbers:
        db.insert({'process': process_id, 'number': num})
        time.sleep(0.01)

numbers = list(range(100)) # these numbers are to be inserted into the database
        
db.delete()
with doctable.Distribute(5) as d:
    d.map_chunk(thread_func, numbers, db)
db.head(10)

starting process DF
starting process MB
starting process SA
starting process PTstarting process AP

	id	updated	process	number
0	1	None	DF	0
1	2	None	DF	1
2	3	None	MB	60
3	4	None	SA	80
4	5	None	DF	2
5	6	None	MB	61
6	7	None	SA	81
7	8	None	DF	3
8	9	None	MB	62
9	10	None	AP	40