bika.lims.browser.fields.uidreferencefield._get_object() - Code Metrics - Inspection of "Refactored and fixed inconsistencies with Analysis..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( caa188...9d0ad3 )

by Ramon

created 2018-10-01 19:11 UTC

_get_object() B

↳ Parent: bika.lims.browser.fields.uidreferencefield

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
#
# Copyright 2018 by it's authors.
# Some rights reserved. See LICENSE.rst, CONTRIBUTORS.rst.

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_object(self, context, value):
        """Resolve a UID to an object.

        :param context: context is the object containing the field's schema.
        :type context: BaseContent
        :param value: A UID.
        :type value: string
        :return: Returns a Content object.
        :rtype: BaseContent
        """
        if not value:
            return None
        obj = _get_object(context, value)
        if obj is None:
            logger.warning(
                "{}.{}: Resolving UIDReference failed for {}.  No object will "
                "be returned.".format(context, self.getName(), value))
        return obj

    @security.public
    def get_uid(self, context, value):
        """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
        """
        # Empty string or list with single empty string, are commonly
        # passed to us from form submissions
        if not value or value == ['']:
            ret = ''
        elif api.is_brain(value):
            ret = value.UID
        elif api.is_at_content(value) or api.is_dexterity_content(value):
            ret = value.UID()
        elif api.is_uid(value):
            ret = value
        else:
            raise ReferenceException("{}.{}: Cannot resolve UID for {}".format(
                context, self.getName(), value))
        return ret

    @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]
        """
        value = StringField.get(self, context, **kwargs)
        if not value:
            return [] if self.multiValued else None
        if self.multiValued:
            # Only return objects which actually exist; this is necessary here
            # because there are no HoldingReferences. This opens the
            # possibility that deletions leave hanging references.
            ret = filter(
                lambda x: x, [self.get_object(context, uid) for uid in value])
        else:
            ret = self.get_object(context, value)
        return ret

    @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

        # UID of the current object
        uid = api.get_uid(context)
        # current set UIDs
        cur = set(self.getRaw(context) or [])
        # UIDs to be set
        new = set(map(api.get_uid, items))
        # removed UIDs
        removed = cur.difference(new)

        # Unlink removed UIDs from the source
        for uid in removed:
            source = api.get_object_by_uid(uid, None)
            if source is None:
                logger.warn("UID {} does not exist anymore".format(uid))
                continue
            self.unlink_reference(source, context)

        # Link backrefs
        for item in items:
            self.link_reference(item, 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 self.multiValued:
            if not value:
                value = []
            if type(value) not in (list, tuple):
                value = [value, ]
            ret = [self.get_object(context, val) for val in value if val]
            self._set_backreferences(context, ret, **kwargs)
            uids = [self.get_uid(context, r) for r in ret if r]
            StringField.set(self, context, uids, **kwargs)
        else:
            # Sometimes we get given a list here with an empty string.
            # This is generated by html forms with empty values.
            # This is a single-valued field though, so:
            if isinstance(value, list) and value:
                if len(value) > 1:
                    logger.warning(
                        "Found values '\'{}\'' for singleValued field <{}>.{} "
                        "- using only the first value in the list.".format(
                            '\',\''.join(value), context.UID(), self.getName()))
                value = value[0]
            ret = self.get_object(context, value)
            if ret:
                self._set_backreferences(context, [ret, ], **kwargs)
                uid = self.get_uid(context, ret)
                StringField.set(self, context, uid, **kwargs)
            else:
                StringField.set(self, context, '', **kwargs)


def _get_object(context, value):
    """Resolve a UID to an object.

    :param context: context is the object containing the field's schema.
    :type context: BaseContent
    :param value: A UID.
    :type value: string
    :return: Returns a Content object or None.
    :rtype: BaseContent
    """
    if not value:
        return None
    if api.is_brain(value):
        return api.get_object(value)
    if api.is_object(value):
        return value
    if api.is_uid(value):
        uc = api.get_tool('uid_catalog', context=context)
        brains = uc(UID=value)
        if len(brains) == 0:
            # Broken Reference!
            logger.warn("Reference on {} with UID {} is broken!"
                        .format(repr(context), value))
            return None
        return brains[0].getObject()
    return None


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			# Copyright 2018 by it's authors.
6			# Some rights reserved. See LICENSE.rst, CONTRIBUTORS.rst.
7
8			from AccessControl import ClassSecurityInfo
9			from Products.Archetypes.Field import Field, StringField
10			from bika.lims import logger
11			from bika.lims import api
12			from bika.lims.interfaces.field import IUIDReferenceField
13			from persistent.list import PersistentList
14			from persistent.dict import PersistentDict
15			from zope.annotation.interfaces import IAnnotations
16			from zope.interface import implements
17
18			BACKREFS_STORAGE = "bika.lims.browser.fields.uidreferencefield.backreferences"
19
20
21			class ReferenceException(Exception):
22			pass
23
24
25			class UIDReferenceField(StringField):
26			"""A field that stores References as UID values. This acts as a drop-in
27			replacement for Archetypes' ReferenceField. A relationship is required
28			but if one is not provided, it will be composed from a concatenation
29			of `portal_type` + `fieldname`.
30			"""
31			_properties = Field._properties.copy()
32			_properties.update({
33			'type': 'uidreference',
34			'default': '',
35			'default_content_type': 'text/plain',
36			'relationship': '',
37			})
38
39			implements(IUIDReferenceField)
40
41			security = ClassSecurityInfo()
42
43			def get_relationship_key(self, context):
44			"""Return the configured relationship key or generate a new one
45			"""
46			if not self.relationship:
47			return context.portal_type + self.getName()
48			return self.relationship
49
50			def link_reference(self, source, target):
51			"""Link the target to the source
52			"""
53			target_uid = api.get_uid(target)
54			# get the annotation storage key
55			key = self.get_relationship_key(target)
56			# get all backreferences from the source
57			# N.B. only like this we get the persistent mapping!
58			backrefs = get_backreferences(source, relationship=None)
59			if key not in backrefs:
60			backrefs[key] = PersistentList()
61			if target_uid not in backrefs[key]:
62			backrefs[key].append(target_uid)
63			return True
64
65			def unlink_reference(self, source, target):
66			"""Unlink the target from the source
67			"""
68			target_uid = api.get_uid(target)
69			# get the 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			logger.warn(
76			"Referenced object {} has no backreferences for the key {}"
77			.format(repr(source), key))
78			return False
79			if target_uid not in backrefs[key]:
80			logger.warn("Target {} was not linked by {}"
81			.format(repr(target), repr(source)))
82			return False
83			backrefs[key].remove(target_uid)
84			return True
85
86			@security.public
87			def get_object(self, context, value):
88			"""Resolve a UID to an object.
89
90			:param context: context is the object containing the field's schema.
91			:type context: BaseContent
92			:param value: A UID.
93			:type value: string
94			:return: Returns a Content object.
95			:rtype: BaseContent
96			"""
97			if not value:
98			return None
99			obj = _get_object(context, value)
100			if obj is None:
101			logger.warning(
102			"{}.{}: Resolving UIDReference failed for {}. No object will "
103			"be returned.".format(context, self.getName(), value))
104			return obj
105
106			@security.public
107			def get_uid(self, context, value):
108			"""Takes a brain or object (or UID), and returns a UID.
109
110			:param context: context is the object who's schema contains this field.
111			:type context: BaseContent
112			:param value: Brain, object, or UID.
113			:type value: Any
114			:return: resolved UID.
115			:rtype: string
116			"""
117			# Empty string or list with single empty string, are commonly
118			# passed to us from form submissions
119			if not value or value == ['']:
120			ret = ''
121			elif api.is_brain(value):
122			ret = value.UID
123			elif api.is_at_content(value) or api.is_dexterity_content(value):
124			ret = value.UID()
125			elif api.is_uid(value):
126			ret = value
127			else:
128			raise ReferenceException("{}.{}: Cannot resolve UID for {}".format(
129			context, self.getName(), value))
130			return ret
131
132			@security.public
133			def get(self, context, **kwargs):
134			"""Grab the stored value, and resolve object(s) from UID catalog.
135
136			:param context: context is the object who's schema contains this field.
137			:type context: BaseContent
138			:param kwargs: kwargs are passed directly to the underlying get.
139			:type kwargs: dict
140			:return: object or list of objects for multiValued fields.
141			:rtype: BaseContent \| list[BaseContent]
142			"""
143			value = StringField.get(self, context, **kwargs)
144			if not value:
145			return [] if self.multiValued else None
146			if self.multiValued:
147			# Only return objects which actually exist; this is necessary here
148			# because there are no HoldingReferences. This opens the
149			# possibility that deletions leave hanging references.
150			ret = filter(
151			lambda x: x, [self.get_object(context, uid) for uid in value])
152			else:
153			ret = self.get_object(context, value)
154			return ret
155
156			@security.public
157			def getRaw(self, context, aslist=False, **kwargs):
158			"""Grab the stored value, and return it directly as UIDs.
159
160			:param context: context is the object who's schema contains this field.
161			:type context: BaseContent
162			:param aslist: Forces a single-valued field to return a list type.
163			:type aslist: bool
164			:param kwargs: kwargs are passed directly to the underlying get.
165			:type kwargs: dict
166			:return: UID or list of UIDs for multiValued fields.
167			:rtype: string \| list[string]
168			"""
169			value = StringField.get(self, context, **kwargs)
170			if not value:
171			return [] if self.multiValued else None
172			if self.multiValued:
173			ret = value
174			else:
175			ret = self.get_uid(context, value)
176			if aslist:
177			ret = [ret]
178			return ret
179
180			def _set_backreferences(self, context, items, **kwargs):
181			"""Set the back references on the linked items
182
183			This will set an annotation storage on the referenced items which point
184			to the current context.
185			"""
186
187			# Don't set any references during initialization.
188			# This might cause a recursion error when calling `getRaw` to fetch the
189			# current set UIDs!
190			initializing = kwargs.get('_initializing_', False)
191			if initializing:
192			return
193
194			# UID of the current object
195			uid = api.get_uid(context)
196			# current set UIDs
197			cur = set(self.getRaw(context) or [])
198			# UIDs to be set
199			new = set(map(api.get_uid, items))
200			# removed UIDs
201			removed = cur.difference(new)
202
203			# Unlink removed UIDs from the source
204			for uid in removed:
205			source = api.get_object_by_uid(uid, None)
206			if source is None:
207			logger.warn("UID {} does not exist anymore".format(uid))
208			continue
209			self.unlink_reference(source, context)
210
211			# Link backrefs
212			for item in items:
213			self.link_reference(item, context)
214
215			@security.public
216			def set(self, context, value, **kwargs):
217			"""Accepts a UID, brain, or an object (or a list of any of these),
218			and stores a UID or list of UIDS.
219
220			:param context: context is the object who's schema contains this field.
221			:type context: BaseContent
222			:param value: A UID, brain or object (or a sequence of these).
223			:type value: Any
224			:param kwargs: kwargs are passed directly to the underlying get.
225			:type kwargs: dict
226			:return: None
227			"""
228			if self.multiValued:
229			if not value:
230			value = []
231			if type(value) not in (list, tuple):
232			value = [value, ]
233			ret = [self.get_object(context, val) for val in value if val]
234			self._set_backreferences(context, ret, **kwargs)
235			uids = [self.get_uid(context, r) for r in ret if r]
236			StringField.set(self, context, uids, **kwargs)
237			else:
238			# Sometimes we get given a list here with an empty string.
239			# This is generated by html forms with empty values.
240			# This is a single-valued field though, so:
241			if isinstance(value, list) and value:
242			if len(value) > 1:
243			logger.warning(
244			"Found values '\'{}\'' for singleValued field <{}>.{} "
245			"- using only the first value in the list.".format(
246			'\',\''.join(value), context.UID(), self.getName()))
247			value = value[0]
248			ret = self.get_object(context, value)
249			if ret:
250			self._set_backreferences(context, [ret, ], **kwargs)
251			uid = self.get_uid(context, ret)
252			StringField.set(self, context, uid, **kwargs)
253			else:
254			StringField.set(self, context, '', **kwargs)
255
256
257			def _get_object(context, value):
258			"""Resolve a UID to an object.
259
260			:param context: context is the object containing the field's schema.
261			:type context: BaseContent
262			:param value: A UID.
263			:type value: string
264			:return: Returns a Content object or None.
265			:rtype: BaseContent
266			"""
267			if not value:
268			return None
269			if api.is_brain(value):
270			return api.get_object(value)
271			if api.is_object(value):
272			return value
273			if api.is_uid(value):
274			uc = api.get_tool('uid_catalog', context=context)
275			brains = uc(UID=value)
276			if len(brains) == 0:
277			# Broken Reference!
278			logger.warn("Reference on {} with UID {} is broken!"
279			.format(repr(context), value))
280			return None
281			return brains[0].getObject()
282			return None
283
284
285			def get_storage(context):
286			annotation = IAnnotations(context)
287			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...
288			annotation[BACKREFS_STORAGE] = PersistentDict()
289			return annotation[BACKREFS_STORAGE]
290
291
292			def _get_catalog_for_uid(uid):
293			at = api.get_tool('archetype_tool')
294			uc = api.get_tool('uid_catalog')
295			pc = api.get_tool('portal_catalog')
296			# get uid_catalog brain for uid
297			ub = uc(UID=uid)[0]
298			# get portal_type of brain
299			pt = ub.portal_type
300			# get the registered catalogs for portal_type
301			cats = at.getCatalogsByType(pt)
302			# try avoid 'portal_catalog'; XXX multiple catalogs in setuphandlers.py?
303			cats = [cat for cat in cats if cat != pc]
304			if cats:
305			return cats[0]
306			return pc
307
308
309			def get_backreferences(context, relationship=None, as_brains=None):
310			"""Return all objects which use a UIDReferenceField to reference context.
311
312			:param context: The object which is the target of references.
313			:param relationship: The relationship name of the UIDReferenceField.
314			:param as_brains: Requests that this function returns only catalog brains.
315			as_brains can only be used if a relationship has been specified.
316
317			This function can be called with or without specifying a relationship.
318
319			- If a relationship is provided, the return value will be a list of items
320			which reference the context using the provided relationship.
321
322			If relationship is provided, then you can request that the backrefs
323			should be returned as catalog brains. If you do not specify as_brains,
324			the raw list of UIDs will be returned.
325
326			- If the relationship is not provided, then the entire set of
327			backreferences to the context object is returned (by reference) as a
328			dictionary. This value can then be modified in-place, to edit the stored
329			backreferences.
330			"""
331
332			instance = context.aq_base
333			raw_backrefs = get_storage(instance)
334
335			if not relationship:
336			assert not as_brains, "You cannot use as_brains with no relationship"
337			return raw_backrefs
338
339			backrefs = list(raw_backrefs.get(relationship, []))
340			if not backrefs:
341			return []
342
343			if not as_brains:
344			return backrefs
345
346			cat = _get_catalog_for_uid(backrefs[0])
347			return cat(UID=backrefs)
348

senaite / senaite.core

Push — master ( caa188...9d0ad3 )

_get_object() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like