Source code for couchbase.options

import time
import copy
from typing import *

import couchbase.exceptions
import ctypes

class FiniteDuration(object):
    def __init__(self, seconds  # type: Union[float,int]
                 ):
        self.value = seconds

    @staticmethod
    def time():
        return FiniteDuration(time.time())

    def __float__(self):
        return float(self.value)

    def __int__(self):
        return self.value

    def __add__(self, other):
        result = copy.deepcopy(self)
        result.value += other.value
        return result

    def __lt__(self, other):
        return self.value < other.value

    def __gt__(self, other):
        return self.value > other.value


class Duration(float):
    def __init__(self, seconds  # type: Union[float,int]
                 ):
        # type: (...)->None
        super(Duration, self).__init__(seconds)


class Seconds(FiniteDuration):
    def __init__(self,
                 seconds  # type: Union[float,int]
                 ):
        # type: (...)->None
        super(Seconds, self).__init__(seconds)


class Durations:
    def __init__(self):
        pass

    @staticmethod
    def minutes(minutes  # type: int
                ):
        return Seconds(minutes * 60)

    @staticmethod
    def days(days  # type: int
             ):
        return Durations.minutes(days * 24 * 60)

    @staticmethod
    def seconds(seconds):
        return Seconds(seconds)


class OptionBlock(dict):
    def __init__(self, *args, **kwargs):
        # type: (*Any, **Any) -> None
        super(OptionBlock, self).__init__(**kwargs)


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


class OptionBlockTimeOut(OptionBlock):
    def __init__(self, *args, **kwargs):
        # type: (*Any, **Any) -> None
        super(OptionBlockTimeOut, self).__init__(**kwargs)

    def timeout(self,  # type: T
                duration):
        # type: (...)->T
        self['ttl'] = duration.__int__()
        return self


class Value(object):
    def __init__(self,
                 value,  # type: Union[str,bytes,SupportsInt]
                 **kwargs  # type: Any
                 ):
        self.value = value
        self.kwargs = kwargs

    def __int__(self):
        return self.value

    def __float__(self):
        return self.value


class Cardinal(OptionBlock):
    class ONE(Value):
        def __init__(self, **kwargs):
            super(Cardinal.ONE, self).__init__(1, **kwargs)

    TWO = Value(2)
    THREE = Value(3)
    NONE = Value(0)


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


def forward_args(arg_vars,  # type: Optional[Dict[str,Any]]
                 *options  # type: Tuple[OptionBlockDeriv,...]
                 ):
    # type: (...)->OptionBlockDeriv[str,Any]
    arg_vars = copy.copy(arg_vars) if arg_vars else {}
    temp_options = copy.copy(options[0]) if (options and options[0]) else OptionBlock()
    kwargs = arg_vars.pop('kwargs', {})
    temp_options.update(kwargs)
    temp_options.update(arg_vars)
    arg_mapping = {'spec': {'specs': lambda x: x}, 'id': {},
                   'replicate_to': {"replicate_to":int},
                   'persist_to': {"persist_to":int},
                   'timeout' : {'ttl': int},
                   'expiration': {'ttl': int}, 'self': {}, 'options': {}}
    end_options = {}
    for k, v in temp_options.items():
        map_item = arg_mapping.get(k, None)
        if not (map_item is None):
            for out_k, out_f in map_item.items():
                end_options[out_k] = out_f(v)
        else:
            end_options[k] = v
    return end_options


AcceptableInts = Union['ConstrainedValue', ctypes.c_int64, ctypes.c_uint64, int]


class ConstrainedInt(object):
    def __init__(self,value):
        """
        A signed integer between cls.min() and cls.max() inclusive

        :param couchbase.options.AcceptableInts value: the value to initialise this with.
        :raise: :exc:`~couchbase.exceptions.ArgumentError` if not in range
        """
        self.value = type(self).verified_value(value)

    @classmethod
    def verified_value(cls, item  # type: AcceptableInts
                        ):
        # type: (...)->int
        value = getattr(item, 'value', item)
        if not isinstance(value, int) or not (cls.min()<=value<=cls.max()):
            raise couchbase.exceptions.ArgumentError("Integer in range {} and {} inclusiverequired".format(cls.min(), cls.max()))
        return value

    @classmethod
    def verified(cls,
                 item  # type: AcceptableInts
                 ):
        if isinstance(item, cls):
            return item
        raise couchbase.exceptions.ArgumentError("Argument is not {}".format(cls))

    def __neg__(self):
        return -self.value

    def __int__(self):
        return self.value

    @classmethod
    def max(cls):
        raise NotImplementedError()

    @classmethod
    def min(cls):
        raise NotImplementedError()


[docs]class SignedInt64(ConstrainedInt):
[docs] def __init__(self, value): """ A signed integer between -0x8000000000000000 and +0x7FFFFFFFFFFFFFFF inclusive. :param couchbase.options.AcceptableInts value: the value to initialise this with. :raise: :exc:`~couchbase.exceptions.ArgumentError` if not in range """ super(SignedInt64,self).__init__(value)
@classmethod def max(cls): return 0x7FFFFFFFFFFFFFFF @classmethod def min(cls): return -0x8000000000000000