Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
existing_entity_id = self._findEntityIdByNameAndParent(properties['name'], properties.get('parentId', ROOT_ENTITY))
if existing_entity_id is None: raise
# get existing properties and annotations
if not bundle:
bundle = self._getEntityBundle(existing_entity_id, bitFlags=0x1|0x2)
# Need some fields from the existing entity: id, etag, and version info.
existing_entity = bundle['entity']
# Update the conflicting Entity
existing_entity.update(properties)
properties = self._updateEntity(existing_entity, forceVersion, versionLabel)
# Merge new annotations with existing annotations
existing_annos = from_synapse_annotations(bundle['annotations'])
existing_annos.update(annotations)
annotations = existing_annos
else:
raise
# Deal with access restrictions
if isRestricted:
self._createAccessRequirementIfNone(properties)
# Update annotations
annotations['etag'] = properties['etag']
annotations = self.setAnnotations(properties, annotations)
properties['etag'] = annotations['etag']
# If the parameters 'used' or 'executed' are given, create an Activity object
activity = kwargs.get('activity', None)
See :py:func:`synapseclient.Synapse.get`.
See :py:func:`synapseclient.Synapse._getEntityBundle`.
See :py:mod:`synapseclient.Entity`.
"""
# Note: This version overrides the version of 'entity' (if the object is Mappable)
version = kwargs.get('version', None)
downloadFile = kwargs.get('downloadFile', True)
downloadLocation = kwargs.get('downloadLocation', None)
ifcollision = kwargs.get('ifcollision', 'keep.both')
submission = kwargs.get('submission', None)
# Make a fresh copy of the Entity
local_state = entity.local_state() if entity and isinstance(entity, Entity) else None
properties = entityBundle['entity']
annotations = from_synapse_annotations(entityBundle['annotations'])
entity = Entity.create(properties, annotations, local_state)
# Handle both FileEntities and Locationables
isLocationable = is_locationable(entity)
if isinstance(entity, File) or isLocationable:
fileName = entity['name']
if not isLocationable:
# Fill in information about the file, even if we don't download it
# Note: fileHandles will be an empty list if there are unmet access requirements
for handle in entityBundle['fileHandles']:
if handle['id'] == entityBundle['entity']['dataFileHandleId']:
entity.md5 = handle.get('contentMd5', '')
entity.fileSize = handle.get('contentSize', None)
entity.contentType = handle.get('contentType', None)
fileName = handle['fileName']
:param entity: An Entity or Synapse ID to update annotations of
:param annotations: A dictionary in Synapse format or a Python format
:param kwargs: Any additional entries to be added to the annotations dictionary
:returns: A dictionary
"""
uri = '/entity/%s/annotations' % id_of(entity)
annotations.update(kwargs)
synapseAnnos = to_synapse_annotations(annotations)
synapseAnnos['id'] = id_of(entity)
if 'etag' in entity and 'etag' not in synapseAnnos:
synapseAnnos['etag'] = entity['etag']
return from_synapse_annotations(self.restPUT(uri, body=json.dumps(synapseAnnos)))
Retrieve annotations for an Entity from the Synapse Repository.
:param entity: An Entity or Synapse ID to lookup
:param version: The version of the Entity to retrieve.
:returns: A dictionary
"""
# Note: Specifying the version results in a zero-ed out etag,
# even if the version is the most recent.
# See `PLFM-1874 `_ for more details.
if version:
uri = '/entity/%s/version/%s/annotations' % (id_of(entity), str(version))
else:
uri = '/entity/%s/annotations' % id_of(entity)
return from_synapse_annotations(self.restGET(uri))