Skip to content

Interfaces

karcytics_sdk.interfaces

ICrashReporter

Bases: Protocol

Abstract interface for reporting errors to the Hub's diagnostic engine.

Implemented today by both the Hub's real karcytics.core.diagnostics. DiagnosticEngine (in-process plugins reach it indirectly via their own logger and the AutoReportHandler) and karcytics_sdk.plugin. runtime_services.DiagnosticsForwarder (isolated plugins, which have no other route back to the Hub's diagnostics engine at all). Call from inside the except: block handling exception so a real traceback is captured, not just the exception's message.

Source code in src/karcytics_sdk/interfaces/i_crash_reporter.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
@runtime_checkable
class ICrashReporter(Protocol):
    """Abstract interface for reporting errors to the Hub's diagnostic engine.

    Implemented today by both the Hub's real `karcytics.core.diagnostics.
    DiagnosticEngine` (in-process plugins reach it indirectly via their own
    logger and the AutoReportHandler) and `karcytics_sdk.plugin.
    runtime_services.DiagnosticsForwarder` (isolated plugins, which have no
    other route back to the Hub's diagnostics engine at all). Call from
    inside the `except:` block handling `exception` so a real traceback is
    captured, not just the exception's message.
    """

    def report_error(
        self,
        message: str,
        exception: BaseException | None = None,
        plugin_id: str | None = None,
        fatal: bool = False,
    ) -> None:
        """Report an error to the Hub's diagnostic engine."""
        ...

report_error(message, exception=None, plugin_id=None, fatal=False)

Report an error to the Hub's diagnostic engine.

Source code in src/karcytics_sdk/interfaces/i_crash_reporter.py 
17
18
19
20
21
22
23
24
25
def report_error(
    self,
    message: str,
    exception: BaseException | None = None,
    plugin_id: str | None = None,
    fatal: bool = False,
) -> None:
    """Report an error to the Hub's diagnostic engine."""
    ...

IEventBus

Bases: Protocol

Abstract interface for the global event bus.

Source code in src/karcytics_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/karcytics_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/karcytics_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/karcytics_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/karcytics_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/karcytics_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/karcytics_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/karcytics_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/karcytics_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/karcytics_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/karcytics_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/karcytics_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."""
    ...