The SSH host key has changed on 8 April, 2022 to this one: SHA256:573uTBSeh74kvOo0HJXi5ijdzRm8me27suzNEDlGyrQ
Python implementation of [Earthstar](https://earthstar-project.org/)
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.
earthsnake/earthsnake/replica/__init__.py

96 lines
2.8 KiB

"""Replica related things"""
from abc import ABC, abstractmethod
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 IngestEvent:
pass
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):
pass
@property
@abstractmethod
def is_closed(self):
"""Tells whether the replica is closed or not
"""
@abstractmethod
def close(self):
"""Closes the replica, preventing further use
Any method called after closing will return ``ReplicaIsClosedError``
"""
@property
@abstractmethod
def max_local_index(self) -> int:
"""Returns the maximum local index of all stored documents"""
@abstractmethod
def get_docs_after_local_index(self) -> List[Document]:
"""Get all documents after a specific local index"""
@abstractmethod
def get_all_docs(self) -> List[Document]:
"""Get all documents, including historical versions by other identities"""
@abstractmethod
def get_latest_docs(self) -> List[Document]:
"""Get the latest from every path"""
@abstractmethod
def get_all_docs_at_path(self, path: Path) -> List[Document]:
"""Get all versions of a document by different authors from a specific path"""
@abstractmethod
def get_latest_doc_at_path(self, path: Path) -> Optional[Document]:
"""Get the most recent version of a document at a specific path"""
@abstractmethod
def query_docs(self, query: Query) -> List[Document]:
"""Get a list of documents matching a given query"""
@abstractmethod
def query_paths(self, query: Query) -> List[Path]:
"""Get all document paths where documents match a given query"""
@abstractmethod
def query_authors(self, query: Query) -> List[Identity]:
"""Get all document authors where documents match a given query"""
@abstractmethod
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.
"""
@abstractmethod
def ingest(self, document: Document) -> IngestEvent:
"""Ingest an existing, signed document to the replica"""
@abstractmethod
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.
"""