Stack Trace
self = <test.integration.gatewaytest.test_get_objects.TestGetObject object at 0x7f486d37e850>
def testGetObjectsAnnotation(self):
# create new group and user - don't rely on existing test user/group
group = self.new_group(perms='rwra--')
client, user = self.new_client_and_user(group=group)
conn = BlitzGateway(client_obj=client)
update = conn.getUpdateService()
ann_types = ["CommentAnnotation", "TagAnnotation",
"LongAnnotation", "XmlAnnotation",
"DoubleAnnotation", "BooleanAnnotation",
"TimestampAnnotation", "TermAnnotation"]
ns = "test_get_objects_namespace"
ns_map = "test_get_objects_map_namespace"
ns_file = "test_get_objects_annotation_file"
def create_dataset_with_annotations(name, dtype="Dataset"):
if dtype == "Dataset":
obj = DatasetI()
obj.name = rstring(name)
obj = update.saveAndReturnObject(obj)
wrapper = omero.gateway.DatasetWrapper(conn, obj)
elif dtype == "Project":
obj = ProjectI()
obj.name = rstring(name)
obj = update.saveAndReturnObject(obj)
wrapper = omero.gateway.ProjectWrapper(conn, obj)
# Create total of 10 annotations on the object...
# create Comment, Tag, Xml, etc
for ann_type in ann_types:
ann = getattr(omero.gateway, ann_type + "Wrapper")(conn)
ann.setNs(ns)
if ann_type in ("LongAnnotation", "DoubleAnnotation",
"TimestampAnnotation"):
ann.setValue(100.0)
elif ann_type == "BooleanAnnotation":
ann.setValue(True)
else:
ann.setValue("Test %s: %s" % (ann_type, name))
wrapper.linkAnnotation(ann)
# create MapAnnotation
mapAnn = omero.gateway.MapAnnotationWrapper(conn)
mapAnn.setNs(ns)
mapAnn.setValue([("key1", "value1"), ("key2", "value2")])
mapAnn.setNs(ns_map)
mapAnn.save()
wrapper.linkAnnotation(mapAnn)
# create FileAnnotation
fileAnn = omero.gateway.FileAnnotationWrapper(conn)
fileObj = omero.model.OriginalFileI()
fileObj = omero.gateway.OriginalFileWrapper(conn, fileObj)
fileObj.setName(omero.rtypes.rstring('fileName'))
fileObj.setPath(omero.rtypes.rstring('path/to/file'))
fileObj.setHash(omero.rtypes.rstring('a'))
fileObj.setSize(omero.rtypes.rlong(0))
fileObj.save()
fileAnn.setFile(fileObj)
fileAnn.setNs(ns_file)
fileAnn.save()
wrapper.linkAnnotation(fileAnn)
return wrapper
dataset1 = create_dataset_with_annotations("Dataset 1")
dataset2 = create_dataset_with_annotations("Dataset 2")
create_dataset_with_annotations("Project 1", dtype="Project")
# Add Tag to Group
groupId = conn.getEventContext().groupId
tag = omero.gateway.TagAnnotationWrapper(conn)
tag.setNs("test_get_objects_group_tag")
tag.setValue("Test Tag on Group")
tag.save()
link = omero.model.ExperimenterGroupAnnotationLinkI()
link.child = omero.model.TagAnnotationI(tag.id, False)
link.parent = omero.model.ExperimenterGroupI(groupId, False)
conn.getUpdateService().saveAndReturnObject(link)
# get all annotations on one Dataset
annGen = conn.getObjects("Annotation",
opts={'parent_type': "Dataset",
'parent_ids': [dataset1.id]})
> assert len(list(annGen)) == 10
^^^^^^^^^^^^
test/integration/gatewaytest/test_get_objects.py:738:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../../../.venv3/lib64/python3.11/site-packages/omero/gateway/__init__.py:3362: in getObjects
query, params, wrapper = self.buildQuery(
../../../../.venv3/lib64/python3.11/site-packages/omero/gateway/__init__.py:3424: in buildQuery
query, clauses, baseParams = cls._getQueryString(opts)
^^^^^^^^^^^^^^^^^^^^^^^^^
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
cls = <class 'omero.gateway.AnnotationWrapper'>
opts = {'parent_type': 'Dataset', 'parent_ids': [572]}
@classmethod
def _getQueryString(cls, opts=None):
"""
Used for building queries in generic methods such as
getObjects("Annotation")
Returns a tuple of (query, clauses, params).
:param opts: Dictionary of optional parameters.
parent_type: (optional) "Project", "Dataset", "Image" etc
parent_ids: (optional) list of IDs for the parent type
ns: (optional) namespace string to filter by
:return: Tuple of string, list, ParametersI
"""
query, clauses, params = super(
AnnotationWrapper, cls)._getQueryString(opts)
if opts is None:
opts = {}
fetch_file = ""
if cls.OMERO_CLASS in ("Annotation", "FileAnnotation"):
fetch_file = "left outer join fetch obj.file as file"
query = (f"select obj from {cls.OMERO_CLASS} obj "
f"{fetch_file} "
"join fetch obj.details.owner as owner "
"join fetch obj.details.creationEvent")
# We want to make parent_type case-insensitive...
if 'parent_type' in opts:
> ann_link_name = ann_link_name(opts['parent_type'])
^^^^^^^^^^^^^
E UnboundLocalError: cannot access local variable 'ann_link_name' where it is not associated with a value
../../../../.venv3/lib64/python3.11/site-packages/omero/gateway/__init__.py:5213: UnboundLocalError
