Skip to content

fathom.FactStore

fathom.FactStore

Bases: Protocol

Protocol defining the async fact-storage interface.

Source code in src/fathom/fleet.py 
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."""
        ...

assert_fact(template, data) async

Assert a fact and return its unique fact_id.

Source code in src/fathom/fleet.py 
38
39
40
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 
42
43
44
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 
46
47
48
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 
50
51
52
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.

Source code in src/fathom/fleet.py 
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."""
    ...