Django - Testing async management command not working as expected

To give some context, i made a async function which works fine when testing manually but when I use Django's testcases, the query is not returning anything back inside the async block, if I try the same query outside, it returns the objects

class Command(BaseCommand):
    help = "Update companies' website field using LLM + Annual reports content"
    def add_arguments(self, parser):
        parser.add_argument("-c", type=int, default=10)

    def handle(self, *args, **kwargs):
        rows = list(
            CompanyInfo.objects.filter(website="")
            .select_related("company_code")
            .exclude(company_code__bse_code="", company_code__nse_code="")
        )
        print(BseAnnouncement.objects.all()) #This is working, returns 1 object
        asyncio.run(self._handle_async(rows, kwargs["c"]))

    async def _handle_async(self, rows, concurrency):
        if not len(rows):
            logger.info("No companies with missing websites found.")
            return
        await _update_websites(rows, concurrency)
        logger.info("Website update process completed.")


async def _filter_recent(company):
    if company.bse_code:
        model = BseAnnouncement
    elif company.nse_code and company.is_sme:
        model = NseEmergeAnnouncement
    else:
        model = NseAnnouncement
    print(f"Fetching announcements for {model}")
    print(f"Company is {company}")
    print(f"Company ID is {company.id}")

    queryset = model.objects.filter(company_code_id=company.id).order_by("-create_date") #Empty return -> No object
    print(BseAnnouncement.objects.all()) #Empty return -> No object
    results = await sync_to_async(list)(queryset)
    return results

Вернуться на верх