Source code for harp.http.typing.messages

from datetime import UTC, datetime
from typing import Optional

from multidict import MultiDictProxy



[docs]
class BaseMessage:
    protocol: str
    kind: str
    created_at: datetime


[docs]
    def __init__(self, *, extensions: Optional[dict] = None):
        self.created_at = datetime.now(UTC)

        self._extensions = extensions or {}


    @property
    def extensions(self) -> dict:
        return self._extensions


[docs]
    async def aread(self):
        pass





[docs]
class BaseHttpMessage(BaseMessage):
    protocol = "http"

    @property
    def headers(self) -> MultiDictProxy:
        raise NotImplementedError

    @property
    def body(self) -> bytes:
        raise NotImplementedError