senaite.core.schema.uidreferencefield.UIDReferenceField.get_uid() - Code Metrics - Inspection of "Added behavior to allow sharing objects across cli..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — 2.x ( 5710a9...27b54e )

by Ramon

created 2022-06-05 19:35 UTC

UIDReferenceField.get_uid() A

↳ Parent: senaite.core.schema.uidreferencefield

Complexity

Conditions

Size

Total Lines	10
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	5
dl	0
loc	10
rs	10
c	0
b	0
f	0
cc	2
nop	2

# -*- coding: utf-8 -*-

import six

from Acquisition import aq_base
from bika.lims import api
from persistent.dict import PersistentDict
from persistent.list import PersistentList
from plone.behavior.interfaces import IBehavior
from plone.uuid.interfaces import ATTRIBUTE_NAME
from senaite.core import logger
from senaite.core.interfaces import IHaveUIDReferences
from senaite.core.schema.fields import BaseField
from senaite.core.schema.interfaces import IUIDReferenceField
from zope.annotation.interfaces import IAnnotations
from zope.component import adapter
from zope.interface import alsoProvides
from zope.interface import implementer
from zope.lifecycleevent.interfaces import IObjectCreatedEvent
from zope.schema import ASCIILine
from zope.schema import List

BACKREFS_STORAGE = "senaite.core.schema.uidreferencefield.backreferences"


@adapter(IHaveUIDReferences, IObjectCreatedEvent)
def on_object_created(object, event):
    """Link backreferences after the object was created

    This is necessary, because objects during initialization have no UID set!
    https://community.plone.org/t/accessing-uid-in-dexterity-field-setter/14468

    Therefore, and only for the creation case, we need to link the back
    references in this handler for all UID reference fields.
    """
    fields = api.get_fields(object)
    for name, field in fields.items():
        if not IUIDReferenceField.providedBy(field):
            continue
        # after creation, we only need to link backreferences
        value = field.get(object)
        # handle single valued reference fields
        if api.is_object(value):
            field.link_backref(value, object)
            logger.info(
                "Adding back reference from %s -> %s" % (
                    value, object))
        # handle multi valued reference fields
        elif isinstance(value, list):
            for ref in value:
                logger.info(
                    "Adding back reference from %s -> %s" % (
                        ref, object))
                field.link_backref(ref, object)


def get_backrefs(context, relationship, as_objects=False):
    """Return backreferences of the context

    :returns: List of UIDs that are linked by the relationship
    """
    context = aq_base(context)
    # get the backref annotation storage of the context
    backrefs = get_backref_storage(context)
    # get the referenced UIDs
    backref_uids = list(backrefs.get(relationship, []))

    if not backref_uids:
        return []

    if as_objects is True:
        return [api.get_object(uid) for uid in backref_uids]

    return backref_uids


def get_backref_storage(context):
    """Get the annotation storage for backreferences of the context
    """
    annotation = IAnnotations(context)
    if annotation.get(BACKREFS_STORAGE) is None:

        annotation[BACKREFS_STORAGE] = PersistentDict()
    return annotation[BACKREFS_STORAGE]


@implementer(IUIDReferenceField)
class UIDReferenceField(List, BaseField):
    """Stores UID references to other objects
    """

    value_type = ASCIILine(title=u"UID")

    def __init__(self, allowed_types=None, multi_valued=True, **kw):
        if allowed_types is None:
            allowed_types = ()
        self.allowed_types = allowed_types
        self.multi_valued = multi_valued
        super(UIDReferenceField, self).__init__(**kw)

    def is_initializing(self, object):
        """Checks if the object is initialized at the moment

        Background:

        Objects being created do not have a UID set.
        Therefore, ths backreferences need to be postponed to an event handler.

        https://community.plone.org/t/accessing-uid-in-dexterity-field-setter/14468
        """
        uid = getattr(object, ATTRIBUTE_NAME, None)
        if uid is None:
            return True
        return False

    def to_list(self, value, filter_empty=True):
        """Ensure the value is a list
        """
        if isinstance(value, six.string_types):
            value = [value]
        elif api.is_object(value):
            value = [value]
        elif value is None:
            value = []
        if filter_empty:
            value = filter(None, value)
        return value

    def get_relationship_key(self, context):
        """Relationship key used for backreferences

        The key used for the annotation storage on the referenced object to
        remember the current object UID.

        :returns: storage key to lookup back references
        """
        portal_type = api.get_portal_type(context)
        return "%s.%s" % (portal_type, self.__name__)

    def get_uid(self, value):
        """Value -> UID

        :parm value: object/UID/SuperModel
        :returns: UID
        """
        try:
            return api.get_uid(value)
        except api.APIError:
            return None

    def get_object(self, value):
        """Value -> object

        :returns: Object or None
        """
        try:
            return api.get_object(value)
        except api.APIError:
            return None

    def get_allowed_types(self):
        """Returns the allowed reference types

        :returns: tuple of allowed_types
        """
        allowed_types = self.allowed_types
        if not allowed_types:
            allowed_types = ()
        elif isinstance(allowed_types, six.string_types):
            allowed_types = (allowed_types, )
        return allowed_types

    def set(self, object, value):
        """Set UID reference

        :param object: the instance of the field
        :param value: object/UID/SuperModel
        :type value: list/tuple/str
        """

        # Object might be a behavior instead of the object itself
        object = self._get_content_object(object)

        # always mark the object if references are set
        # NOTE: there might be multiple UID reference field set on this object!
        if value:
            alsoProvides(object, IHaveUIDReferences)

        # always handle all values internally as a list
        value = self.to_list(value)

        # convert to UIDs
        uids = []
        for v in value:
            uid = self.get_uid(v)
            if uid is None:
                continue
            uids.append(uid)

        # current set UIDs
        existing = self.to_list(self.get_raw(object))

        # filter out new/removed UIDs
        added_uids = [u for u in uids if u not in existing]
        added_objs = filter(None, map(self.get_object, added_uids))

        removed_uids = [u for u in existing if u not in uids]
        removed_objs = filter(None, map(self.get_object, removed_uids))

        # link backreferences of new uids
        for added_obj in added_objs:
            self.link_backref(added_obj, object)

        # unlink backreferences of removed UIDs
        for removed_obj in removed_objs:
            self.unlink_backref(removed_obj, object)

        super(UIDReferenceField, self).set(object, uids)

    def unlink_backref(self, source, target):
        """Remove backreference from the source to the target

        :param source: the object where the backref is stored (our reference)
        :param target: the object where the backref points to (our object)
        :returns: True when the backref was removed, False otherwise
        """

        # Target might be a behavior instead of the object itself
        target = self._get_content_object(target)

        # This should be actually not possible
        if self.is_initializing(target):
            raise ValueError("Objects in initialization state "
                             "can not have existing back references!")

        target_uid = self.get_uid(target)
        # get the storage key
        key = self.get_relationship_key(target)
        # get all backreferences from the source
        backrefs = get_backref_storage(source)
        if key not in backrefs:
            logger.warn(
                "Referenced object {} has no backreferences for the key {}"
                .format(repr(source), key))
            return False
        if target_uid not in backrefs[key]:
            logger.warn("Target {} was not linked by {}"
                        .format(repr(target), repr(source)))
            return False
        backrefs[key].remove(target_uid)
        return True

    def link_backref(self, source, target):
        """Add backreference from the source to the target

        :param source: the object where the backref is stored (our reference)
        :param target: the object where the backref points to (our object)
        :returns: True when the backref was written
        """

        # Target might be a behavior instead of the object itself
        target = self._get_content_object(target)

        # Object is initializing and don't have an UID!
        # -> Postpone to set back references in event handler
        if self.is_initializing(target):
            logger.info("Object is in initialization state. "
                        "Back references will be set in event handler")
            return

        target_uid = api.get_uid(target)
        # get the annotation storage key
        key = self.get_relationship_key(target)
        # get all backreferences
        backrefs = get_backref_storage(source)
        if key not in backrefs:
            backrefs[key] = PersistentList()
        if target_uid not in backrefs[key]:
            backrefs[key].append(target_uid)
        return True

    def get(self, object):
        """Get referenced objects

        :param object: instance of the field
        :returns: list of referenced objects
        """
        return self._get(object, as_objects=True)

    def get_raw(self, object):
        """Get referenced UIDs

        NOTE: Called from the data manager `query` method
              to get the widget value

        :param object: instance of the field
        :returns: list of referenced UIDs
        """
        return self._get(object, as_objects=False)

    def _get(self, object, as_objects=False):
        """Returns single/multi value

        :param object: instance of the field
        :param as_objects: Flag for UID/object returns
        :returns: list of referenced UIDs
        """

        # Object might be a behavior instead of the object itself
        object = self._get_content_object(object)

        uids = super(UIDReferenceField, self).get(object)

        if not uids:
            uids = []
        elif not isinstance(uids, (list, tuple)):
            uids = [uids]

        if as_objects is True:
            uids = filter(None, map(self.get_object, uids))

        if self.multi_valued:
            return uids
        if len(uids) == 0:
            return None
        return uids[0]

    def _validate(self, value):
        """Validator when called from form submission
        """
        super(UIDReferenceField, self)._validate(value)
        # check if the fields accepts single values only
        if not self.multi_valued and len(value) > 1:
            raise ValueError("Single valued field accepts at most 1 value")

        # check for valid UIDs
        for uid in value:
            if not api.is_uid(uid):
                raise ValueError("Invalid UID: '%s'" % uid)

        # check if the type is allowed
        allowed_types = self.get_allowed_types()
        if allowed_types:
            objs = filter(None, map(self.get_object, value))
            types = set(map(api.get_portal_type, objs))
            if not types.issubset(allowed_types):
                raise ValueError("Only the following types are allowed: %s"
                                 % ",".join(allowed_types))

    def _get_content_object(self, thing):
        """Returns the underlying content object
        """
        if IBehavior.providedBy(thing):
            return self._get_content_object(thing.context)
        if api.is_dexterity_content(thing):
            return thing
        raise ValueError("Not a valid object: %s" % repr(thing))


1			# -- coding: utf-8 --
2
3			import six
4
5			from Acquisition import aq_base
6			from bika.lims import api
7			from persistent.dict import PersistentDict
8			from persistent.list import PersistentList
9			from plone.behavior.interfaces import IBehavior
10			from plone.uuid.interfaces import ATTRIBUTE_NAME
11			from senaite.core import logger
12			from senaite.core.interfaces import IHaveUIDReferences
13			from senaite.core.schema.fields import BaseField
14			from senaite.core.schema.interfaces import IUIDReferenceField
15			from zope.annotation.interfaces import IAnnotations
16			from zope.component import adapter
17			from zope.interface import alsoProvides
18			from zope.interface import implementer
19			from zope.lifecycleevent.interfaces import IObjectCreatedEvent
20			from zope.schema import ASCIILine
21			from zope.schema import List
22
23			BACKREFS_STORAGE = "senaite.core.schema.uidreferencefield.backreferences"
24
25
26			@adapter(IHaveUIDReferences, IObjectCreatedEvent)
27			def on_object_created(object, event):
28			"""Link backreferences after the object was created
29
30			This is necessary, because objects during initialization have no UID set!
31			https://community.plone.org/t/accessing-uid-in-dexterity-field-setter/14468
32
33			Therefore, and only for the creation case, we need to link the back
34			references in this handler for all UID reference fields.
35			"""
36			fields = api.get_fields(object)
37			for name, field in fields.items():
38			if not IUIDReferenceField.providedBy(field):
39			continue
40			# after creation, we only need to link backreferences
41			value = field.get(object)
42			# handle single valued reference fields
43			if api.is_object(value):
44			field.link_backref(value, object)
45			logger.info(
46			"Adding back reference from %s -> %s" % (
47			value, object))
48			# handle multi valued reference fields
49			elif isinstance(value, list):
50			for ref in value:
51			logger.info(
52			"Adding back reference from %s -> %s" % (
53			ref, object))
54			field.link_backref(ref, object)
55
56
57			def get_backrefs(context, relationship, as_objects=False):
58			"""Return backreferences of the context
59
60			:returns: List of UIDs that are linked by the relationship
61			"""
62			context = aq_base(context)
63			# get the backref annotation storage of the context
64			backrefs = get_backref_storage(context)
65			# get the referenced UIDs
66			backref_uids = list(backrefs.get(relationship, []))
67
68			if not backref_uids:
69			return []
70
71			if as_objects is True:
72			return [api.get_object(uid) for uid in backref_uids]
73
74			return backref_uids
75
76
77			def get_backref_storage(context):
78			"""Get the annotation storage for backreferences of the context
79			"""
80			annotation = IAnnotations(context)
81			if annotation.get(BACKREFS_STORAGE) is None:
			0 ignored issues – show Comprehensibility Best Practice introduced 2021-10-21 15:27 UTC by Report Bug Copy Issue Report The variable `BACKREFS_STORAGE` does not seem to be defined. Loading history...
82			annotation[BACKREFS_STORAGE] = PersistentDict()
83			return annotation[BACKREFS_STORAGE]
84
85
86			@implementer(IUIDReferenceField)
87			class UIDReferenceField(List, BaseField):
88			"""Stores UID references to other objects
89			"""
90
91			value_type = ASCIILine(title=u"UID")
92
93			def __init__(self, allowed_types=None, multi_valued=True, **kw):
94			if allowed_types is None:
95			allowed_types = ()
96			self.allowed_types = allowed_types
97			self.multi_valued = multi_valued
98			super(UIDReferenceField, self).__init__(**kw)
99
100			def is_initializing(self, object):
101			"""Checks if the object is initialized at the moment
102
103			Background:
104
105			Objects being created do not have a UID set.
106			Therefore, ths backreferences need to be postponed to an event handler.
107
108			https://community.plone.org/t/accessing-uid-in-dexterity-field-setter/14468
109			"""
110			uid = getattr(object, ATTRIBUTE_NAME, None)
111			if uid is None:
112			return True
113			return False
114
115			def to_list(self, value, filter_empty=True):
116			"""Ensure the value is a list
117			"""
118			if isinstance(value, six.string_types):
119			value = [value]
120			elif api.is_object(value):
121			value = [value]
122			elif value is None:
123			value = []
124			if filter_empty:
125			value = filter(None, value)
126			return value
127
128			def get_relationship_key(self, context):
129			"""Relationship key used for backreferences
130
131			The key used for the annotation storage on the referenced object to
132			remember the current object UID.
133
134			:returns: storage key to lookup back references
135			"""
136			portal_type = api.get_portal_type(context)
137			return "%s.%s" % (portal_type, self.__name__)
138
139			def get_uid(self, value):
140			"""Value -> UID
141
142			:parm value: object/UID/SuperModel
143			:returns: UID
144			"""
145			try:
146			return api.get_uid(value)
147			except api.APIError:
148			return None
149
150			def get_object(self, value):
151			"""Value -> object
152
153			:returns: Object or None
154			"""
155			try:
156			return api.get_object(value)
157			except api.APIError:
158			return None
159
160			def get_allowed_types(self):
161			"""Returns the allowed reference types
162
163			:returns: tuple of allowed_types
164			"""
165			allowed_types = self.allowed_types
166			if not allowed_types:
167			allowed_types = ()
168			elif isinstance(allowed_types, six.string_types):
169			allowed_types = (allowed_types, )
170			return allowed_types
171
172			def set(self, object, value):
173			"""Set UID reference
174
175			:param object: the instance of the field
176			:param value: object/UID/SuperModel
177			:type value: list/tuple/str
178			"""
179
180			# Object might be a behavior instead of the object itself
181			object = self._get_content_object(object)
182
183			# always mark the object if references are set
184			# NOTE: there might be multiple UID reference field set on this object!
185			if value:
186			alsoProvides(object, IHaveUIDReferences)
187
188			# always handle all values internally as a list
189			value = self.to_list(value)
190
191			# convert to UIDs
192			uids = []
193			for v in value:
194			uid = self.get_uid(v)
195			if uid is None:
196			continue
197			uids.append(uid)
198
199			# current set UIDs
200			existing = self.to_list(self.get_raw(object))
201
202			# filter out new/removed UIDs
203			added_uids = [u for u in uids if u not in existing]
204			added_objs = filter(None, map(self.get_object, added_uids))
205
206			removed_uids = [u for u in existing if u not in uids]
207			removed_objs = filter(None, map(self.get_object, removed_uids))
208
209			# link backreferences of new uids
210			for added_obj in added_objs:
211			self.link_backref(added_obj, object)
212
213			# unlink backreferences of removed UIDs
214			for removed_obj in removed_objs:
215			self.unlink_backref(removed_obj, object)
216
217			super(UIDReferenceField, self).set(object, uids)
218
219			def unlink_backref(self, source, target):
220			"""Remove backreference from the source to the target
221
222			:param source: the object where the backref is stored (our reference)
223			:param target: the object where the backref points to (our object)
224			:returns: True when the backref was removed, False otherwise
225			"""
226
227			# Target might be a behavior instead of the object itself
228			target = self._get_content_object(target)
229
230			# This should be actually not possible
231			if self.is_initializing(target):
232			raise ValueError("Objects in initialization state "
233			"can not have existing back references!")
234
235			target_uid = self.get_uid(target)
236			# get the storage key
237			key = self.get_relationship_key(target)
238			# get all backreferences from the source
239			backrefs = get_backref_storage(source)
240			if key not in backrefs:
241			logger.warn(
242			"Referenced object {} has no backreferences for the key {}"
243			.format(repr(source), key))
244			return False
245			if target_uid not in backrefs[key]:
246			logger.warn("Target {} was not linked by {}"
247			.format(repr(target), repr(source)))
248			return False
249			backrefs[key].remove(target_uid)
250			return True
251
252			def link_backref(self, source, target):
253			"""Add backreference from the source to the target
254
255			:param source: the object where the backref is stored (our reference)
256			:param target: the object where the backref points to (our object)
257			:returns: True when the backref was written
258			"""
259
260			# Target might be a behavior instead of the object itself
261			target = self._get_content_object(target)
262
263			# Object is initializing and don't have an UID!
264			# -> Postpone to set back references in event handler
265			if self.is_initializing(target):
266			logger.info("Object is in initialization state. "
267			"Back references will be set in event handler")
268			return
269
270			target_uid = api.get_uid(target)
271			# get the annotation storage key
272			key = self.get_relationship_key(target)
273			# get all backreferences
274			backrefs = get_backref_storage(source)
275			if key not in backrefs:
276			backrefs[key] = PersistentList()
277			if target_uid not in backrefs[key]:
278			backrefs[key].append(target_uid)
279			return True
280
281			def get(self, object):
282			"""Get referenced objects
283
284			:param object: instance of the field
285			:returns: list of referenced objects
286			"""
287			return self._get(object, as_objects=True)
288
289			def get_raw(self, object):
290			"""Get referenced UIDs
291
292			NOTE: Called from the data manager `query` method
293			to get the widget value
294
295			:param object: instance of the field
296			:returns: list of referenced UIDs
297			"""
298			return self._get(object, as_objects=False)
299
300			def _get(self, object, as_objects=False):
301			"""Returns single/multi value
302
303			:param object: instance of the field
304			:param as_objects: Flag for UID/object returns
305			:returns: list of referenced UIDs
306			"""
307
308			# Object might be a behavior instead of the object itself
309			object = self._get_content_object(object)
310
311			uids = super(UIDReferenceField, self).get(object)
312
313			if not uids:
314			uids = []
315			elif not isinstance(uids, (list, tuple)):
316			uids = [uids]
317
318			if as_objects is True:
319			uids = filter(None, map(self.get_object, uids))
320
321			if self.multi_valued:
322			return uids
323			if len(uids) == 0:
324			return None
325			return uids[0]
326
327			def _validate(self, value):
328			"""Validator when called from form submission
329			"""
330			super(UIDReferenceField, self)._validate(value)
331			# check if the fields accepts single values only
332			if not self.multi_valued and len(value) > 1:
333			raise ValueError("Single valued field accepts at most 1 value")
334
335			# check for valid UIDs
336			for uid in value:
337			if not api.is_uid(uid):
338			raise ValueError("Invalid UID: '%s'" % uid)
339
340			# check if the type is allowed
341			allowed_types = self.get_allowed_types()
342			if allowed_types:
343			objs = filter(None, map(self.get_object, value))
344			types = set(map(api.get_portal_type, objs))
345			if not types.issubset(allowed_types):
346			raise ValueError("Only the following types are allowed: %s"
347			% ",".join(allowed_types))
348
349			def _get_content_object(self, thing):
350			"""Returns the underlying content object
351			"""
352			if IBehavior.providedBy(thing):
353			return self._get_content_object(thing.context)
354			if api.is_dexterity_content(thing):
355			return thing
356			raise ValueError("Not a valid object: %s" % repr(thing))
357

senaite / senaite.core

Push — 2.x ( 5710a9...27b54e )

UIDReferenceField.get_uid() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like