72 lines
1.7 KiB
Python
72 lines
1.7 KiB
Python
from abc import ABC, abstractmethod
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import List, Optional, Sequence, Tuple, Union
|
|
|
|
from ..utilities import ConfigFile
|
|
from ..views import Filter, SortBy, Trace
|
|
|
|
|
|
class TracingDatabaseDriver(ABC):
|
|
is_production_ready: bool
|
|
initialized: bool = False
|
|
|
|
@classmethod
|
|
def configure_credentials_from_file(
|
|
cls,
|
|
secrets: Union[Path, str, ConfigFile],
|
|
) -> None:
|
|
if not isinstance(secrets, ConfigFile):
|
|
secrets = ConfigFile(secrets)
|
|
cls.configure_credentials(**{k.lower(): v for k, v in secrets.items()})
|
|
|
|
@classmethod
|
|
def configure_credentials(
|
|
cls,
|
|
) -> None:
|
|
cls.initialized = True
|
|
|
|
@abstractmethod
|
|
def save(self, document: Trace) -> str:
|
|
pass
|
|
|
|
@abstractmethod
|
|
def save_batch(
|
|
self,
|
|
documents: List[Trace],
|
|
) -> List[str]:
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get(self, id: str) -> Optional[Trace]:
|
|
pass
|
|
|
|
@abstractmethod
|
|
def query(
|
|
self,
|
|
*,
|
|
skip: int = 0,
|
|
take: Optional[int] = None,
|
|
conjunctive_filters: Sequence[Filter] = [],
|
|
conjunctive_tags: Sequence[str] = [],
|
|
until: Optional[datetime] = None,
|
|
since: Optional[datetime] = None,
|
|
has_feedback: Optional[bool] = None,
|
|
sort_by: Sequence[SortBy] = []
|
|
) -> Tuple[List[Trace], int]:
|
|
pass
|
|
|
|
@abstractmethod
|
|
def update(self, id: str, new_version: Trace) -> None:
|
|
pass
|
|
|
|
@abstractmethod
|
|
def delete(self, id: str) -> None:
|
|
pass
|
|
|
|
@abstractmethod
|
|
def delete_batch(
|
|
self,
|
|
ids: List[str],
|
|
) -> None:
|
|
pass
|