def add_package(self, owner, package, template, example):
self.prepare_execute_return("INSERT INTO public.contracts (id, owner, package, template, example) \
VALUES (?, ?, ?, ?, ?) IF NOT EXISTS", (uuid.uuid1(), owner, package, template, example))
return self.check_package(owner, package)
python类uuid1()的实例源码
def __init__(self, api_version=1, request_id=None):
self.api_version = api_version
if request_id:
self.request_id = request_id
else:
self.request_id = str(uuid.uuid1())
self.ops = []
def __init__(self, api_version=1, request_id=None):
self.api_version = api_version
if request_id:
self.request_id = request_id
else:
self.request_id = str(uuid.uuid1())
self.ops = []
def acquire(self, blocking=None, blocking_timeout=None):
"""
Use Redis to hold a shared, distributed lock named ``name``.
Returns True once the lock is acquired.
If ``blocking`` is False, always return immediately. If the lock
was acquired, return True, otherwise return False.
``blocking_timeout`` specifies the maximum number of seconds to
wait trying to acquire the lock.
"""
sleep = self.sleep
token = b(uuid.uuid1().hex)
if blocking is None:
blocking = self.blocking
if blocking_timeout is None:
blocking_timeout = self.blocking_timeout
stop_trying_at = None
if blocking_timeout is not None:
stop_trying_at = mod_time.time() + blocking_timeout
while 1:
if self.do_acquire(token):
self.local.token = token
return True
if not blocking:
return False
if stop_trying_at is not None and mod_time.time() > stop_trying_at:
return False
mod_time.sleep(sleep)
def uuid1(self):
raise NotImplementedError()
def uuid1(self):
_LibUUID._libuuid.uuid_generate_time(self.output)
return self._get_output_bytes()
def uuid1(self):
return uuid.uuid1().bytes
def uuid1(self):
"""Generates a uuid1 - a device specific uuid
Returns:
bytes: 16-byte uuid
"""
return self._uuid_in_use.uuid1()
def open_url(url, handler):
global _handler
global _requestID
_requestID = uuid.uuid1()
_handler = handler
url_with_uuid = url + 'xcallbackresponse-' + str(_requestID)
webbrowser.open(url_with_uuid)
def test_times_from_uuid1(self):
node = uuid.getnode()
now = time.time()
u = uuid.uuid1(node, 0)
t = util.unix_time_from_uuid1(u)
self.assertAlmostEqual(now, t, 2)
dt = util.datetime_from_uuid1(u)
t = calendar.timegm(dt.timetuple()) + dt.microsecond / 1e6
self.assertAlmostEqual(now, t, 2)
def test_model_over_write(self):
"""
Test to ensure overwriting of primary keys in model inheritance is allowed
This is currently only an issue in PyPy. When PYTHON-504 is introduced this should
be updated error out and warn the user
@since 3.6.0
@jira_ticket PYTHON-576
@expected_result primary keys can be overwritten via inheritance
@test_category object_mapper
"""
class TimeModelBase(Model):
uuid = columns.TimeUUID(primary_key=True)
class DerivedTimeModel(TimeModelBase):
__table_name__ = 'derived_time'
uuid = columns.TimeUUID(primary_key=True, partition_key=True)
value = columns.Text(required=False)
# In case the table already exists in keyspace
drop_table(DerivedTimeModel)
sync_table(DerivedTimeModel)
uuid_value = uuid1()
uuid_value2 = uuid1()
DerivedTimeModel.create(uuid=uuid_value, value="first")
DerivedTimeModel.create(uuid=uuid_value2, value="second")
DerivedTimeModel.objects.filter(uuid=uuid_value)
def test_composite(self):
# double bool
insert, select = self.create_prepare(('double', 'boolean'))
self.insert_select_token(insert, select, (3.1459, True))
self.insert_select_token(insert, select, (1.21e9, False))
# uuid string int
insert, select = self.create_prepare(('timeuuid', 'varchar', 'int'))
self.insert_select_token(insert, select, (uuid1(), 'asdf', 400))
self.insert_select_token(insert, select, (uuid1(), 'fdsa', -1))
def set_uuid(nodes, **kwargs):
"""
Given a list of ``nodes`` and a set of keyword arguments ``kwargs``,
set UUID(s) on ``nodes``.
:param nodes: Nodes
:type nodes: list
:param \**kwargs: UUID dictionary
Usage:
>>> import nukeuuid
>>> kw = {'': 'fca7201e-b53d-4918-9ab0-bb4ec5590f3c',
'utility': '5f2d525d-3e00-4bc5-88c4-794ad87f5699'}
>>> nukeuuid.set_uuid(nuke.selectedNodes(), **kw)
"""
if not isinstance(nodes, list):
nodes = [nodes]
for node in nodes:
for type_, uuid_ in kwargs.iteritems():
if not type_ and not uuid_:
uuid_ = str(uuid.uuid1())
attr = _convert_type(type_)
_make_attr(node, attr)
node[attr].setValue(uuid_)
node[attr].setEnabled(False)
test_todo_model_integration.py 文件源码
项目:serverless-pynamodb-rest-example
作者: helveticafire
项目源码
文件源码
阅读 25
收藏 0
点赞 0
评论 0
def setUp(self, load_dbs=None):
TodoModel.setup_model(TodoModel, 'region', 'todo' + str(uuid.uuid1()), 'ENVIRONMENT' not in os.environ)
if not TodoModel.exists():
TodoModel.create_table(wait=True)
if load_dbs:
for db_file in load_dbs:
TodoModel.load(db_file)
super().setUp()
test_todo_model_integration.py 文件源码
项目:serverless-pynamodb-rest-example
作者: helveticafire
项目源码
文件源码
阅读 24
收藏 0
点赞 0
评论 0
def test_create(self):
todo_text = 'text'
todo_id = str(uuid.uuid1())
t1 = TodoModel(todo_id=todo_id,
text=todo_text,
created_at=datetime.now())
t1.save()
self.assertEquals(t1.text, todo_text)
self.assertEquals(t1.checked, False)
# get and check
t2 = TodoModel.get(todo_id)
self.assertEquals(t2.text, todo_text)
def handle(event, context):
try:
table_name = os.environ[ENV_VAR_DYNAMODB_TABLE]
region = os.environ[ENV_VAR_DYNAMODB_REGION]
except KeyError as err:
error_message = '{0} is missing from environment variables'.format(str(err))
return HttpResponseServerError(error_code='ENV_VAR_NOT_SET',
error_message=error_message).__dict__()
TodoModel.setup_model(TodoModel, region, table_name, ENV_VAR_ENVIRONMENT not in os.environ)
try:
data = json.loads(event['body'])
except ValueError as err:
return HttpResponseBadRequest(error_code='JSON_IRREGULAR',
error_message=str(err)).__dict__()
if 'text' not in data:
logging.error('Validation Failed')
return HttpResponseBadRequest(error_code='BODY_PROPERTY_MISSING',
error_message='Could not create the todo item.').__dict__()
if not data['text']:
logging.error('Validation Failed - text was empty. %s', data)
return HttpResponseBadRequest(error_code='VALIDATION_FAILED',
error_message='Could not create the todo item. As text was empty.').__dict__()
a_todo = TodoModel(todo_id=str(uuid.uuid1()),
text=data['text'],
checked=False,
created_at=datetime.now())
# write the todo to the database
a_todo.save()
# create a response
return HttpCreatedJSONResponse(body=dict(a_todo)).__dict__()
def set(self, data, **kwargs):
key = uuid.uuid1().hex
self.__mem__[key] = None
return self.__do_set__(key, data, **kwargs)
def new_browser_session_test(request):
if 'UID' in request.session:
uid = request.session['UID']
message = "Welcome back"
else:
uid = uuid.uuid1()
request.session['UID'] = str(uid)
message = "Hello new user"
return render(request, "django_functest/tests/new_browser_session_test.html",
{'uid': uid,
'message': message,
})
def __init__(self, api_version=1, request_id=None):
self.api_version = api_version
if request_id:
self.request_id = request_id
else:
self.request_id = str(uuid.uuid1())
self.ops = []
def __init__(self, api_version=1, request_id=None):
self.api_version = api_version
if request_id:
self.request_id = request_id
else:
self.request_id = str(uuid.uuid1())
self.ops = []