apscheduler blocks the main loop

[billsioros]

Hello 👋🏽

I'm using apscheduler for background tasks with AsyncIOScheduler and it seems to block the main application loop. Has anyone run into this or have any suggestions?

Thanks in advance 🙏🏽

[cve-zh00]

Can u send a code fragment?

[billsioros]

Yes, ofc! I'm sending bits and pieces as the code base is quite large. I know somethings might be redundant:

The Job Service class that wraps the scheduler:

import logging
from datetime import datetime, timezone
from pathlib import Path
from typing import Coroutine, List, cast

from apscheduler.schedulers.base import BaseScheduler

from api.core.search import SearchCriteria, SearchOperator
from api.core.service_result import ServiceResult
from api.models.ingestion import PipelineStatus
from api.schemas.ingestion import IngestionRequestSchema, IngestionRequestSearchCriteria
from api.schemas.job import JobSchema
from api.services.ingestion_request_service import IngestionRequestService
from api.services.ingestion_service import IngestionService
from api.settings import Settings

logger = logging.getLogger(__name__)


class JobService:
    def __init__(self, scheduler: BaseScheduler) -> None:
        self._scheduler = scheduler

    def submit(self, func, *args, job_id=None, name=None, **kwargs):
        """Submit a job to be executed immediately.

        If the function is asynchronous, it's wrapped to make it synchronous
        so that APScheduler can execute it.

        Args:
            func: The function to be executed
            *args: Arguments to pass to the function
            job_id: Optional ID for the job
            name: Optional name for the job
            **kwargs: Additional keyword arguments for the job

        Returns:
            The job ID
        """
        try:
            # Use add_job with 'date' trigger for immediate execution
            job = self._scheduler.add_job(
                func=func,
                args=args,
                id=job_id,
                name=name,
                trigger="date",  # Execute once at the specified time
                next_run_time=datetime.now(timezone.utc),  # Execute immediately
                replace_existing=True,  # Replace if job with same ID exists
                misfire_grace_time=None,  # Allow job to run even if scheduler is backed up
            )

            if not self._scheduler.running:
                self._scheduler.start()

            logger.info(f"Job submitted with ID: {job.id}")
            return job.id
        except Exception as e:
            logger.error(f"Error submitting job: {e}", exc_info=True)
            raise

Example Usage:

scheduler = AsyncIOScheduler(
    jobstores={"default": MemoryJobStore()},
    job_defaults={"coalesce": False, "max_instances": 1},
    timezone=timezone.utc,
)

# The job itself seems completely fine
async def process_ingestion(
    id: int,
    settings: Settings,
    ingestion_request_service: IngestionRequestService,
    ingestion_service: IngestionService,
):
   ...

job_service = JobService(scheduler)
job_service.submit(
    process_ingestion,
    ingestion_result.payload.id,
    settings,
    ingestion_request_service,
    ingestion_service,
    job_id=str(ingestion_result.payload.id),
    job_name=f"ingest-{ingestion_result.payload.id}",
)