senaite.core.schema.uidreferencefield.UIDReferenceField.get_raw() - Code Metrics - Inspection of "Fix issues with new UID reference field for dexter..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — 2.x (#1873)

by Jordi

created 2021-11-10 09:52 UTC

UIDReferenceField.get_raw() A

↳ Parent: senaite.core.schema.uidreferencefield

Complexity

Conditions

Size

Total Lines	10
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	2
dl	0
loc	10
rs	10
c	0
b	0
f	0
cc	1
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.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
        """

        # 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
        if not api.is_object(target):
            target = target.context

        # 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
        if not api.is_object(target):
            target = target.context

        # 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
        """

        # when creating a new object the context is the container
        # which does not have the field
        if self.interface and not self.interface.providedBy(object):
            if self.multi_valued:
                return []
            return None

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

        if not 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))


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

senaite / senaite.core

Pull Request — 2.x (#1873)

UIDReferenceField.get_raw() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like