azad.network.interfaces Module ¶

azad.network.interfaces ¶

Network protocol interfaces.

Attributes ¶

T `module-attribute` ¶

T = TypeVar('T')

TaskStartCallback `module-attribute` ¶

TaskStartCallback = Callable[[str, TaskConfig, Optional[str]], Awaitable[DataResponse[TaskStatusResponse]]]

TaskExecuteCallback `module-attribute` ¶

TaskExecuteCallback = Callable[[str], Awaitable[None]]

TaskAbortCallback `module-attribute` ¶

TaskAbortCallback = Callable[[str, bool], Awaitable[MessageResponse]]

TaskStatusCallback `module-attribute` ¶

TaskStatusCallback = Callable[[Optional[str]], Awaitable[DataResponse[TaskStatusResponse]]]

ShutdownCallback `module-attribute` ¶

ShutdownCallback = Callable[[bool], Awaitable[MessageResponse]]

UpdateConfigCallback `module-attribute` ¶

UpdateConfigCallback = Callable[[TaskConfig], Awaitable[MessageResponse]]

EnvironmentRegisterCallback `module-attribute` ¶

EnvironmentRegisterCallback = Callable[[ProtocolHandler, bool], Awaitable[MessageResponse]]

EnvironmentUnregisterCallback `module-attribute` ¶

EnvironmentUnregisterCallback = Callable[[ProtocolHandler], Awaitable[MessageResponse]]

LoopAdvanceCallback `module-attribute` ¶

LoopAdvanceCallback = Callable[[str], Awaitable[MessageResponse]]

LoopResumeCallback `module-attribute` ¶

LoopResumeCallback = Callable[[str, list[Any]], Awaitable[MessageResponse]]

HardPauseCallback `module-attribute` ¶

HardPauseCallback = Callable[[], Awaitable[DataResponse[dict[str, Any]]]]

Classes ¶

BaseResponse ¶

Bases: BaseModel

Base response model.

Attributes ¶

success `instance-attribute` ¶

success: bool

SuccessResponse ¶

Bases: BaseResponse

Base for successful responses.

Attributes ¶

success `class-attribute` `instance-attribute` ¶

success: bool = True

ErrorResponse ¶

Bases: BaseResponse

Response with an error message.

Attributes ¶

success `class-attribute` `instance-attribute` ¶

success: bool = False

error `instance-attribute` ¶

error: str

MessageResponse ¶

Bases: SuccessResponse

Response with a message string.

Attributes ¶

message `instance-attribute` ¶

message: str

success `class-attribute` `instance-attribute` ¶

success: bool = True

DataResponse ¶

Bases: SuccessResponse, Generic[T]

Response with generic data payload.

Attributes ¶

data `instance-attribute` ¶

data: T

success `class-attribute` `instance-attribute` ¶

success: bool = True

TaskStatusResponse ¶

Bases: BaseModel

Task status response data.

Attributes ¶

task_id `instance-attribute` ¶

task_id: str

status `instance-attribute` ¶

status: str

error `class-attribute` `instance-attribute` ¶

error: Optional[str] = None

PENDING: Task is waiting to start RUNNING: Task is currently executing COMPLETED: Task completed successfully FAILED: Task failed with an error ABORTED: Task was forcefully aborted TERMINATED: Task ended normally (e.g. server shutdown)

Attributes ¶

PENDING `class-attribute` `instance-attribute` ¶

PENDING = 'pending'

RUNNING `class-attribute` `instance-attribute` ¶

RUNNING = 'running'

COMPLETED `class-attribute` `instance-attribute` ¶

COMPLETED = 'completed'

FAILED `class-attribute` `instance-attribute` ¶

FAILED = 'failed'

ABORTED `class-attribute` `instance-attribute` ¶

ABORTED = 'aborted'

TERMINATED `class-attribute` `instance-attribute` ¶

TERMINATED = 'terminated'

EventData ¶

Bases: BaseModel

Event data with type and payload.

Attributes ¶

type `instance-attribute` ¶

type: str

data `instance-attribute` ¶

data: dict[str, Any]

timestamp `instance-attribute` ¶

timestamp: str

LogData ¶

Bases: BaseModel

Log event data.

Attributes ¶

level `instance-attribute` ¶

level: str

message `instance-attribute` ¶

message: str

logger `instance-attribute` ¶

logger: str

extra `class-attribute` `instance-attribute` ¶

extra: Optional[dict[str, Any]] = None

TaskResult `dataclass` ¶

TaskResult(task_id: str, status: TaskStatus, error: Optional[str] = None, forced: bool = False)

Result of a task operation.

Attributes ¶

task_id `instance-attribute` ¶

task_id: str

status `instance-attribute` ¶

status: TaskStatus

error `class-attribute` `instance-attribute` ¶

error: Optional[str] = None

forced `class-attribute` `instance-attribute` ¶

forced: bool = False

ProtocolCallbacks `dataclass` ¶

ProtocolCallbacks(on_task_start: TaskStartCallback, on_task_execute: TaskExecuteCallback, on_task_abort: TaskAbortCallback, on_status_request: TaskStatusCallback, on_shutdown: ShutdownCallback, on_environment_register: EnvironmentRegisterCallback, on_environment_unregister: EnvironmentUnregisterCallback)

Callbacks for protocol operations.

Attributes ¶

on_task_start `instance-attribute` ¶

on_task_start: TaskStartCallback

on_task_execute `instance-attribute` ¶

on_task_execute: TaskExecuteCallback

on_task_abort `instance-attribute` ¶

on_task_abort: TaskAbortCallback

on_status_request `instance-attribute` ¶

on_status_request: TaskStatusCallback

on_shutdown `instance-attribute` ¶

on_shutdown: ShutdownCallback

on_environment_register `instance-attribute` ¶

on_environment_register: EnvironmentRegisterCallback

on_environment_unregister `instance-attribute` ¶

on_environment_unregister: EnvironmentUnregisterCallback

EventPublisher ¶

Bases: ABC

Abstract base class for event publishing.

Functions ¶

publish_event `abstractmethod` `async` ¶

publish_event(event_type: str, payload: dict[str, Any]) -> None

Publish an event.

Source code in azad/network/interfaces.py

@abstractmethod
async def publish_event(self, event_type: str, payload: dict[str, Any]) -> None:
    """Publish an event."""
    pass

publish_log `abstractmethod` `async` ¶

publish_log(level: str, message: str, logger_name: str, **kwargs) -> None

Publish a log message.

Source code in azad/network/interfaces.py

@abstractmethod
async def publish_log(self, level: str, message: str, logger_name: str, **kwargs) -> None:
    """Publish a log message."""
    pass

EventSubscriber ¶

Bases: ABC

Abstract base class for event subscription.

Functions ¶

subscribe_events `abstractmethod` `async` ¶

subscribe_events(client_id: str) -> AsyncIterator[dict[str, Any]]

Subscribe to events.

Source code in azad/network/interfaces.py

@abstractmethod
async def subscribe_events(self, client_id: str) -> AsyncIterator[dict[str, Any]]:
    """Subscribe to events."""
    pass

subscribe_logs `abstractmethod` `async` ¶

subscribe_logs(client_id: str) -> AsyncIterator[dict[str, Any]]

Subscribe to logs.

Source code in azad/network/interfaces.py

@abstractmethod
async def subscribe_logs(self, client_id: str) -> AsyncIterator[dict[str, Any]]:
    """Subscribe to logs."""
    pass