Module: instance_method_task

Classes

class InstanceMethodTask (*, 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, key_type_str: str = None, key_str: str = None, method_name: str = None)

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

Expand source code

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

    key_type_str: str = missing()
    """Key type as dot-delimited string in module.ClassNameKey format inclusive of Key suffix if present."""

    key_str: str = missing()
    """Key as semicolon-delimited string."""

    method_name: str = missing()
    """The name of instance method in snake_case or PascalCase format, do not use for @classmethod or @staticmethod."""

    def _execute(self) -> None:
        """Invoke the specified instance method."""

        # Get current context
        context = Context.current()

        # Save self to ensure the worker process loads the same record
        context.save_one(self)

        key_type = ClassInfo.get_class_type(self.key_type_str)
        key = key_serializer.deserialize_key(self.key_str, key_type)

        # Load record from storage
        record = context.load_one(key_type, key)  # TODO: Require record type?

        # Convert the name to snake_case and get method callable
        method_name = self.normalize_method_name(self.method_name)
        method = getattr(record, method_name)

        # Invoke the callable
        method()

    @classmethod
    def create(
        cls,
        *,
        queue: TaskQueueKey,
        record_or_key: KeyProtocol | None = None,
        method_callable: Callable,
    ) -> Self:
        """
        Create from the record or its key and an instance-bound or class-bound method callable.

        Notes:
            - The key is required if the callable is for a class rather than an instance.

        Args:
            queue: Queue that will run the task
            record_or_key: Record or its key
            method_callable: Callable bound to a class (ClassName.method_name) or its instance (obj.method_name)
        """

        # Populate known fields
        result = cls(queue=queue)

        # Get key type and key
        key_type = record_or_key.get_key_type()
        result.key_type_str = f"{key_type.__module__}.{key_type.__name__}"
        result.key_str = key_serializer.serialize_key(record_or_key)

        # Two tokens because the callable is bound to a class or its instance
        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"{key_type.__name__};{result.key_str};{method_name_pascal_case}"
        return result

Ancestors

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

Static methods

def create(*, queue: TaskQueueKey, record_or_key: KeyProtocol | None = None, method_callable: Callable) -> Self

Create from the record or its key and an instance-bound or class-bound method callable.

Notes

• The key is required if the callable is for a class rather than an instance.

Args

queue

Queue that will run the task

record_or_key

Record or its key

method_callable

Callable bound to a class (ClassName.method_name) or its instance (obj.method_name)

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 key_str -> str

Key as semicolon-delimited string.

var key_type_str -> str

Key type as dot-delimited string in module.ClassNameKey format inclusive of Key suffix if present.

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 instance method in snake_case or PascalCase format, do not use for @classmethod or @staticmethod.

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.

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.