diff env/lib/python3.7/site-packages/boto/sdb/db/sequence.py @ 5:9b1c78e6ba9c draft default tip

"planemo upload commit 6c0a8142489327ece472c84e558c47da711a9142"
author shellac
date Mon, 01 Jun 2020 08:59:25 -0400
parents 79f47841a781
children
line wrap: on
line diff
--- a/env/lib/python3.7/site-packages/boto/sdb/db/sequence.py	Thu May 14 16:47:39 2020 -0400
+++ /dev/null	Thu Jan 01 00:00:00 1970 +0000
@@ -1,224 +0,0 @@
-# Copyright (c) 2010 Chris Moyer http://coredumped.org/
-#
-# Permission is hereby granted, free of charge, to any person obtaining a
-# copy of this software and associated documentation files (the
-# "Software"), to deal in the Software without restriction, including
-# without limitation the rights to use, copy, modify, merge, publish, dis-
-# tribute, sublicense, and/or sell copies of the Software, and to permit
-# persons to whom the Software is furnished to do so, subject to the fol-
-# lowing conditions:
-#
-# The above copyright notice and this permission notice shall be included
-# in all copies or substantial portions of the Software.
-#
-# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
-# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL-
-# ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
-# SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, 
-# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
-# IN THE SOFTWARE.
-
-from boto.exception import SDBResponseError
-from boto.compat import six
-
-class SequenceGenerator(object):
-    """Generic Sequence Generator object, this takes a single
-    string as the "sequence" and uses that to figure out
-    what the next value in a string is. For example
-    if you give "ABC" and pass in "A" it will give you "B",
-    and if you give it "C" it will give you "AA".
-
-    If you set "rollover" to True in the above example, passing
-    in "C" would give you "A" again.
-
-    The Sequence string can be a string or any iterable
-    that has the "index" function and is indexable.
-    """
-    __name__ = "SequenceGenerator"
-
-    def __init__(self, sequence_string, rollover=False):
-        """Create a new SequenceGenerator using the sequence_string
-        as how to generate the next item.
-
-        :param sequence_string: The string or list that explains
-        how to generate the next item in the sequence
-        :type sequence_string: str,iterable
-
-        :param rollover: Rollover instead of incrementing when
-        we hit the end of the sequence
-        :type rollover: bool
-        """
-        self.sequence_string = sequence_string
-        self.sequence_length = len(sequence_string[0])
-        self.rollover = rollover
-        self.last_item = sequence_string[-1]
-        self.__name__ = "%s('%s')" % (self.__class__.__name__, sequence_string)
-
-    def __call__(self, val, last=None):
-        """Get the next value in the sequence"""
-        # If they pass us in a string that's not at least
-        # the lenght of our sequence, then return the
-        # first element in our sequence
-        if val is None or len(val) < self.sequence_length:
-            return self.sequence_string[0]
-        last_value = val[-self.sequence_length:]
-        if (not self.rollover) and (last_value == self.last_item):
-            val = "%s%s" % (self(val[:-self.sequence_length]), self._inc(last_value))
-        else:
-            val = "%s%s" % (val[:-self.sequence_length], self._inc(last_value))
-        return val
-
-    def _inc(self, val):
-        """Increment a single value"""
-        assert(len(val) == self.sequence_length)
-        return self.sequence_string[(self.sequence_string.index(val) + 1) % len(self.sequence_string)]
-
-
-#
-# Simple Sequence Functions
-#
-def increment_by_one(cv=None, lv=None):
-    if cv is None:
-        return 0
-    return cv + 1
-
-def double(cv=None, lv=None):
-    if cv is None:
-        return 1
-    return cv * 2
-
-def fib(cv=1, lv=0):
-    """The fibonacci sequence, this incrementer uses the
-    last value"""
-    if cv is None:
-        cv = 1
-    if lv is None:
-        lv = 0
-    return cv + lv
-
-increment_string = SequenceGenerator("ABCDEFGHIJKLMNOPQRSTUVWXYZ")
-
-
-class Sequence(object):
-    """A simple Sequence using the new SDB "Consistent" features
-    Based largly off of the "Counter" example from mitch garnaat:
-    http://bitbucket.org/mitch/stupidbototricks/src/tip/counter.py"""
-
-    def __init__(self, id=None, domain_name=None, fnc=increment_by_one, init_val=None):
-        """Create a new Sequence, using an optional function to
-        increment to the next number, by default we just increment by one.
-        Every parameter here is optional, if you don't specify any options
-        then you'll get a new SequenceGenerator with a random ID stored in the
-        default domain that increments by one and uses the default botoweb
-        environment
-
-        :param id: Optional ID (name) for this counter
-        :type id: str
-
-        :param domain_name: Optional domain name to use, by default we get this out of the
-            environment configuration
-        :type domain_name:str
-
-        :param fnc: Optional function to use for the incrementation, by default we just increment by one
-            There are several functions defined in this module.
-            Your function must accept "None" to get the initial value
-        :type fnc: function, str
-
-        :param init_val: Initial value, by default this is the first element in your sequence,
-            but you can pass in any value, even a string if you pass in a function that uses
-            strings instead of ints to increment
-        """
-        self._db = None
-        self._value = None
-        self.last_value = None
-        self.domain_name = domain_name
-        self.id = id
-        if init_val is None:
-            init_val = fnc(init_val)
-
-        if self.id is None:
-            import uuid
-            self.id = str(uuid.uuid4())
-
-        self.item_type = type(fnc(None))
-        self.timestamp = None
-        # Allow us to pass in a full name to a function
-        if isinstance(fnc, six.string_types):
-            from boto.utils import find_class
-            fnc = find_class(fnc)
-        self.fnc = fnc
-
-        # Bootstrap the value last
-        if not self.val:
-            self.val = init_val
-
-    def set(self, val):
-        """Set the value"""
-        import time
-        now = time.time()
-        expected_value = []
-        new_val = {}
-        new_val['timestamp'] = now
-        if self._value is not None:
-            new_val['last_value'] = self._value
-            expected_value = ['current_value', str(self._value)]
-        new_val['current_value'] = val
-        try:
-            self.db.put_attributes(self.id, new_val, expected_value=expected_value)
-            self.timestamp = new_val['timestamp']
-        except SDBResponseError as e:
-            if e.status == 409:
-                raise ValueError("Sequence out of sync")
-            else:
-                raise
-
-
-    def get(self):
-        """Get the value"""
-        val = self.db.get_attributes(self.id, consistent_read=True)
-        if val:
-            if 'timestamp' in val:
-                self.timestamp = val['timestamp']
-            if 'current_value' in val:
-                self._value = self.item_type(val['current_value'])
-            if "last_value" in val and val['last_value'] is not None:
-                self.last_value = self.item_type(val['last_value'])
-        return self._value
-
-    val = property(get, set)
-
-    def __repr__(self):
-        return "%s('%s', '%s', '%s.%s', '%s')" % (
-            self.__class__.__name__,
-            self.id,
-            self.domain_name,
-            self.fnc.__module__, self.fnc.__name__,
-            self.val)
-
-
-    def _connect(self):
-        """Connect to our domain"""
-        if not self._db:
-            import boto
-            sdb = boto.connect_sdb()
-            if not self.domain_name:
-                self.domain_name = boto.config.get("DB", "sequence_db", boto.config.get("DB", "db_name", "default"))
-            try:
-                self._db = sdb.get_domain(self.domain_name)
-            except SDBResponseError as e:
-                if e.status == 400:
-                    self._db = sdb.create_domain(self.domain_name)
-                else:
-                    raise
-        return self._db
-
-    db = property(_connect)
-
-    def next(self):
-        self.val = self.fnc(self.val, self.last_value)
-        return self.val
-
-    def delete(self):
-        """Remove this sequence"""
-        self.db.delete_attributes(self.id)