Source code for couchbase.durability

from functools import wraps
from typing import *

from couchbase_core.supportability import internal

from .options import Cardinal, OptionBlock, OptionBlockTimeOut
from couchbase_core.durability import Durability
from datetime import timedelta

try:
    from typing import TypedDict
except BaseException:
    from typing_extensions import TypedDict

ReplicateTo = Cardinal
PersistTo = Cardinal


T = TypeVar('T', bound=OptionBlock)


class DurabilityTypeBase(dict):
    def __init__(self, content):
        super(DurabilityTypeBase, self).__init__(**content)


[docs]class DurabilityType(dict):
[docs] @internal def __init__(self, # type: DurabilityType content # type: Dict[str, Any] ): # type: (...) -> None """ Durability configuration options :param content: dictionary passed up from subclasses """ super(DurabilityType, self).__init__(content)
[docs]class ClientDurability(DurabilityType): Storage = TypedDict( 'Storage', { 'replicate_to': ReplicateTo, 'persist_to': PersistTo}, total=True)
[docs] def __init__(self, # type: T replicate_to=ReplicateTo.NONE, # type: ReplicateTo persist_to=PersistTo.NONE # type: PersistTo ): # type: (...) -> None """ Client Durability :param persist_to: If set, wait for the item to be removed from the storage of at least these many nodes :param replicate_to: If set, wait for the item to be removed from the cache of at least these many nodes (excluding the master) """ super( ClientDurability, self).__init__( ClientDurability.Storage( replicate_to=replicate_to, persist_to=persist_to))
[docs]class ServerDurability(DurabilityType): Storage = TypedDict('Storage', {'level': Durability}, total=True)
[docs] def __init__(self, # type: ServerDurability level, # type: Durability ): # type: (...) -> None """ Server-based Durability (Synchronous Replication) :param Durability level: durability level """ super( ServerDurability, self).__init__( ServerDurability.Storage( level=level))
[docs]class ClientDurableOptionBlock(OptionBlockTimeOut):
[docs] def __init__(self, # type: ClientDurableOptionBlock timeout=None, # type: timedelta durability=None # type: ClientDurability ): # type: (...) -> None """ Options for operations with client-type durability :param durability: Client durability settings :param timeout: Timeout for operation """ super( ClientDurableOptionBlock, self).__init__( durability=durability, timeout=timeout)
[docs]class ServerDurableOptionBlock(OptionBlockTimeOut):
[docs] def __init__(self, # type: ServerDurableOptionBlock timeout=None, # type: timedelta durability=None # type: ServerDurability ): # type: (...) -> None """ Options for operations with server-type durability :param durability: Server durability settings :param timeout: Timeout for operation """ super( ServerDurableOptionBlock, self).__init__( durability=durability, timeout=timeout)
[docs]class DurabilityOptionBlock(OptionBlockTimeOut):
[docs] def __init__(self, # type: DurabilityOptionBlock timeout=None, # type: timedelta durability=None, # type: DurabilityType expiry=None, # type: timedelta **kwargs): # type: (...) -> None """ Options for operations with any type of durability :param durability: Durability settings :param expiry: When any mutation should expire :param timeout: Timeout for operation """ super( DurabilityOptionBlock, self).__init__( durability=durability, expiry=expiry, timeout=timeout, **kwargs)
@property def expiry(self): return self.get('expiry', None)