When using mongodb MotorClient with scrapy, async with await will introduce Runtime error

[keatonLiu]

When using mongodb MotorClient with scrapy, async with await will introduce Runtime error

[Gallaecio]

parse was just a suggestion for a quick test. The right approach would probably be to use the engine_started signal.

[wRAR]

Please provide a complete minimal reproducible example.

[keatonLiu]

This is a reproduce example:

import scrapy
from motor.core import AgnosticCollection, AgnosticDatabase
from motor.motor_asyncio import (
    AsyncIOMotorClient as MotorClient,
)
from pymongo.errors import OperationFailure
from scrapy.settings import Settings
from scrapy.utils.project import get_project_settings


class QccSpiderBase(scrapy.Spider, metaclass=ABCMeta):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.settings: Settings = get_project_settings()

        self.client: MotorClient = MotorClient(self.settings['MONGO_URI'])
        self.db_name = self.settings['DB_NAME']
        self.db: Optional[AgnosticDatabase] = self.client[self.db_name]
        self.collection: Optional[AgnosticCollection] = self.db[self.collection_name]

    async def parse(self, response, **kwargs):
        async with await self.client.start_session() as s:
            count = await self.collection.count_documents({"compName": compName}, limit=1, session=s)

[Gallaecio]

Is async with await correct? Shouldn’t it be async with?

[Gallaecio]

Regarding the actual issue, does it work if you move the client creation to the parse method?

[keatonLiu]

It does not raise an error before I update my libraries including scrapy and motor to the latest, I don't know if it is a new feature in twisted.
It looks like the error happens because motor client created a new asyncio loop other than using the loop used by the crawler's reactor. If I don't specifiy the io_loop paramter when initialize, it will create a new loop when first getting the loop.
I've fixed it by moving the motor client initialization in parse(), and set the event loop to the current event loop used by scrapy's twisted reactor.

    def setup_mongo_client(self):
        self.client = MotorClient(self.settings['MONGO_URI'], io_loop=get_event_loop())
        self.db = self.client[self.db_name]
        self.collection = self.db[self.collection_name]
       
        print(f"self.client.get_io_loop(): {self.client.io_loop}")
        print(f"get_event_loop(): {get_event_loop()}")
        reactor = sys.modules["twisted.internet.reactor"]
        print(f"twisted.internet.reactor: {reactor._asyncioEventloop}")

    def parse(self, response, **kwargs):
        self.setup_mongo_client()

[keatonLiu]

Regarding the actual issue, does it work if you move the client creation to the parse method?

Yes, it is the solution, but I used to initialize it in init and it always works fine

[wRAR]

motor client created a new asyncio loop other than using the loop used by the crawler's reactor. If I don't specifiy the io_loop paramter when initialize, it will create a new loop when first getting the loop.

Makes sense.

[keatonLiu]

Let me correct it, motor client uses motor.frameworks.asyncio.get_event_loop() by default to get the current running loop or create a new loop when it does not exist. If I put the initialzation in init, the reactor haven't created a loop so it will create a new one.

It does not raise an error before I update my libraries including scrapy and motor to the latest, I don't know if it is a new feature in twisted. It looks like the error happens because motor client created a new asyncio loop other than using the loop used by the crawler's reactor. If I don't specifiy the io_loop paramter when initialize, it will create a new loop when first getting the loop. I've fixed it by moving the motor client initialization in parse(), and set the event loop to the current event loop used by scrapy's twisted reactor.

    def setup_mongo_client(self):
        self.client = MotorClient(self.settings['MONGO_URI'], io_loop=get_event_loop())
        self.db = self.client[self.db_name]
        self.collection = self.db[self.collection_name]
       
        print(f"self.client.get_io_loop(): {self.client.io_loop}")
        print(f"get_event_loop(): {get_event_loop()}")
        reactor = sys.modules["twisted.internet.reactor"]
        print(f"twisted.internet.reactor: {reactor._asyncioEventloop}")

    def parse(self, response, **kwargs):
        self.setup_mongo_client()

[keatonLiu]

But why is scrapy not using the current event loop when there already have one to avoid this error?

[wRAR]

Scrapy currently uses asyncio.get_event_loop() to get the loop.