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:
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)