Hi I'm using AsyncIOMotorClient for asynchronous db calls to mongoDb. Below is my code.
xyz.py
async def insertMany(self,collection_name,documents_to_insert):
try:
collection=self.database[collection_name]
document_inserted = await collection.insert_many(documents_to_insert)
return document_inserted
except Exception:
raise
def insertManyFn(self,collection_name,documents_to_insert):
try:
loop=asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop1=asyncio.get_event_loop()
inserted_documents_count = loop1.run_until_complete(self.insertMany(collection_name, documents_to_insert))
if inserted_documents_count==len(documents_to_insert):
document_to_insert={Config.DB_JOB_COLUMN:Job.job_id,Config.DB_JOB_RESULT_COLUMN:Config.DB_JOB_RESULT_SUCCESS}
loop1.run_until_complete(self.insertOne(Config.DB_JOB_COLLECTION, document_to_insert))
except Exception:
raise
xyz1.py
t=Timer(10,xyz.insertManyFn,\
(collection_name,documents_to_insert))
t.start()
While running this I'm getting an exception
RuntimeError: Task <Task pending coro=<xyz.insertMany() running at <my workspace location>/xyz.py:144> cb=[_run_until_complete_cb() at /usr/lib64/python3.5/asyncio/base_events.py:164]> got Future <Future pending cb=[_chain_future.<locals>._call_check_cancel() at /usr/lib64/python3.5/asyncio/futures.py:431]> attached to a different loop
In the above program insertManyFn will be called after 10sec and do the insert operation. But when it make the first call to insertMany I'm getting an exception.
I still want my MotorClient to be at the top level of the module, so this is what I do: I patch MotorClient.get_io_loop to always return the current loop.
import asyncio
import motor.core
from motor.motor_asyncio import (
AsyncIOMotorClient as MotorClient,
)
# MongoDB client
client = MotorClient('mongodb://localhost:27017/test')
client.get_io_loop = asyncio.get_running_loop
# The current database ("test")
db = client.get_default_database()
# async context
async def main():
posts = db.posts
await posts.insert_one({'title': 'great success!')
# Run main()
asyncio.run(main())
According to the documentation, AsyncIOMotorClient should be passed an ioloop if you don't use the default one. Try creating the client after you create your event loop:
loop=asyncio.new_event_loop()
asyncio.set_event_loop(loop)
client = AsyncIOMotorClient(io_loop=loop)
I've modified the code and it's working.
def insertManyFn(self,loop,collection_name,documents_to_insert):
try:
inserted_documents_count = loop.run_until_complete(self.insertMany(event_loop,collection_name, documents_to_insert))
if len(inserted_documents_count)==len(documents_to_insert):
document_to_insert={Config.DB_JOB_COLUMN:Job.job_id,Config.DB_JOB_RESULT_COLUMN:Config.DB_JOB_RESULT_SUCCESS}
loop1.run_until_complete(self.insertOne(Config.DB_JOB_COLLECTION, document_to_insert))
except Exception:
raise
loop=asyncio.get_event_loop()
t=Timer(10,self.xyz.insertManyFn,(loop,collection_name,documents_to_insert))
t.start()
Explanation- I'm using python threading timer, which creates own thread to execute a function after a certain time. So, inside this thread I was getting event loop which should not be the correct approach, it should be first getting the event loop and create a timer thread in it. I guess this is the only reason.