Skip to content

fathom.FactStore

fathom.FactStore

Bases: Protocol

Protocol defining the async fact-storage interface.

Source code in src/fathom/fleet.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
@runtime_checkable
class FactStore(Protocol):
    """Protocol defining the async fact-storage interface."""

    async def assert_fact(self, template: str, data: FactData) -> FactId:
        """Assert a fact and return its unique fact_id."""
        ...

    async def query(self, template: str, fact_filter: FactFilter | None = None) -> list[FactData]:
        """Return facts matching the template and optional filter."""
        ...

    async def retract(self, template: str, fact_filter: FactFilter | None = None) -> int:
        """Retract facts matching the template and optional filter. Return count removed."""
        ...

    async def count(self, template: str, fact_filter: FactFilter | None = None) -> int:
        """Count facts matching the template and optional filter."""
        ...

    async def subscribe(
        self,
        template: str,
        callback: Callable[[FactChangeNotification], Coroutine[Any, Any, None]],
    ) -> Callable[[], None]:
        """Subscribe to changes on a template. Return an unsubscribe callable.

        A change is delivered once to each subscriber. A callback that raises
        is logged and skipped: it never fails the write that caused the
        change, and it never ends the subscription or reaches the other
        subscribers.
        """
        ...

assert_fact(template, data) async

Assert a fact and return its unique fact_id.

Source code in src/fathom/fleet.py 
32
33
34
async def assert_fact(self, template: str, data: FactData) -> FactId:
    """Assert a fact and return its unique fact_id."""
    ...

query(template, fact_filter=None) async

Return facts matching the template and optional filter.

Source code in src/fathom/fleet.py 
36
37
38
async def query(self, template: str, fact_filter: FactFilter | None = None) -> list[FactData]:
    """Return facts matching the template and optional filter."""
    ...

retract(template, fact_filter=None) async

Retract facts matching the template and optional filter. Return count removed.

Source code in src/fathom/fleet.py 
40
41
42
async def retract(self, template: str, fact_filter: FactFilter | None = None) -> int:
    """Retract facts matching the template and optional filter. Return count removed."""
    ...

count(template, fact_filter=None) async

Count facts matching the template and optional filter.

Source code in src/fathom/fleet.py 
44
45
46
async def count(self, template: str, fact_filter: FactFilter | None = None) -> int:
    """Count facts matching the template and optional filter."""
    ...

subscribe(template, callback) async

Subscribe to changes on a template. Return an unsubscribe callable.

A change is delivered once to each subscriber. A callback that raises is logged and skipped: it never fails the write that caused the change, and it never ends the subscription or reaches the other subscribers.

Source code in src/fathom/fleet.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
async def subscribe(
    self,
    template: str,
    callback: Callable[[FactChangeNotification], Coroutine[Any, Any, None]],
) -> Callable[[], None]:
    """Subscribe to changes on a template. Return an unsubscribe callable.

    A change is delivered once to each subscriber. A callback that raises
    is logged and skipped: it never fails the write that caused the
    change, and it never ends the subscription or reaches the other
    subscribers.
    """
    ...