在异步python中运行sqlite插入操作答案

【问题标题】：Running sqlite insert operations in Async python在异步python中运行sqlite插入操作
【发布时间】：2019-11-22 22:42:22
【问题描述】：

我正在尝试在 python3.6 中异步执行许多阻塞任务。所有阻塞任务都将数据存储在 SQLITE3 (peewee orm) 中。虽然执行这些任务偶尔会给我一个 sqlite3 的 DB Block 错误。我正在使用 sanic、Peewee 和 python3.6。

任何解决方法或改进代码以阻止此 DBBlock 错误。

#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})

【问题讨论】：

标签： python-3.x sqlite asynchronous peewee

【解决方案1】：

有几点需要注意。

Sqlite 嵌入到 python 进程中。不是您通过套接字与之通信的单独服务器，因此您已经不能利用事件循环来异步查询 sqlite db。

剩下的就是使用线程池了。我们都希望知道，sqlite 使用全局写锁，因此在任何给定时间只有一个连接可以写入数据库。这意味着，如果您使用线程池，则需要在写入周围放置一个互斥锁，为所有写入使用专用线程，或者优雅地处理获取锁的失败。

【讨论】：