feat: initial log

Author: brian14708

pyproject.toml

@@ -5,7 +5,6 @@ description = ""
 readme = "README.md"
 requires-python = ">=3.10"
 dependencies = [
-    "pydantic>=2.11.7",
     "typing-extensions>=4.15.0",
 ]
 

src/duron/log/__init__.py

@@ -0,0 +1,4 @@
+from .entry import Entry
+from .storage import BaseLogStorage, FileLogStorage, Lease, Offset
+
+__all__ = ["Entry", "BaseLogStorage", "Offset", "Lease", "FileLogStorage"]

src/duron/log/entry.py

@@ -0,0 +1,57 @@
+from typing import Literal
+
+from typing_extensions import TypedDict
+
+
+class BaseEntry(TypedDict):
+    id: str
+    timestamp: int
+    metadata: dict[str, str]
+
+
+class ErrorInfo(TypedDict):
+    code: int
+    message: str
+    data: object
+
+
+class PromiseCreateEntry(BaseEntry):
+    type: Literal["promise/create"]
+
+
+class PromiseCompleteEntry(BaseEntry):
+    type: Literal["promise/complete"]
+    promise_id: str
+    result: object
+    error: ErrorInfo | None
+
+
+class StreamCreateEntry(BaseEntry):
+    type: Literal["stream/create"]
+
+
+class StreamEmitEntry(BaseEntry):
+    type: Literal["stream/emit"]
+    stream_id: str
+    value: object
+    state: object
+
+
+class StreamCloseEntry(BaseEntry):
+    type: Literal["stream/close"]
+    stream_id: str
+    error: ErrorInfo | None
+
+
+class UnknownEntry(BaseEntry):
+    type: str
+
+
+Entry = (
+    PromiseCreateEntry
+    | PromiseCompleteEntry
+    | StreamCreateEntry
+    | StreamEmitEntry
+    | StreamCloseEntry
+    | UnknownEntry
+)

src/duron/log/storage/__init__.py

@@ -0,0 +1,4 @@
+from .base import BaseLogStorage, Lease, Offset
+from .file import FileLogStorage
+
+__all__ = ["BaseLogStorage", "Offset", "Lease", "FileLogStorage"]

src/duron/log/storage/base.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING, NewType
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncGenerator
+
+    from ..entry import Entry
+
+Offset = NewType("Offset", bytes)
+Lease = NewType("Lease", bytes)
+
+
+class BaseLogStorage(ABC):
+    @abstractmethod
+    def stream(
+        self, start: Offset | None, live: bool
+    ) -> AsyncGenerator[tuple[Offset, Entry], None]: ...
+
+    @abstractmethod
+    async def acquire_lease(self) -> Lease: ...
+
+    @abstractmethod
+    async def release_lease(self, token: Lease): ...
+
+    @abstractmethod
+    async def append(self, token: Lease, entry: Entry) -> Offset: ...

src/duron/log/storage/file.py

@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+import asyncio
+import json
+import threading
+import uuid
+from pathlib import Path
+from typing import TYPE_CHECKING, cast
+
+from typing_extensions import override
+
+from .base import BaseLogStorage, Lease, Offset
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncGenerator
+
+    from ..entry import Entry
+
+
+class FileLogStorage(BaseLogStorage):
+    log_file: Path
+    _leases: Lease | None
+    _lock: threading.Lock
+
+    def __init__(self, log_file: str | Path):
+        self.log_file = Path(log_file)
+        self.log_file.parent.mkdir(parents=True, exist_ok=True)
+        self._leases = None
+        self._lock = threading.Lock()
+
+    @override
+    async def stream(
+        self, start: Offset | None, live: bool
+    ) -> AsyncGenerator[tuple[Offset, Entry], None]:
+        if not self.log_file.exists():
+            return
+
+        start_offset: int = int.from_bytes(start, "little") if start is not None else 0
+        current_line = 0
+
+        with open(self.log_file) as f:
+            # Skip to start offset
+            for _ in range(start_offset):
+                _ = f.readline()
+                current_line += 1
+
+            # Read existing lines from start offset
+            while True:
+                line = f.readline()
+                if line:
+                    try:
+                        entry = json.loads(line.strip())
+                        if isinstance(entry, dict):
+                            yield (
+                                Offset(current_line.to_bytes(8, "little")),
+                                cast("Entry", cast("object", entry)),
+                            )
+                    except json.JSONDecodeError:
+                        pass
+                    current_line += 1
+                else:
+                    # Reached end of file
+                    break
+
+            # If live mode, continue tailing
+            if live:
+                while True:
+                    line = f.readline()
+                    if line:
+                        try:
+                            entry = json.loads(line.strip())
+                            if isinstance(entry, dict):
+                                yield (
+                                    Offset(current_line.to_bytes(8, "little")),
+                                    cast("Entry", cast("object", entry)),
+                                )
+                        except json.JSONDecodeError:
+                            pass
+                        current_line += 1
+                    else:
+                        await asyncio.sleep(0.1)
+
+    @override
+    async def acquire_lease(self) -> Lease:
+        lease_id = Lease(uuid.uuid4().bytes)
+        with self._lock:
+            self._leases = lease_id
+        return lease_id
+
+    @override
+    async def release_lease(self, token: Lease) -> None:
+        with self._lock:
+            if token == self._leases:
+                self._leases = None
+
+    @override
+    async def append(self, token: Lease, entry: Entry) -> Offset:
+        with self._lock:
+            if token != self._leases:
+                raise ValueError("Invalid lease token")
+
+            with open(self.log_file, "a") as f:
+                json.dump(entry, f, separators=(",", ":"))
+                _ = f.write("\n")
+
+            with open(self.log_file) as f:
+                lines = f.readlines()
+
+            return Offset((len(lines) - 1).to_bytes(8, "little"))