Interfaces

Base interface for classes that acts like factories.

class core_mixins.interfaces.factory.IFactory

Bases: ABC

Base interface for classes that acts like factories. The IFactory pattern allows you to:

• Automatically register concrete implementations.

• Discover available implementations at runtime.

• Create instances by string reference.

• Maintain separate registries for different factory types.

IFactory is a pure protocol, it provides the registration mechanism but does not store any references itself. Registries live exclusively on concrete sub-hierarchies (abstract subclasses of IFactory). Every abstract subclass automatically gets its own isolated _impls dict when it is defined.

Usage pattern:

class IAnimalFactory(IFactory, ABC):
    pass  # gets its own registry

class Dog(IAnimalFactory):
    @classmethod
    def registration_key(cls) -> str:
        return "dog"

IAnimalFactory.get_registered_classes()  # {"dog"}
IFactory.get_registered_classes()        # always empty

abstractmethod classmethod registration_key() → str

It returns the name (reference) for the key used to register, like: return self.__name__

classmethod get_registered_classes() → Set[str]

It returns the current registered implementations

classmethod get_class(cls_reference: str) → Type[Self] | None

It returns the reference (class type) by reference

This module provides a base implementation for ETL tasks under the project ecosystem.

class core_mixins.interfaces.task.TaskStatus(*values)

Bases: str, Enum

Possible status a task can have during its life

CREATED = 'CREATED'

EXECUTING = 'EXECUTING'

SUCCESS = 'SUCCESS'

ERROR = 'ERROR'

class core_mixins.interfaces.task.ITask(name: str | None = None, description: str | None = None, logger: Logger | None = None)

Bases: IFactory, ABC

Base implementations for different tasks/processes

__init__(name: str | None = None, description: str | None = None, logger: Logger | None = None) → None

classmethod registration_key() → str

It returns the key used to register the class

property name

It returns the task identification

property status: TaskStatus

It returns the current status of the task

abstractmethod execute(*args, **kwargs) → Any

You must implement the task’s process

info(message, *args, **kwargs) → None

Log entry with severity ‘INFO’

warning(message, *args, **kwargs) → None

Log entry with severity ‘WARNING’

error(error, *args, **kwargs) → None

Log entry with severity ‘ERROR’

debug(message, *args, **kwargs) → None

Log entry with severity ‘DEBUG’

class core_mixins.interfaces.task.TaskResult(status: TaskStatus, result: Any | None = None, error: Exception | None = None, execution_time: float = 0.0)

Bases: object

Represents the result of a task execution that exposes status, result, or the error.

__init__(status: TaskStatus, result: Any | None = None, error: Exception | None = None, execution_time: float = 0.0) → None

property is_success: bool

Check if the task execution was successful

property is_error: bool

Check if the task execution failed

exception core_mixins.interfaces.task.TaskException

Bases: Exception

Custom exception for Tasks