spakky-task

태스크 추상화 — 스케줄링, 디스패치

Stereotype

spakky.task.stereotype.task_handler

TaskHandler stereotype and task routing decorators.

This module provides @TaskHandler stereotype and @task decorator for organizing task-queue-driven architectures.

TaskRoute() dataclass

Bases: FunctionAnnotation

Annotation for marking methods as dispatchable tasks.

Associates a method as a task that can be dispatched to a task queue.

TaskHandler(*, name='', scope=Scope.SINGLETON) dataclass

Bases: Pod

Stereotype for task handler classes.

TaskHandlers contain methods decorated with @task that can be dispatched to task queues asynchronously.

task(obj)

Decorator for marking methods as dispatchable tasks.

All @task methods are dispatched to the task queue by the plugin aspect.

Example

@TaskHandler() class EmailTaskHandler: @task def send_email(self, to: str, subject: str, body: str) -> None: ...

Parameters:

Name	Type	Description	Default
obj	Callable[P, T]	The method to mark as a task.	required

Returns:

Type	Description
Callable[P, T]	The annotated method.

Source code in core/spakky-task/src/spakky/task/stereotype/task_handler.py

def task(obj: Callable[P, T]) -> Callable[P, T]:
    """Decorator for marking methods as dispatchable tasks.

    All @task methods are dispatched to the task queue by the plugin aspect.

    Example:
        @TaskHandler()
        class EmailTaskHandler:
            @task
            def send_email(self, to: str, subject: str, body: str) -> None:
                ...

    Args:
        obj: The method to mark as a task.

    Returns:
        The annotated method.
    """
    route = TaskRoute()
    return cast(Callable[P, T], route(obj))

options: show_root_heading: false

spakky.task.stereotype.schedule

Schedule stereotype for periodic task execution.

Provides @schedule decorator to mark TaskHandler methods for periodic execution (interval, daily, or crontab-based).

ScheduleRoute(interval=None, at=None, crontab=None) dataclass

Bases: FunctionAnnotation

Annotation for marking methods as periodically scheduled tasks.

Exactly one of interval, at, or crontab must be provided.

interval = None class-attribute instance-attribute

Fixed interval between executions.

at = None class-attribute instance-attribute

Daily execution at a specific time.

crontab = None class-attribute instance-attribute

Cron-like schedule specification.

schedule(*, interval=None, at=None, crontab=None)

Decorator for marking methods as periodically scheduled tasks.

Exactly one of interval, at, or crontab must be specified.

Example

@TaskHandler() class MaintenanceHandler: @schedule(interval=timedelta(minutes=30)) def health_check(self) -> None: ...

@schedule(at=time(3, 0))
def daily_cleanup(self) -> None:
    ...

@schedule(crontab=Crontab(hour=9, weekday=(Weekday.MONDAY, Weekday.WEDNESDAY, Weekday.FRIDAY)))
def triweekly_report(self) -> None:
    ...

Parameters:

Name	Type	Description	Default
interval	timedelta | None	Fixed interval between executions.	None
at	time | None	Daily execution at a specific time.	None
crontab	Crontab | None	Cron-like schedule specification.	None

Returns:

Type	Description
Callable[[Callable[P, T]], Callable[P, T]]	A decorator that annotates the method with ScheduleRoute.

Source code in core/spakky-task/src/spakky/task/stereotype/schedule.py

def schedule(
    *,
    interval: timedelta | None = None,
    at: time | None = None,
    crontab: Crontab | None = None,
) -> Callable[[Callable[P, T]], Callable[P, T]]:
    """Decorator for marking methods as periodically scheduled tasks.

    Exactly one of ``interval``, ``at``, or ``crontab`` must be specified.

    Example:
        @TaskHandler()
        class MaintenanceHandler:
            @schedule(interval=timedelta(minutes=30))
            def health_check(self) -> None:
                ...

            @schedule(at=time(3, 0))
            def daily_cleanup(self) -> None:
                ...

            @schedule(crontab=Crontab(hour=9, weekday=(Weekday.MONDAY, Weekday.WEDNESDAY, Weekday.FRIDAY)))
            def triweekly_report(self) -> None:
                ...

    Args:
        interval: Fixed interval between executions.
        at: Daily execution at a specific time.
        crontab: Cron-like schedule specification.

    Returns:
        A decorator that annotates the method with ScheduleRoute.
    """
    route = ScheduleRoute(interval=interval, at=at, crontab=crontab)
    return cast(Callable[[Callable[P, T]], Callable[P, T]], route)

options: show_root_heading: false

spakky.task.stereotype.crontab

Crontab value object for schedule specification.

Weekday

Bases: IntEnum

Day of the week (ISO 8601: Monday=0).

Month

Bases: IntEnum

Month of the year (1-12).

Crontab(month=None, day=None, weekday=None, hour=0, minute=0) dataclass

Cron-like schedule specification using Python native types.

Fields use None to mean "every" (wildcard). A single int means exactly that value; a tuple means multiple values.

Example

Every Monday at 03:00

Crontab(weekday=Weekday.MONDAY, hour=3)

Mon/Wed/Fri at 09:00

Crontab(weekday=(Weekday.MONDAY, Weekday.WEDNESDAY, Weekday.FRIDAY), hour=9)

1st and 15th of every month at midnight

Crontab(day=(1, 15))

month = None class-attribute instance-attribute

Month of the year. None means every month.

day = None class-attribute instance-attribute

Day of the month (1-31). None means every day.

weekday = None class-attribute instance-attribute

Day of the week. None means every day.

hour = 0 class-attribute instance-attribute

Hour of the day (0-23).

minute = 0 class-attribute instance-attribute

Minute of the hour (0-59).

options: show_root_heading: false

Interfaces

spakky.task.interfaces.task_result

Abstract task result handle for background task dispatchers.

AbstractTaskResult

Bases: ABC, Generic[T]

Abstract handle for the result of a dispatched background task.

Concrete adapters (e.g. CeleryTaskResult) implement this for each broker.

task_id abstractmethod property

Unique identifier for the dispatched task.

get() abstractmethod

Block until the task completes and return its result.

Returns:

Type	Description
T	The return value of the executed task method.

Source code in core/spakky-task/src/spakky/task/interfaces/task_result.py

@abstractmethod
def get(self) -> T:
    """Block until the task completes and return its result.

    Returns:
        The return value of the executed task method.
    """
    ...

options: show_root_heading: false

Post Processor

spakky.task.post_processor

Task handler registration post-processor.

TaskRegistrationPostProcessor()

Bases: IPostProcessor

Post-processor that scans @TaskHandler pods for @task methods.

This post-processor collects all task routes from TaskHandler pods and makes them available for task queue implementations to register.

Source code in core/spakky-task/src/spakky/task/post_processor.py

def __init__(self) -> None:
    self._task_routes = {}

post_process(pod)

Scan pod for @task methods and register their routes.

Parameters:

Name	Type	Description	Default
pod	object	The pod instance to process.	required

Returns:

Type	Description
object	The unmodified pod instance.

Source code in core/spakky-task/src/spakky/task/post_processor.py

def post_process(self, pod: object) -> object:
    """Scan pod for @task methods and register their routes.

    Args:
        pod: The pod instance to process.

    Returns:
        The unmodified pod instance.
    """
    pod_type = type(pod)

    if not TaskHandler.exists(pod_type):
        return pod

    for name, method in getmembers(pod, predicate=ismethod):
        route = TaskRoute.get_or_none(method)
        if route is None:
            continue

        self._task_routes[method] = route
        logger.debug(f"Registered task {pod_type.__name__}.{name}")

    return pod

get_task_routes()

Get all registered task routes.

Returns:

Type	Description
dict[Func, TaskRoute]	Dictionary mapping task methods to their routes.

Source code in core/spakky-task/src/spakky/task/post_processor.py

def get_task_routes(self) -> dict[Func, TaskRoute]:
    """Get all registered task routes.

    Returns:
        Dictionary mapping task methods to their routes.
    """
    return self._task_routes.copy()

options: show_root_heading: false

Errors

spakky.task.error

Spakky Task error hierarchy.

AbstractSpakkyTaskError

Bases: AbstractSpakkyFrameworkError, ABC

Base class for all spakky-task errors.

TaskNotFoundError

Bases: AbstractSpakkyTaskError

Raised when a task reference cannot be found in the registry.

DuplicateTaskError

Bases: AbstractSpakkyTaskError

Raised when attempting to register a task that already exists.

InvalidScheduleSpecificationError

Bases: AbstractSpakkyTaskError

Raised when a ScheduleRoute has invalid schedule options.

options: show_root_heading: false