Skip to content

Interfaces

biopro_sdk.interfaces

IEventBus

Bases: Protocol

Abstract interface for the global event bus.

Source code in src/biopro_sdk/interfaces/i_event_bus.py 
 5
 6
 7
 8
 9
10
11
12
13
14
15
@runtime_checkable
class IEventBus(Protocol):
    """Abstract interface for the global event bus."""

    def subscribe(self, event_type: str, callback: Callable[..., Any]) -> None:
        """Subscribe a callback to a specific event type."""
        ...

    def publish(self, event_type: str, *args: Any, **kwargs: Any) -> None:
        """Publish an event to all subscribers."""
        ...

publish(event_type, *args, **kwargs)

Publish an event to all subscribers.

Source code in src/biopro_sdk/interfaces/i_event_bus.py 
13
14
15
def publish(self, event_type: str, *args: Any, **kwargs: Any) -> None:
    """Publish an event to all subscribers."""
    ...

subscribe(event_type, callback)

Subscribe a callback to a specific event type.

Source code in src/biopro_sdk/interfaces/i_event_bus.py 
 9
10
11
def subscribe(self, event_type: str, callback: Callable[..., Any]) -> None:
    """Subscribe a callback to a specific event type."""
    ...

ILogger

Bases: Protocol

Abstract interface for logging.

Source code in src/biopro_sdk/interfaces/i_logger.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
@runtime_checkable
class ILogger(Protocol):
    """Abstract interface for logging."""

    def debug(self, msg: str) -> None:
        """Log a debug message."""
        ...

    def info(self, msg: str) -> None:
        """Log an informational message."""
        ...

    def warning(self, msg: str) -> None:
        """Log a warning message."""
        ...

    def error(self, msg: str, exception: Exception | None = None) -> None:
        """Log an error message, optionally with an exception trace."""
        ...

    def exception(self, msg: str) -> None:
        """Log an exception message."""
        ...

debug(msg)

Log a debug message.

Source code in src/biopro_sdk/interfaces/i_logger.py 
 8
 9
10
def debug(self, msg: str) -> None:
    """Log a debug message."""
    ...

error(msg, exception=None)

Log an error message, optionally with an exception trace.

Source code in src/biopro_sdk/interfaces/i_logger.py 
20
21
22
def error(self, msg: str, exception: Exception | None = None) -> None:
    """Log an error message, optionally with an exception trace."""
    ...

exception(msg)

Log an exception message.

Source code in src/biopro_sdk/interfaces/i_logger.py 
24
25
26
def exception(self, msg: str) -> None:
    """Log an exception message."""
    ...

info(msg)

Log an informational message.

Source code in src/biopro_sdk/interfaces/i_logger.py 
12
13
14
def info(self, msg: str) -> None:
    """Log an informational message."""
    ...

warning(msg)

Log a warning message.

Source code in src/biopro_sdk/interfaces/i_logger.py 
16
17
18
def warning(self, msg: str) -> None:
    """Log a warning message."""
    ...

ITaskScheduler

Bases: Protocol

Abstract interface for background task scheduling.

Source code in src/biopro_sdk/interfaces/i_task_scheduler.py 
 7
 8
 9
10
11
12
13
14
15
16
17
@runtime_checkable
class ITaskScheduler(Protocol):
    """Abstract interface for background task scheduling."""

    def submit(self, analyzer: AnalysisBase, state: PluginState | None = None) -> Any:
        """Submit a background analysis task for execution."""
        ...

    def cancel_all(self) -> None:
        """Cancel all pending tasks."""
        ...

cancel_all()

Cancel all pending tasks.

Source code in src/biopro_sdk/interfaces/i_task_scheduler.py 
15
16
17
def cancel_all(self) -> None:
    """Cancel all pending tasks."""
    ...

submit(analyzer, state=None)

Submit a background analysis task for execution.

Source code in src/biopro_sdk/interfaces/i_task_scheduler.py 
11
12
13
def submit(self, analyzer: AnalysisBase, state: PluginState | None = None) -> Any:
    """Submit a background analysis task for execution."""
    ...