senaite.core.schema.uidreferencefield.on_object_created() - Code Metrics - Inspection of "Added UID reference field/widget for Dexterity Con..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — 2.x (#1864)

by Ramon

created 2021-10-29 13:37 UTC

on_object_created() B

↳ Parent: senaite.core.schema.uidreferencefield

Complexity

Conditions

Size

Total Lines	29
Code Lines	18

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	18
dl	0
loc	29
rs	8.5666
c	0
b	0
f	0
cc	6
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 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
        if isinstance(value, six.string_types):
            value = [value]
        elif api.is_object(value):
            value = [value]
        elif value is None:
            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.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
        """

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

        # 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 get_relationship_key(self, context):
115			"""Relationship key used for backreferences
116
117			The key used for the annotation storage on the referenced object to
118			remember the current object UID.
119
120			:returns: storage key to lookup back references
121			"""
122			portal_type = api.get_portal_type(context)
123			return "%s.%s" % (portal_type, self.__name__)
124
125			def get_uid(self, value):
126			"""Value -> UID
127
128			:parm value: object/UID/SuperModel
129			:returns: UID
130			"""
131			try:
132			return api.get_uid(value)
133			except api.APIError:
134			return None
135
136			def get_object(self, value):
137			"""Value -> object
138
139			:returns: Object or None
140			"""
141			try:
142			return api.get_object(value)
143			except api.APIError:
144			return None
145
146			def get_allowed_types(self):
147			"""Returns the allowed reference types
148
149			:returns: tuple of allowed_types
150			"""
151			allowed_types = self.allowed_types
152			if not allowed_types:
153			allowed_types = ()
154			elif isinstance(allowed_types, six.string_types):
155			allowed_types = (allowed_types, )
156			return allowed_types
157
158			def set(self, object, value):
159			"""Set UID reference
160
161			:param object: the instance of the field
162			:param value: object/UID/SuperModel
163			:type value: list/tuple/str
164			"""
165
166			# always mark the object if references are set
167			# NOTE: there might be multiple UID reference field set on this object!
168			if value:
169			alsoProvides(object, IHaveUIDReferences)
170
171			# always handle all values internally as a list
172			if isinstance(value, six.string_types):
173			value = [value]
174			elif api.is_object(value):
175			value = [value]
176			elif value is None:
177			value = []
178
179			# convert to UIDs
180			uids = []
181			for v in value:
182			uid = self.get_uid(v)
183			if uid is None:
184			continue
185			uids.append(uid)
186
187			# current set UIDs
188			existing = self.get_raw(object)
189
190			# filter out new/removed UIDs
191			added_uids = [u for u in uids if u not in existing]
192			added_objs = filter(None, map(self.get_object, added_uids))
193
194			removed_uids = [u for u in existing if u not in uids]
195			removed_objs = filter(None, map(self.get_object, removed_uids))
196
197			# link backreferences of new uids
198			for added_obj in added_objs:
199			self.link_backref(added_obj, object)
200
201			# unlink backreferences of removed UIDs
202			for removed_obj in removed_objs:
203			self.unlink_backref(removed_obj, object)
204
205			super(UIDReferenceField, self).set(object, uids)
206
207			def unlink_backref(self, source, target):
208			"""Remove backreference from the source to the target
209
210			:param source: the object where the backref is stored (our reference)
211			:param target: the object where the backref points to (our object)
212			:returns: True when the backref was removed, False otherwise
213			"""
214
215			# This should be actually not possible
216			if self.is_initializing(target):
217			raise ValueError("Objects in initialization state "
218			"can not have existing back references!")
219
220			target_uid = self.get_uid(target)
221			# get the storage key
222			key = self.get_relationship_key(target)
223			# get all backreferences from the source
224			backrefs = get_backref_storage(source)
225			if key not in backrefs:
226			logger.warn(
227			"Referenced object {} has no backreferences for the key {}"
228			.format(repr(source), key))
229			return False
230			if target_uid not in backrefs[key]:
231			logger.warn("Target {} was not linked by {}"
232			.format(repr(target), repr(source)))
233			return False
234			backrefs[key].remove(target_uid)
235			return True
236
237			def link_backref(self, source, target):
238			"""Add backreference from the source to the target
239
240			:param source: the object where the backref is stored (our reference)
241			:param target: the object where the backref points to (our object)
242			:returns: True when the backref was written
243			"""
244
245			# Object is initializing and don't have an UID!
246			# -> Postpone to set back references in event handler
247			if self.is_initializing(target):
248			logger.info("Object is in initialization state. "
249			"Back references will be set in event handler")
250			return
251
252			target_uid = api.get_uid(target)
253			# get the annotation storage key
254			key = self.get_relationship_key(target)
255			# get all backreferences
256			backrefs = get_backref_storage(source)
257			if key not in backrefs:
258			backrefs[key] = PersistentList()
259			if target_uid not in backrefs[key]:
260			backrefs[key].append(target_uid)
261			return True
262
263			def get(self, object):
264			"""Get referenced objects
265
266			:param object: instance of the field
267			:returns: list of referenced objects
268			"""
269			return self._get(object, as_objects=True)
270
271			def get_raw(self, object):
272			"""Get referenced UIDs
273
274			NOTE: Called from the data manager `query` method
275			to get the widget value
276
277			:param object: instance of the field
278			:returns: list of referenced UIDs
279			"""
280			return self._get(object, as_objects=False)
281
282			def _get(self, object, as_objects=False):
283			"""Returns single/multi value
284
285			:param object: instance of the field
286			:param as_objects: Flag for UID/object returns
287			:returns: list of referenced UIDs
288			"""
289
290			# when creating a new object the context is the container
291			# which does not have the field
292			if self.interface and not self.interface.providedBy(object):
293			if self.multi_valued:
294			return []
295			return None
296
297			uids = super(UIDReferenceField, self).get(object)
298
299			if not uids:
300			uids = []
301
302			if as_objects is True:
303			uids = filter(None, map(self.get_object, uids))
304
305			if self.multi_valued:
306			return uids
307			if len(uids) == 0:
308			return None
309			return uids[0]
310
311			def _validate(self, value):
312			"""Validator when called from form submission
313			"""
314			super(UIDReferenceField, self)._validate(value)
315			# check if the fields accepts single values only
316			if not self.multi_valued and len(value) > 1:
317			raise ValueError("Single valued field accepts at most 1 value")
318
319			# check for valid UIDs
320			for uid in value:
321			if not api.is_uid(uid):
322			raise ValueError("Invalid UID: '%s'" % uid)
323
324			# check if the type is allowed
325			allowed_types = self.get_allowed_types()
326			if allowed_types:
327			objs = filter(None, map(self.get_object, value))
328			types = set(map(api.get_portal_type, objs))
329			if not types.issubset(allowed_types):
330			raise ValueError("Only the following types are allowed: %s"
331			% ",".join(allowed_types))
332

senaite / senaite.core

Pull Request — 2.x (#1864)

on_object_created() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like