Source code for minos.common.database.operations

from __future__ import (
    annotations,
)

from abc import (
    ABC,
)
from collections.abc import (
    Hashable,
    Iterable,
)
from typing import (
    Optional,
)


[docs]class DatabaseOperation(ABC): """Database Operation base class."""
[docs] def __init__(self, *args, lock: Optional[Hashable] = None, timeout: Optional[float] = None, **kwargs): self.lock = lock self.timeout = timeout
[docs]class ComposedDatabaseOperation(DatabaseOperation): """Composed Database Operation class."""
[docs] def __init__(self, operations: Iterable[DatabaseOperation], *args, **kwargs): super().__init__(*args, **kwargs) self.operations = tuple(operations)
[docs]class DatabaseOperationFactory(ABC): """Database Operation Factory base class."""