Module: static_method_task

Classes

class StaticMethodTask (*, task_id: str = None, label: str | None = None, queue: TaskQueueKey = None, status: TaskStatusEnum = None, progress_pct: float = None, elapsed_sec: float | None = None, remaining_sec: float | None = None, error_message: str | None = None, type_str: str = None, method_name: str = None)

Invoke a @staticmethod or @classmethod, do not use for instance methods.

Expand source code

@dataclass(slots=True, kw_only=True)
class StaticMethodTask(CallableTask):
    """Invoke a @staticmethod or @classmethod, do not use for instance methods."""

    type_str: str = missing()
    """Class type as dot-delimited string in module.ClassName format."""

    method_name: str = missing()
    """The name of @staticmethod in snake_case or PascalCase format."""

    def _execute(self) -> None:
        """Invoke the specified @staticmethod or @classmethod."""

        # Get record type from fully qualified name in module.ClassName format
        record_type = ClassInfo.get_class_type(self.type_str)

        # Method callable is already bound to cls, it is not necessary to pass cls as an explicit parameter
        method_name = self.normalize_method_name(self.method_name)
        method = getattr(record_type, method_name)

        # Invoke the callable
        method()

    @classmethod
    def create(
            cls,
            *,
            queue: TaskQueueKey,
            record_type: Type,
            method_callable: Callable,
    ) -> Self:
        """Create from @staticmethod callable and record type."""

        # Populate known fields
        result = cls(queue=queue)
        result.type_str = f"{record_type.__module__}.{record_type.__name__}"

        # Check that __self__ is either absent (@staticmethod) or is a class (@classmethod)
        if (method_cls := getattr(method_callable, "__self__", None)) is not None and not inspect.isclass(method_cls):
            raise RuntimeError(
                f"Callable '{method_callable.__qualname__}' for task_id='{result.task_id}' is "
                f"an instance method rather than @staticmethod or @classmethod, "
                f"use 'InstanceMethodTask' instead of 'StaticMethodTask'."
            )

        # Two tokens because the callable is bound to a class
        method_tokens = method_callable.__qualname__.split(".")
        if len(method_tokens) == 2:
            # Second token is method name
            result.method_name = method_tokens[1]
        else:
            raise RuntimeError(
                f"Callable '{method_callable.__qualname__}' for task_id='{result.task_id}' does not "
                f"have two dot-delimited tokens indicating it is not a method bound to a class."
            )

        # Set label and return
        method_name_pascal_case = CaseUtil.snake_to_pascal_case(result.method_name)
        result.label = f"{record_type.__name__};{method_name_pascal_case}"
        return result

Ancestors

• CallableTask
• Task
• TaskKey
• KeyMixin
• RecordMixin
• abc.ABC
• typing.Generic

Static methods

def create(*, queue: TaskQueueKey, record_type: Type, method_callable: Callable) -> Self

Create from @staticmethod callable and record type.

def get_key_type() -> Type

Inherited from: CallableTask.get_key_type

Return key type even when called from a record.

def normalize_method_name(method_name: str) -> str

Inherited from: CallableTask.normalize_method_name

If method name has uppercase letters, assume it is PascalCase and convert to snake_case.

def wait_for_completion(task_key: TaskKey, timeout_sec: int = 10) -> None

Inherited from: CallableTask.wait_for_completion

Wait for completion of the specified task run before exiting from this method (not async/await).

Fields

var elapsed_sec -> float | None

Inherited from: CallableTask.elapsed_sec

Elapsed time in seconds if available.

var error_message -> str | None

Inherited from: CallableTask.error_message

Error message for Failed status if available.

var label -> str | None

Inherited from: CallableTask.label

Label for information purposes only (should not be used in processing).

var method_name -> str

The name of @staticmethod in snake_case or PascalCase format.

var progress_pct -> float

Inherited from: CallableTask.progress_pct

Task progress in percent from 0 to 100.

var queue -> TaskQueueKey

Inherited from: CallableTask.queue

The queue that will run the task once it is saved.

var remaining_sec -> float | None

Inherited from: CallableTask.remaining_sec

Remaining time in seconds if available.

var status -> TaskStatusEnum

Inherited from: CallableTask.status

Begins from Pending, continues to Running or Paused, and ends with Completed, Failed, or Cancelled.

var task_id -> str

Inherited from: CallableTask.task_id

Unique task identifier.

var type_str -> str

Class type as dot-delimited string in module.ClassName format.

Methods

def get_key(self) -> TaskKey

Inherited from: CallableTask.get_key

Return a new key object whose fields populated from self, do not return self.

def init_all(self) -> None

Inherited from: CallableTask.init_all

Invoke ‘init’ for each class in the order from base to derived, then validate against schema.

def run_task(self) -> None

Inherited from: CallableTask.run_task

Invoke execute with task status updates and exception handling.