[Python] Why is await queue.get() Blocking and Causing My Async Consumers to Hang?

I’m having trouble with an asynchronous queue. Specifically, the await queue.get() seems to block the rest of my application, causing it to hang.

I came across a similar issue discussed here: Why is asyncio queue await get() blocking? blocking, but I’m still having difficulty understanding how to resolve the problem in my case.

In my setup, I am trying to create a web scraping API (with FastAPI) that allows users submit URLs which are batched and processed by different web scrapers running in consumer tasks. However, it seems that once the consumers are waiting for new batches from the queue, the app halts. Specifically, I’m calling start_consumers() during the FastAPI lifespan event before the application starts, but the app never fully initialises because it gets blocked.

Is there a way to modify my setup so that the consumers can wait for new items in the queue without blocking the rest of the application? Or is this approach doomed?

import asyncio
from asyncio.queues import Queue

from internal.services.webscraping.base import BaseScraper


class QueueController:

_logger = create_logger(__name__)

def __init__(
self,
scrapers: list[BaseScraper],
batch_size: int = 50
):
self.queue = Queue()
self.batch_size = batch_size
self.scrapers = scrapers
self.consumers = []
self.running = False

async def put(self, urls: list[str]) -> None:
"""
Add batches of URLs to the queue.
"""
# Split the list of URLs into batches of size self.batch_size and add them to the queue.
# If the queue is full, wait until there is space available.
for i in range(0, len(urls), self.batch_size):
batch = urls[i:i + self.batch_size]
await self.queue.put(batch)

async def get(self) -> list[str]:
"""
Retrieve a batch from the queue.
"""
# Get a batch of URLs from the queue.
# If queue is empty, wait until an item is available.
return await self.queue.get()

async def consumer(self, scraper: BaseScraper) -> None:
"""
Consumer coroutine that processes batches of URLs.
"""
# Consumer tasks are designed to run in an infinite loop (as long as self.running is
# True) and fetch batches of URLs from the queue.
while self.running:
try:
batch = await self.get()
if batch:
records = await scraper.run(batch)
# TODO: Handle saving of result
except Exception as e:
# TODO: Add proper error handling
...
raise e

async def start_consumers(self) -> None:
"""
Start the consumer tasks.

Notes:
https://docs.python.org/3/library/asyncio-task.html
https://superfastpython.com/asyncio-task/
"""
self.running = True
self.consumers = [
asyncio.create_task(self.consumer(scraper)) for scraper in self.scrapers
]
await asyncio.gather(*self.consumers)

async def stop_consumers(self) -> None:
"""
Stop all consumer tasks gracefully.
"""
self.running = False
for task in self.consumers:
task.cancel()

Continue reading...