Core Framework API

loglife.core

Chat interface core package (transports, clients, shared protocols).

Message dataclass

Normalized representation of transport messages.

Source code in src/loglife/core/messaging.py

@dataclass(slots=True)
class Message:
    """Normalized representation of transport messages."""

    sender: str
    msg_type: str
    raw_payload: str
    client_type: str
    metadata: dict[str, Any] = field(default_factory=dict)
    attachments: dict[str, Any] = field(default_factory=dict)

    @classmethod
    def from_payload(cls, payload: Mapping[str, Any]) -> "Message":
        """Construct a message from a raw transport payload."""
        return cls(
            sender=payload["sender"],
            msg_type=payload["msg_type"],
            raw_payload=payload.get("raw_msg", ""),
            client_type=payload.get("client_type", "unknown"),
            metadata=dict(payload.get("metadata") or {}),
        )

from_payload(payload) classmethod

Construct a message from a raw transport payload.

Source code in src/loglife/core/messaging.py

@classmethod
def from_payload(cls, payload: Mapping[str, Any]) -> "Message":
    """Construct a message from a raw transport payload."""
    return cls(
        sender=payload["sender"],
        msg_type=payload["msg_type"],
        raw_payload=payload.get("raw_msg", ""),
        client_type=payload.get("client_type", "unknown"),
        metadata=dict(payload.get("metadata") or {}),
    )

init()

Initialize the core system (DB, Logging, Workers).

Call this at the start of your application.

Source code in src/loglife/core/interface.py

def init() -> None:
    """Initialize the core system (DB, Logging, Workers).

    Call this at the start of your application.
    """
    setup_logging()
    init_db()

    # Start the sender worker to handle outbound traffic
    start_sender_worker()

recv_msg(block=True, timeout=None)

Receive the next message from the inbound queue.

Blocks until a message is available unless block=False.

Source code in src/loglife/core/interface.py

def recv_msg(block: bool = True, timeout: float | None = None) -> Message:  # noqa: FBT001, FBT002
    """Receive the next message from the inbound queue.

    Blocks until a message is available unless block=False.
    """
    return _inbound_queue.get(block=block, timeout=timeout)

send_msg(message, to=None)

Send a message to the outbound queue.

Parameters:

Name	Type	Description	Default
message	Message | str	A Message object OR a string text.	required
to	str | None	The phone number to send to (required if message is a string).	None

Source code in src/loglife/core/interface.py

def send_msg(message: Message | str, to: str | None = None) -> None:
    """Send a message to the outbound queue.

    Args:
        message: A Message object OR a string text.
        to: The phone number to send to (required if message is a string).

    """
    if isinstance(message, str):
        if not to:
            msg = "Target 'to' number is required when sending a string."
            raise ValueError(msg)
        queue_async_message(to, message)
    else:
        enqueue_outbound_message(message)