def process_item(self, item, spider):
if 'id' not in item:
if 'link' in item:
item['id'] = uuid.uuid5(uuid.NAMESPACE_DNS, item['link']).urn
else:
raise DropItem('A link is required to autogenerate the feed '
'id for: {}'.format(item))
if 'title' not in item:
# Having a title is mandatory, so we use an empty string if none
# is set.
item['title'] = ''
return item
python类uuid5()的实例源码
def to_uuid(string):
"""
Mirroring the jinja filter implemented in ansible
Input a string. Returns the uuid ansible would generate as a string.
"""
return str(uuid.uuid5(uuid.UUID('361E6D51-FAEC-444A-9079-341386DA8E2E'), string.encode('utf-8')))
def uuid(self):
"""Generate UUID for the user basee on the the primary email."""
return uuid.uuid5(uuid.NAMESPACE_URL, "mailto:" + (self.email or self.eppn))
def setUp(self):
"""Setup test."""
from uuid import uuid5, NAMESPACE_DNS
class UUIDModel(db.Document):
uuid = db.UUIDField()
self.model_cls = UUIDModel
uuid = uuid5(NAMESPACE_DNS, "This is a test")
self.expected_data = {
"uuid": uuid
}
self.data = json.dumps({"uuid": str(uuid)})
self.hook = generate_object_hook(UUIDModel)
def setUp(self):
"""Setup class."""
from uuid import uuid5, NAMESPACE_DNS
self.encoder = GoodJSONEncoder()
self.data = uuid5(NAMESPACE_DNS, "This is a test")
self.expected = str(self.data)
def __init__(self, name, uuid=None):
if uuid is None:
self.uuid = uuid5(NAMESPACE_DNS, name).hex
else:
self.uuid = uuid
self.name = name
def get_deterministic_id(external_host, sid):
"""
Get deterministic FTS job id.
:param external_host: FTS server as a string.
:param sid: FTS seed id.
:returns: FTS transfer identifier.
"""
baseid, voname = get_transfer_baseid_voname(external_host)
if baseid is None or voname is None:
return None
root = uuid.UUID(baseid)
atlas = uuid.uuid5(root, voname)
jobid = uuid.uuid5(atlas, sid)
return str(jobid)
def create_uuid(self):
if not self.version or self.version == 4:
return uuid.uuid4()
elif self.version == 1:
return uuid.uuid1(self.node, self.clock_seq)
elif self.version == 2:
raise UUIDVersionError("UUID version 2 is not supported.")
elif self.version == 3:
return uuid.uuid3(self.namespace, self.name)
elif self.version == 5:
return uuid.uuid5(self.namespace, self.name)
else:
raise UUIDVersionError("UUID version %s is not valid." % self.version)
def test_create_timeuuid_with_uuid4_string_should_fail(self):
''' creating a TimeUUID with a hex uuid4 should fail'''
for i in range(1,100):
u = uuid.uuid4()
with self.assertRaises(ValueError) as cm:
t = timeuuid.TimeUUID(s=u.hex)
self.assertEqual(str(cm.exception), 'Invalid UUID type')
for fn in [uuid.uuid3, uuid.uuid5]:
for i in range(1,100):
u = fn(uuid.NAMESPACE_DNS,str(os.urandom(10)))
with self.assertRaises(ValueError) as cm:
t = timeuuid.TimeUUID(s=u.hex)
self.assertEqual(str(cm.exception), 'Invalid UUID type')
def generate_container_id(container_name):
# container id is a uuid in the namespace of the machine
if not config["container_mode"]:
return str(uuid.uuid5(uuid.UUID(generate_machine_id()), container_name.encode('utf8')))
else:
return container_name.encode('utf8')
def to_uuid(string):
return str(uuid.uuid5(UUID_NAMESPACE_ANSIBLE, str(string)))
def random_id(stream_arn, kinesis_shard_id):
namespace = uuid.UUID(bytes=hashlib.sha1(stream_arn.encode('utf-8')).digest()[:16])
return uuid.uuid5(namespace, kinesis_shard_id.encode('utf-8')).hex
def build_id(*names):
x = NAMESPACE
for name in names:
x = uuid.uuid5(x, str(name))
return str(x)
def set_ready(cls, environment, version, resource_uuid, resource_id, status):
"""
Mark a resource as deployed in the configuration model status
"""
entry_uuid = uuid.uuid5(resource_uuid, resource_id)
resource_key = "status.%s" % entry_uuid
yield cls._coll.update_one({"environment": environment, "version": version},
{"$set": {resource_key: {"status": cls._value_to_dict(status), "id": resource_id}}})
def update_resource(cls, dryrun_id, resource_id, dryrun_data):
"""
Register a resource update with a specific query that sets the dryrun_data and decrements the todo counter, only
if the resource has not been saved yet.
"""
entry_uuid = uuid.uuid5(dryrun_id, resource_id)
resource_key = "resources.%s" % entry_uuid
query = {"_id": dryrun_id, resource_key: {"$exists": False}}
update = {"$inc": {"todo": int(-1)}, "$set": {resource_key: cls._value_to_dict(dryrun_data)}}
yield cls._coll.update_one(query, update)
def make_token(user):
date = localized_date()
# Automatically expires different after an hour
return str(uuid5(NAMESPACE_X500, "-".join([str(user.id), user.username,
date.strftime('%Y%m%d%h')])))
def _id(self):
_id = str(uuid.uuid5(NAMESPACE_UUID, self.__hash__()))
return _id
def save(self, *args, **kwargs):
"""save"""
if self.status and self.status.is_final:
self.done = True
elif self.type and self.type.allowed_status.filter(is_final=True).exists():
self.done = False
if not self.done_date and self.done:
self.done_date = now_rounded()
elif self.done_date and not self.done:
self.done_date = None
#generate number automatically based on action type
if self.number == 0 and self.type and self.type.number_auto_generated:
self.number = self.type.last_number = self.type.last_number + 1
self.type.save()
ret = super(Action, self).save(*args, **kwargs)
if self.type:
if self.type.generate_uuid and not self.uuid:
name = u'{0}-action-{1}-{2}'.format(
project_settings.SECRET_KEY, self.id, self.type.id if self.type else 0
)
name = unicodedata.normalize('NFKD', unicode(name)).encode("ascii", 'ignore')
self.uuid = unicode(uuid.uuid5(uuid.NAMESPACE_URL, name))
super(Action, self).save()
if not self.type.generate_uuid and self.uuid:
self.uuid = ''
super(Action, self).save()
return ret
def save(self, *args, **kwargs):
"""save"""
super(MagicLink, self).save(*args, **kwargs)
if not self.uuid:
name = u'{0}-magic-link-{1}-{2}'.format(settings.SECRET_KEY, self.id, self.url)
name = unicodedata.normalize('NFKD', unicode(name)).encode("ascii", 'ignore')
self.uuid = uuid.uuid5(uuid.NAMESPACE_URL, name)
return super(MagicLink, self).save()
def generate_analysis_target_id(analysis_target, name):
# this function generates 'machine-id's for analysis target's that
# might not be hosts.
#
# 'machine_id' is what Insights uses to uniquely identify
# the thing-to-be-analysed. Primarily it determines when two uploads
# are for the 'same thing', and so the latest upload should update the
# later one Up till now that has only been hosts (machines), and so a
# random uuid (uuid4) machine-id was generated for the host as its machine-id,
# and written to a file on the host, and reused for all insights
# uploads for that host.
#
# For docker images and containers, it will be difficult to impossible
# to save their machine id's anywhere. Also, while containers change
# over time just like hosts, images don't change over time, though they
# can be rebuilt. So for images we want the 'machine-id' for an 'image'
# to follow the rebuilt image, not change every time the image is rebuilt.
# Typically when an image is rebuilt, the rebuilt image will have the same
# name as its predicessor, but a different version (tag).
#
# So for images and containers, instead of random uuids, we use namespace uuids
# (uuid5's). This generates a new uuid based on a combination of another
# uuid, and a name (a character string). This will always generate the
# same uuid for the same given base uuid and name. This saves us from
# having to save the image's uuid anywhere, and lets us control the created uuid
# by controlling the name used to generate it. Keep the name and base uuid) the
# same, we get the same uuid.
#
# For the base uuid we use the uuid of the host we are running on.
# For containers this is the obvious choice, for images it is less obviously
# what base uuid is correct. For now we will just go with the host's uuid also.
#
# For the name, we leave that outside this function, but in general it should
# be the name of the container or the name of the image, and if you want to
# replace the results on the insights server, you have to use the same name
if analysis_target == "host":
return generate_machine_id()
elif (analysis_target == "docker_image" or
analysis_target == "docker_container" or
analysis_target == "compressed_file" or
analysis_target == "mountpoint"):
return generate_container_id(name)
else:
raise ValueError("Unknown analysis target: %s" % analysis_target)
