symfonic.infra.protocols

protocols

TaskSchedulerProtocol -- abstract interface for background task scheduling.

The library depends only on this protocol. Concrete adapters (Celery, APScheduler, etc.) live in the application layer and are injected at runtime via dependency inversion.

NullScheduler is the default implementation that silently discards all task requests, ensuring the library works standalone without any external scheduler.

NullScheduler

No-op scheduler that silently discards all task requests.

Used as the default when no external scheduler is configured, ensuring the library functions standalone without Celery or any other task infrastructure.

cancel_task async

cancel_task(task_id: str) -> bool

Always return False (nothing to cancel).

Source code in src/symfonic/infra/protocols.py

async def cancel_task(self, task_id: str) -> bool:
    """Always return False (nothing to cancel)."""
    logger.debug("NullScheduler: discarding cancel_task(%s)", task_id)
    return False

enqueue_task async

enqueue_task(
    task_name: str, params: dict[str, Any] | None = None
) -> str

Accept and discard the task request.

Source code in src/symfonic/infra/protocols.py

async def enqueue_task(
    self,
    task_name: str,
    params: dict[str, Any] | None = None,
) -> str:
    """Accept and discard the task request."""
    self._counter += 1
    task_id = f"null-{self._counter}"
    logger.debug(
        "NullScheduler: discarding enqueue_task(%s) -> %s",
        task_name,
        task_id,
    )
    return task_id

schedule_recurring async

schedule_recurring(
    task_name: str,
    interval_seconds: float,
    params: dict[str, Any] | None = None,
) -> str

Accept and discard the recurring schedule request.

Source code in src/symfonic/infra/protocols.py

async def schedule_recurring(
    self,
    task_name: str,
    interval_seconds: float,
    params: dict[str, Any] | None = None,
) -> str:
    """Accept and discard the recurring schedule request."""
    self._counter += 1
    task_id = f"null-recurring-{self._counter}"
    logger.debug(
        "NullScheduler: discarding schedule_recurring(%s, %ss) -> %s",
        task_name,
        interval_seconds,
        task_id,
    )
    return task_id

TaskSchedulerProtocol

Bases: Protocol

Abstract interface for background task scheduling.

All methods accept a task_name string that the adapter maps to a concrete task implementation (e.g. a Celery task, an APScheduler job, etc.).

cancel_task async

cancel_task(task_id: str) -> bool

Cancel a previously enqueued or recurring task.

Parameters:

Name	Type	Description	Default
task_id	str	The identifier returned by enqueue_task or schedule_recurring.	required

Returns:

Type	Description
bool	True if the task was found and cancelled, False otherwise.

Source code in src/symfonic/infra/protocols.py

async def cancel_task(self, task_id: str) -> bool:
    """Cancel a previously enqueued or recurring task.

    Args:
        task_id: The identifier returned by ``enqueue_task``
            or ``schedule_recurring``.

    Returns:
        True if the task was found and cancelled, False otherwise.
    """
    ...

enqueue_task async

enqueue_task(
    task_name: str, params: dict[str, Any] | None = None
) -> str

Enqueue a task for immediate background execution.

Parameters:

Name	Type	Description	Default
task_name	str	Identifier for the task to execute.	required
params	dict[str, Any] | None	Optional keyword arguments for the task.	None

Returns:

Type	Description
str	A task ID string that can be used with cancel_task.

Source code in src/symfonic/infra/protocols.py

async def enqueue_task(
    self,
    task_name: str,
    params: dict[str, Any] | None = None,
) -> str:
    """Enqueue a task for immediate background execution.

    Args:
        task_name: Identifier for the task to execute.
        params: Optional keyword arguments for the task.

    Returns:
        A task ID string that can be used with ``cancel_task``.
    """
    ...

schedule_recurring async

schedule_recurring(
    task_name: str,
    interval_seconds: float,
    params: dict[str, Any] | None = None,
) -> str

Schedule a task to run repeatedly at a fixed interval.

Parameters:

Name	Type	Description	Default
task_name	str	Identifier for the task.	required
interval_seconds	float	Seconds between successive executions.	required
params	dict[str, Any] | None	Optional keyword arguments for each invocation.	None

Returns:

Type	Description
str	A task ID string for cancellation.

Source code in src/symfonic/infra/protocols.py

async def schedule_recurring(
    self,
    task_name: str,
    interval_seconds: float,
    params: dict[str, Any] | None = None,
) -> str:
    """Schedule a task to run repeatedly at a fixed interval.

    Args:
        task_name: Identifier for the task.
        interval_seconds: Seconds between successive executions.
        params: Optional keyword arguments for each invocation.

    Returns:
        A task ID string for cancellation.
    """
    ...