bika.lims.browser.fields.uidreferencefield._get_object() - Code Metrics - Inspection of "Fix traceback when value is ['null'] (#2220)" - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — 2.x ( bd408b...df866e )

by Jordi

created 2023-01-02 17:34 UTC

bika.lims.browser.fields.uidreferencefield._get_object() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	26
Code Lines	16

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	16
dl	0
loc	26
rs	8.6666
c	0
b	0
f	0
cc	6
nop	2

# -*- coding: utf-8 -*-
#
# This file is part of SENAITE.CORE.
#
# SENAITE.CORE is free software: you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation, version 2.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
# details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 51
# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# Copyright 2018-2021 by it's authors.
# Some rights reserved, see README and LICENSE.

import six

from AccessControl import ClassSecurityInfo
from Products.Archetypes.Field import Field, StringField
from bika.lims import logger
from bika.lims import api
from bika.lims.interfaces.field import IUIDReferenceField
from persistent.list import PersistentList
from persistent.dict import PersistentDict
from zope.annotation.interfaces import IAnnotations
from zope.interface import implements

BACKREFS_STORAGE = "bika.lims.browser.fields.uidreferencefield.backreferences"


class ReferenceException(Exception):
    pass


class UIDReferenceField(StringField):
    """A field that stores References as UID values.  This acts as a drop-in
    replacement for Archetypes' ReferenceField.  A relationship is required
    but if one is not provided, it will be composed from a concatenation
    of `portal_type` + `fieldname`.
    """
    _properties = Field._properties.copy()
    _properties.update({
        'type': 'uidreference',
        'default': '',
        'default_content_type': 'text/plain',
        'relationship': '',
    })

    implements(IUIDReferenceField)

    security = ClassSecurityInfo()

    def get_relationship_key(self, context):
        """Return the configured relationship key or generate a new one
        """
        if not self.relationship:
            return context.portal_type + self.getName()
        return self.relationship

    def link_reference(self, source, target):
        """Link the target to the source
        """
        target_uid = api.get_uid(target)
        # get the annotation storage key
        key = self.get_relationship_key(target)
        # get all backreferences from the source
        # N.B. only like this we get the persistent mapping!
        backrefs = get_backreferences(source, relationship=None)
        if key not in backrefs:
            backrefs[key] = PersistentList()
        if target_uid not in backrefs[key]:
            backrefs[key].append(target_uid)
        return True

    def unlink_reference(self, source, target):
        """Unlink the target from the source
        """
        target_uid = api.get_uid(target)
        # get the storage key
        key = self.get_relationship_key(target)
        # get all backreferences from the source
        # N.B. only like this we get the persistent mapping!
        backrefs = get_backreferences(source, relationship=None)
        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

    @security.public
    def get_uid(self, context, value, default=""):
        """Takes a brain or object (or UID), and returns a UID.

        :param context: context is the object who's schema contains this field.
        :type context: BaseContent
        :param value: Brain, object, or UID.
        :type value: Any
        :return: resolved UID.
        :rtype: string
        """
        if api.is_object(value):
            value = api.get_uid(value)
        elif not api.is_uid(value):
            value = default
        return value

    @security.public
    def get(self, context, **kwargs):
        """Grab the stored value, and resolve object(s) from UID catalog.

        :param context: context is the object who's schema contains this field.
        :type context: BaseContent
        :param kwargs: kwargs are passed directly to the underlying get.
        :type kwargs: dict
        :return: object or list of objects for multiValued fields.
        :rtype: BaseContent | list[BaseContent]
        """
        uids = StringField.get(self, context, **kwargs)
        if not isinstance(uids, list):
            uids = [uids]

        # Do a direct search for all brains at once
        uc = api.get_tool("uid_catalog")
        references = uc(UID=uids)

        # Keep the original order of items
        references = sorted(references, key=lambda it: uids.index(it.UID))

        # Return objects by default
        full_objects = kwargs.pop("full_objects", True)
        if full_objects:
            references = [api.get_object(ref) for ref in references]

        if self.multiValued:
            return references
        elif references:
            return references[0]
        return None

    @security.public
    def getRaw(self, context, aslist=False, **kwargs):
        """Grab the stored value, and return it directly as UIDs.

        :param context: context is the object who's schema contains this field.
        :type context: BaseContent
        :param aslist: Forces a single-valued field to return a list type.
        :type aslist: bool
        :param kwargs: kwargs are passed directly to the underlying get.
        :type kwargs: dict
        :return: UID or list of UIDs for multiValued fields.
        :rtype: string | list[string]
        """
        value = StringField.get(self, context, **kwargs)
        if not value:
            return [] if self.multiValued else None
        if self.multiValued:
            ret = value
        else:
            ret = self.get_uid(context, value)
            if aslist:
                ret = [ret]
        return ret

    def _set_backreferences(self, context, items, **kwargs):
        """Set the back references on the linked items

        This will set an annotation storage on the referenced items which point
        to the current context.
        """

        # Don't set any references during initialization.
        # This might cause a recursion error when calling `getRaw` to fetch the
        # current set UIDs!
        initializing = kwargs.get('_initializing_', False)
        if initializing:
            return

        # current set UIDs
        raw = self.getRaw(context) or []
        # handle single reference fields
        if isinstance(raw, six.string_types):
            raw = [raw, ]
        cur = set(raw)
        # UIDs to be set
        uids = set(map(api.get_uid, items))
        # removed UIDs
        removed = cur.difference(uids)
        # missing UIDs
        missing = uids.difference(cur)

        # Unlink removed UIDs from the source
        uc = api.get_tool("uid_catalog")
        for brain in uc(UID=list(removed)):
            source = api.get_object(brain)
            self.unlink_reference(source, context)

        # Link missing UIDs
        for brain in uc(UID=list(missing)):
            target = api.get_object(brain)
            self.link_reference(target, context)

    @security.public
    def set(self, context, value, **kwargs):
        """Accepts a UID, brain, or an object (or a list of any of these),
        and stores a UID or list of UIDS.

        :param context: context is the object who's schema contains this field.
        :type context: BaseContent
        :param value: A UID, brain or object (or a sequence of these).
        :type value: Any
        :param kwargs: kwargs are passed directly to the underlying get.
        :type kwargs: dict
        :return: None
        """
        if not isinstance(value, (list, tuple)):
            value = [value]

        # Extract uids and remove empties
        uids = [self.get_uid(context, item) for item in value]
        uids = filter(api.is_uid, uids)

        # Back-reference current object to referenced objects
        self._set_backreferences(context, uids, **kwargs)

        # Store the referenced objects as uids
        if not self.multiValued:
            uids = uids[0] if uids else ""
        StringField.set(self, context, uids, **kwargs)


def get_storage(context):
    annotation = IAnnotations(context)
    if annotation.get(BACKREFS_STORAGE) is None:

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


def _get_catalog_for_uid(uid):
    at = api.get_tool('archetype_tool')
    uc = api.get_tool('uid_catalog')
    pc = api.get_tool('portal_catalog')
    # get uid_catalog brain for uid
    ub = uc(UID=uid)[0]
    # get portal_type of brain
    pt = ub.portal_type
    # get the registered catalogs for portal_type
    cats = at.getCatalogsByType(pt)
    # try avoid 'portal_catalog'; XXX multiple catalogs in setuphandlers.py?
    cats = [cat for cat in cats if cat != pc]
    if cats:
        return cats[0]
    return pc


def get_backreferences(context, relationship=None, as_brains=None):
    """Return all objects which use a UIDReferenceField to reference context.

    :param context: The object which is the target of references.
    :param relationship: The relationship name of the UIDReferenceField.
    :param as_brains: Requests that this function returns only catalog brains.
        as_brains can only be used if a relationship has been specified.

    This function can be called with or without specifying a relationship.

    - If a relationship is provided, the return value will be a list of items
      which reference the context using the provided relationship.

      If relationship is provided, then you can request that the backrefs
      should be returned as catalog brains.  If you do not specify as_brains,
      the raw list of UIDs will be returned.

    - If the relationship is not provided, then the entire set of
      backreferences to the context object is returned (by reference) as a
      dictionary.  This value can then be modified in-place, to edit the stored
      backreferences.
    """

    instance = context.aq_base
    raw_backrefs = get_storage(instance)

    if not relationship:
        assert not as_brains, "You cannot use as_brains with no relationship"
        return raw_backrefs

    backrefs = list(raw_backrefs.get(relationship, []))
    if not backrefs:
        return []

    if not as_brains:
        return backrefs

    cat = _get_catalog_for_uid(backrefs[0])
    return cat(UID=backrefs)


1			# -- coding: utf-8 --
2			#
3			# This file is part of SENAITE.CORE.
4			#
5			# SENAITE.CORE is free software: you can redistribute it and/or modify it under
6			# the terms of the GNU General Public License as published by the Free Software
7			# Foundation, version 2.
8			#
9			# This program is distributed in the hope that it will be useful, but WITHOUT
10			# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
11			# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
12			# details.
13			#
14			# You should have received a copy of the GNU General Public License along with
15			# this program; if not, write to the Free Software Foundation, Inc., 51
16			# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
17			#
18			# Copyright 2018-2021 by it's authors.
19			# Some rights reserved, see README and LICENSE.
20
21			import six
22
23			from AccessControl import ClassSecurityInfo
24			from Products.Archetypes.Field import Field, StringField
25			from bika.lims import logger
26			from bika.lims import api
27			from bika.lims.interfaces.field import IUIDReferenceField
28			from persistent.list import PersistentList
29			from persistent.dict import PersistentDict
30			from zope.annotation.interfaces import IAnnotations
31			from zope.interface import implements
32
33			BACKREFS_STORAGE = "bika.lims.browser.fields.uidreferencefield.backreferences"
34
35
36			class ReferenceException(Exception):
37			pass
38
39
40			class UIDReferenceField(StringField):
41			"""A field that stores References as UID values. This acts as a drop-in
42			replacement for Archetypes' ReferenceField. A relationship is required
43			but if one is not provided, it will be composed from a concatenation
44			of `portal_type` + `fieldname`.
45			"""
46			_properties = Field._properties.copy()
47			_properties.update({
48			'type': 'uidreference',
49			'default': '',
50			'default_content_type': 'text/plain',
51			'relationship': '',
52			})
53
54			implements(IUIDReferenceField)
55
56			security = ClassSecurityInfo()
57
58			def get_relationship_key(self, context):
59			"""Return the configured relationship key or generate a new one
60			"""
61			if not self.relationship:
62			return context.portal_type + self.getName()
63			return self.relationship
64
65			def link_reference(self, source, target):
66			"""Link the target to the source
67			"""
68			target_uid = api.get_uid(target)
69			# get the annotation storage key
70			key = self.get_relationship_key(target)
71			# get all backreferences from the source
72			# N.B. only like this we get the persistent mapping!
73			backrefs = get_backreferences(source, relationship=None)
74			if key not in backrefs:
75			backrefs[key] = PersistentList()
76			if target_uid not in backrefs[key]:
77			backrefs[key].append(target_uid)
78			return True
79
80			def unlink_reference(self, source, target):
81			"""Unlink the target from the source
82			"""
83			target_uid = api.get_uid(target)
84			# get the storage key
85			key = self.get_relationship_key(target)
86			# get all backreferences from the source
87			# N.B. only like this we get the persistent mapping!
88			backrefs = get_backreferences(source, relationship=None)
89			if key not in backrefs:
90			logger.warn(
91			"Referenced object {} has no backreferences for the key {}"
92			.format(repr(source), key))
93			return False
94			if target_uid not in backrefs[key]:
95			logger.warn("Target {} was not linked by {}"
96			.format(repr(target), repr(source)))
97			return False
98			backrefs[key].remove(target_uid)
99			return True
100
101			@security.public
102			def get_uid(self, context, value, default=""):
103			"""Takes a brain or object (or UID), and returns a UID.
104
105			:param context: context is the object who's schema contains this field.
106			:type context: BaseContent
107			:param value: Brain, object, or UID.
108			:type value: Any
109			:return: resolved UID.
110			:rtype: string
111			"""
112			if api.is_object(value):
113			value = api.get_uid(value)
114			elif not api.is_uid(value):
115			value = default
116			return value
117
118			@security.public
119			def get(self, context, **kwargs):
120			"""Grab the stored value, and resolve object(s) from UID catalog.
121
122			:param context: context is the object who's schema contains this field.
123			:type context: BaseContent
124			:param kwargs: kwargs are passed directly to the underlying get.
125			:type kwargs: dict
126			:return: object or list of objects for multiValued fields.
127			:rtype: BaseContent \| list[BaseContent]
128			"""
129			uids = StringField.get(self, context, **kwargs)
130			if not isinstance(uids, list):
131			uids = [uids]
132
133			# Do a direct search for all brains at once
134			uc = api.get_tool("uid_catalog")
135			references = uc(UID=uids)
136
137			# Keep the original order of items
138			references = sorted(references, key=lambda it: uids.index(it.UID))
139
140			# Return objects by default
141			full_objects = kwargs.pop("full_objects", True)
142			if full_objects:
143			references = [api.get_object(ref) for ref in references]
144
145			if self.multiValued:
146			return references
147			elif references:
148			return references[0]
149			return None
150
151			@security.public
152			def getRaw(self, context, aslist=False, **kwargs):
153			"""Grab the stored value, and return it directly as UIDs.
154
155			:param context: context is the object who's schema contains this field.
156			:type context: BaseContent
157			:param aslist: Forces a single-valued field to return a list type.
158			:type aslist: bool
159			:param kwargs: kwargs are passed directly to the underlying get.
160			:type kwargs: dict
161			:return: UID or list of UIDs for multiValued fields.
162			:rtype: string \| list[string]
163			"""
164			value = StringField.get(self, context, **kwargs)
165			if not value:
166			return [] if self.multiValued else None
167			if self.multiValued:
168			ret = value
169			else:
170			ret = self.get_uid(context, value)
171			if aslist:
172			ret = [ret]
173			return ret
174
175			def _set_backreferences(self, context, items, **kwargs):
176			"""Set the back references on the linked items
177
178			This will set an annotation storage on the referenced items which point
179			to the current context.
180			"""
181
182			# Don't set any references during initialization.
183			# This might cause a recursion error when calling `getRaw` to fetch the
184			# current set UIDs!
185			initializing = kwargs.get('_initializing_', False)
186			if initializing:
187			return
188
189			# current set UIDs
190			raw = self.getRaw(context) or []
191			# handle single reference fields
192			if isinstance(raw, six.string_types):
193			raw = [raw, ]
194			cur = set(raw)
195			# UIDs to be set
196			uids = set(map(api.get_uid, items))
197			# removed UIDs
198			removed = cur.difference(uids)
199			# missing UIDs
200			missing = uids.difference(cur)
201
202			# Unlink removed UIDs from the source
203			uc = api.get_tool("uid_catalog")
204			for brain in uc(UID=list(removed)):
205			source = api.get_object(brain)
206			self.unlink_reference(source, context)
207
208			# Link missing UIDs
209			for brain in uc(UID=list(missing)):
210			target = api.get_object(brain)
211			self.link_reference(target, context)
212
213			@security.public
214			def set(self, context, value, **kwargs):
215			"""Accepts a UID, brain, or an object (or a list of any of these),
216			and stores a UID or list of UIDS.
217
218			:param context: context is the object who's schema contains this field.
219			:type context: BaseContent
220			:param value: A UID, brain or object (or a sequence of these).
221			:type value: Any
222			:param kwargs: kwargs are passed directly to the underlying get.
223			:type kwargs: dict
224			:return: None
225			"""
226			if not isinstance(value, (list, tuple)):
227			value = [value]
228
229			# Extract uids and remove empties
230			uids = [self.get_uid(context, item) for item in value]
231			uids = filter(api.is_uid, uids)
232
233			# Back-reference current object to referenced objects
234			self._set_backreferences(context, uids, **kwargs)
235
236			# Store the referenced objects as uids
237			if not self.multiValued:
238			uids = uids[0] if uids else ""
239			StringField.set(self, context, uids, **kwargs)
240
241
242			def get_storage(context):
243			annotation = IAnnotations(context)
244			if annotation.get(BACKREFS_STORAGE) is None:
			0 ignored issues – show Comprehensibility Best Practice introduced 2018-07-25 08:58 UTC by Report Bug Copy Issue Report The variable `BACKREFS_STORAGE` does not seem to be defined. Loading history...
245			annotation[BACKREFS_STORAGE] = PersistentDict()
246			return annotation[BACKREFS_STORAGE]
247
248
249			def _get_catalog_for_uid(uid):
250			at = api.get_tool('archetype_tool')
251			uc = api.get_tool('uid_catalog')
252			pc = api.get_tool('portal_catalog')
253			# get uid_catalog brain for uid
254			ub = uc(UID=uid)[0]
255			# get portal_type of brain
256			pt = ub.portal_type
257			# get the registered catalogs for portal_type
258			cats = at.getCatalogsByType(pt)
259			# try avoid 'portal_catalog'; XXX multiple catalogs in setuphandlers.py?
260			cats = [cat for cat in cats if cat != pc]
261			if cats:
262			return cats[0]
263			return pc
264
265
266			def get_backreferences(context, relationship=None, as_brains=None):
267			"""Return all objects which use a UIDReferenceField to reference context.
268
269			:param context: The object which is the target of references.
270			:param relationship: The relationship name of the UIDReferenceField.
271			:param as_brains: Requests that this function returns only catalog brains.
272			as_brains can only be used if a relationship has been specified.
273
274			This function can be called with or without specifying a relationship.
275
276			- If a relationship is provided, the return value will be a list of items
277			which reference the context using the provided relationship.
278
279			If relationship is provided, then you can request that the backrefs
280			should be returned as catalog brains. If you do not specify as_brains,
281			the raw list of UIDs will be returned.
282
283			- If the relationship is not provided, then the entire set of
284			backreferences to the context object is returned (by reference) as a
285			dictionary. This value can then be modified in-place, to edit the stored
286			backreferences.
287			"""
288
289			instance = context.aq_base
290			raw_backrefs = get_storage(instance)
291
292			if not relationship:
293			assert not as_brains, "You cannot use as_brains with no relationship"
294			return raw_backrefs
295
296			backrefs = list(raw_backrefs.get(relationship, []))
297			if not backrefs:
298			return []
299
300			if not as_brains:
301			return backrefs
302
303			cat = _get_catalog_for_uid(backrefs[0])
304			return cat(UID=backrefs)
305

senaite / senaite.core

Push — 2.x ( bd408b...df866e )

bika.lims.browser.fields.uidreferencefield._get_object() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like