The SSH host key has changed on 8 April, 2022 to this one: SHA256:573uTBSeh74kvOo0HJXi5ijdzRm8me27suzNEDlGyrQ
Python implementation of [Earthstar](
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

107 lines
3.0 KiB

"""Replica related things"""
from abc import ABC, abstractmethod
from enum import Enum, auto
from typing import List, Optional
from ..document import Document
from ..identity import Identity
from ..path import Path
from ..query import Query
from ..share import Share
class HistoryMode(Enum):
ALL = auto()
LATEST = auto()
class IngestEvent:
class Replica(ABC):
"""A replica of a share’s data
Used to read, write, and synchronise data to.
Should be closed using the ``close()`` method when no longer being used.
def __init__(self, share: Share, **driver_kwargs):
def is_closed(self):
"""Tells whether the replica is closed or not
def close(self):
"""Closes the replica, preventing further use
Any method called after closing will return ``ReplicaIsClosedError``
def max_local_index(self) -> int:
"""Returns the maximum local index of all stored documents"""
def get_docs_after_local_index(
history_mode: HistoryMode,
start_after: int,
limit: Optional[int] = None,
) -> List[Document]:
"""Get all documents after a specific local index"""
def get_all_docs(self) -> List[Document]:
"""Get all documents, including historical versions by other identities"""
def get_latest_docs(self) -> List[Document]:
"""Get the latest from every path"""
def get_all_docs_at_path(self, path: Path) -> List[Document]:
"""Get all versions of a document by different authors from a specific path"""
def get_latest_doc_at_path(self, path: Path) -> Optional[Document]:
"""Get the most recent version of a document at a specific path"""
def query_docs(self, query: Query) -> List[Document]:
"""Get a list of documents matching a given query"""
def query_paths(self, query: Query) -> List[Path]:
"""Get all document paths where documents match a given query"""
def query_authors(self, query: Query) -> List[Identity]:
"""Get all document authors where documents match a given query"""
def set(self, doc_to_set: Document) -> IngestEvent:
"""Add a new document to the replica
If a document signed by the same identity exists at the same path, it will be overwritten.
def ingest(self, document: Document) -> IngestEvent:
"""Ingest an existing, signed document to the replica"""
def overwrite_docs_by_author(self, identity: Identity) -> int:
"""Overwrite every document from this author with an empty doc
This includes historical versions of documents.
:returns: the number of documents changed.