Files
mostovik-backend/src/organizations/tasks.py

68 lines
1.9 KiB
Python

"""Celery tasks for organization source extensions."""
from __future__ import annotations
import logging
from dataclasses import asdict
from celery import shared_task
from organizations.cache import invalidate_organization_api_cache
from organizations.source_backfill import OrganizationSourceBackfillService
logger = logging.getLogger(__name__)
@shared_task
def backfill_all_organization_sources(batch_size: int = 100) -> dict:
"""Backfill all organization source extensions from legacy parser tables."""
result = OrganizationSourceBackfillService.backfill()
invalidate_organization_api_cache()
payload = {"batch_size": batch_size, **asdict(result)}
logger.info("All organization source extensions backfilled: %s", payload)
return payload
@shared_task
def backfill_organization_sources_for_parser_batch(
*,
source: str,
batch_id: int,
batch_size: int = 100,
) -> dict:
"""Backfill source extensions affected by one parser batch."""
result = OrganizationSourceBackfillService.backfill(
source=source,
batch_id=batch_id,
)
invalidate_organization_api_cache()
payload = {
"source": source,
"batch_id": batch_id,
"batch_size": batch_size,
**asdict(result),
}
logger.info("Organization source extensions backfilled: %s", payload)
return payload
@shared_task
def refresh_all_organization_data_snapshots(batch_size: int = 100) -> dict:
"""Deprecated compatibility wrapper for the old snapshot schedule."""
return backfill_all_organization_sources(batch_size=batch_size)
@shared_task
def refresh_organization_data_snapshots_for_parser_batch(
*,
source: str,
batch_id: int,
batch_size: int = 100,
) -> dict:
"""Deprecated compatibility wrapper for old parser task imports."""
return backfill_organization_sources_for_parser_batch(
source=source,
batch_id=batch_id,
batch_size=batch_size,
)