Running sqlite insert operations in Async python

Question

I am trying to execute lots of blocking tasks asynchronously in python3.6. All the blocking tasks will store data in the SQLITE3 (peewee orm). While execution of these tasks occasionally gives me a DB Block error of sqlite3. I am using sanic, Peewee and python3.6.

Any workaound or improvement in the code to stop this DBBlock error.

#sqlite3_store_func is a blocking sqlite3 store function, which insert #data into the database

async def function_a(config, loop, executor):

    ins = SomesyncClass(path, config)

    ##this gives a list of data
    _, purchases = await ins.parse()

    #Blocking functions should not be called directly. 
    #For example, if a function blocks for 1 second, other tasks are
    # delayed by 1 second which can have an important impact on
    # reactivity.
    # An executor can be used to run a task in a different thread or 
    #even in a different process, to not block the thread of the 
    #event loop. See the AbstractEventLoop.run_in_executor() method. 
    await asyncio.wait(
            fs=[loop.run_in_executor(executor,  
                functools.partial(sqlite3_store_func, **purchase)) for purchase in purchases],
            return_when=asyncio.ALL_COMPLETED)
    return


async def parse(config, loop, executor):

    await asyncio.gather(*[
                function_a(config, loop, executor), 
                function_b(config, loop, executor),
                ])

    logger.info('Periodic task has finished execution')

@SANIC_BLUEPRINT.post('parse')
async def parse(request):
    """
    To get all the assets created by the requester
    """
    request.app.config.VALIDATE_FIELDS(["path"], request.json)

    #loop = asyncio.get_event_loop()
    executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)

    ##task which is non blocking
    request.app.add_task(parse_takeout(request.app.config, request.app.loop, executor))
    return response.json({"error": True, "success": False})

coleifer · Accepted Answer · 2019-07-15 14:55:25Z

3

A couple things to note.

Sqlite is embedded into the python process. It is not a separate server you communicate with over a socket, so already it's not like you are going to be able to leverage an event-loop to asynchronously query a sqlite db.

That leaves you with using a threadpool. As we all hopefully know, sqlite uses a global write lock, so only one connection can ever write to the database at any given time. This means that if you're using a threadpool, you either need to put a mutex around your writes, use a dedicated thread for all writes, or gracefully handle the failure to acquire the lock.

answered Jul 15, 2019 at 14:55

coleifer

26.4k7 gold badges67 silver badges81 bronze badges

Sign up to request clarification or add additional context in comments.

Collectives™ on Stack Overflow

Running sqlite insert operations in Async python

1 Answer 1

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Related