build.bika.lims.api.snapshot._process_value() - Code Metrics - Inspection of "Reduce the noise in the diff for object paths (#13..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( f38d56...9d71bf )

by Jordi

created 2019-05-20 08:29 UTC

build.bika.lims.api.snapshot._process_value() C

↳ Parent: build.bika.lims.api.snapshot

Complexity

Conditions

Size

Total Lines	31
Code Lines	21

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	21
dl	0
loc	31
rs	5.9999
c	0
b	0
f	0
cc	10
nop	1

How to fix Complexity

# -*- 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-2019 by it's authors.
# Some rights reserved, see README and LICENSE.

import json

from bika.lims import _
from bika.lims import api
from bika.lims import logger
from bika.lims.api.security import get_roles
from bika.lims.api.security import get_user_id
from bika.lims.interfaces import IAuditable
from bika.lims.interfaces import IDoNotSupportSnapshots
from DateTime import DateTime
from persistent.list import PersistentList
from plone.memoize.ram import cache
from senaite.core.supermodel import SuperModel
from zope.annotation.interfaces import IAnnotatable
from zope.annotation.interfaces import IAnnotations
from zope.interface import alsoProvides

SNAPSHOT_STORAGE = "senaite.core.snapshots"


def _objectdata_cache_key(func, obj):
    """Cache Key for object data
    """
    uid = api.get_uid(obj)
    modified = api.get_modification_date(obj).millis()
    review_state = api.get_review_status(obj)
    return "{}-{}-{}".format(uid, review_state, modified)


def supports_snapshots(obj):
    """Checks if the object supports snapshots

    Only objects which can hold an annotation storage can be auditable

    :param obj: Content object
    :returns: True/False
    """
    if IDoNotSupportSnapshots.providedBy(obj):
        return False
    return IAnnotatable.providedBy(obj)


def get_storage(obj):
    """Get or create the audit log storage for the given object

    :param obj: Content object
    :returns: PersistentList
    """
    annotation = IAnnotations(obj)
    if annotation.get(SNAPSHOT_STORAGE) is None:

        annotation[SNAPSHOT_STORAGE] = PersistentList()
    return annotation[SNAPSHOT_STORAGE]


def get_snapshots(obj):
    """Get all snapshots from the storage

    :param obj: Content object
    :returns: List of snapshot dictionaries
    """
    snapshots = get_storage(obj)
    return map(json.loads, snapshots)


def has_snapshots(obj):
    """Checks if the object has snapshots

    :param obj: Content object
    :returns: True/False
    """
    storage = get_storage(obj)
    return len(storage) > 0


def get_snapshot_count(obj):
    """Returns the number of snapsots

    :param obj: Content object
    :returns: Current snapshots in the storage
    """
    try:
        annotation = IAnnotations(obj)
    except TypeError:
        return 0
    storage = annotation.get(SNAPSHOT_STORAGE, [])
    return len(storage)


def get_version(obj):
    """Returns the version of the object

    NOTE: Object versions start with 0!

    :param obj: Content object
    :returns: Current version of the object or -1
    """
    count = get_snapshot_count(obj)
    if count == 0:
        return -1
    return count - 1


def get_snapshot_by_version(obj, version=0):
    """Get a snapshot by version

    Snapshot versions begin with `0`, because this is the first index of the
    storage, which is a list.

    :param obj: Content object
    :param version: The index position of the snapshot in the storage
    :returns: Snapshot at the given index position
    """
    if version < 0:
        return None
    snapshots = get_snapshots(obj)
    if version > len(snapshots) - 1:
        return None
    return snapshots[version]


def get_snapshot_version(obj, snapshot):
    """Returns the version of the given snapshot

    :param obj: Content object
    :param snapshot: Snapshot dictionary
    :returns: Index where the object is lcated
    """
    snapshots = get_snapshots(obj)
    return snapshots.index(snapshot)


def get_last_snapshot(obj):
    """Get the last snapshot

    :param obj: Content object
    :returns: Last Snapshot or None
    """
    version = get_version(obj)
    return get_snapshot_by_version(obj, version)


def get_snapshot_metadata(snapshot):
    """Returns the snapshot metadata

    :param snapshot: Snapshot dictionary
    :returns: Metadata dictionary of the snapshot
    """
    return snapshot.get("__metadata__", {})


@cache(_objectdata_cache_key)
def get_object_data(obj):
    """Get object schema data

    NOTE: We RAM cache this data because it should only change when the object
    was modified!

    XXX: We need to set at least the modification date when we set fields in
    Ajax Listing when we take a snapshot there!

    :param obj: Content object
    :returns: Dictionary of extracted schema data
    """

    model = SuperModel(obj)
    try:
        data = model.to_dict()
    except Exception as exc:
        logger.error("Failed to get schema data for {}: {}"
                     .format(repr(obj), str(exc)))
        data = {}

    return data


def get_request_data(request=None):
    """Get request header/form data

    A typical request behind NGINX looks like this:

    {
        'CONNECTION_TYPE': 'close',
        'CONTENT_LENGTH': '52',
        'CONTENT_TYPE': 'application/x-www-form-urlencoded; charset=UTF-8',
        'GATEWAY_INTERFACE': 'CGI/1.1',
        'HTTP_ACCEPT': 'application/json, text/javascript, */*; q=0.01',
        'HTTP_ACCEPT_ENCODING': 'gzip, deflate, br',
        'HTTP_ACCEPT_LANGUAGE': 'de-DE,de;q=0.9,en-US;q=0.8,en;q=0.7',
        'HTTP_COOKIE': '_ga=GA1.2.1058345096.1522506452; ...',
        'HTTP_HOST': 'senaite.ridingbytes.com',
        'HTTP_ORIGIN': 'https://senaite.ridingbytes.com',
        'HTTP_REFERER': 'https://senaite.ridingbytes.com/clients/client-1/H2O-0054',
        'HTTP_USER_AGENT': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/73.0.3683.103 Safari/537.36',
        'HTTP_X_FORWARDED_FOR': '93.238.47.95',
        'HTTP_X_REAL_IP': '93.238.47.95',
        'HTTP_X_REQUESTED_WITH': 'XMLHttpRequest',
        'PATH_INFO': '/VirtualHostBase/https/senaite.ridingbytes.com/senaite/VirtualHostRoot//@@API/update',
        'PATH_TRANSLATED': '/VirtualHostBase/https/senaite.ridingbytes.com/senaite/VirtualHostRoot/@@API/update',
        'QUERY_STRING': '',
        'REMOTE_ADDR': '127.0.0.1',
        'REQUEST_METHOD': 'POST',
        'SCRIPT_NAME': '',
        'SERVER_NAME': 'localhost',
        'SERVER_PORT': '8081',
        'SERVER_PROTOCOL': 'HTTP/1.0',
        'SERVER_SOFTWARE': 'Zope/(2.13.28, python 2.7.12, linux2) ZServer/1.1',
        'channel.creation_time': 1556086048
    }

    :param request: Request object
    :returns: Dictionary of extracted request header/form data
    """ # noqa

    if request is None:
        # get the request
        request = api.get_request()

    # Happens in the test runner
    if not request:
        return {}

    # Try to obtain the real IP address of the client
    forwarded_for = request.get_header("X_FORWARDED_FOR")
    real_ip = request.get_header("X_REAL_IP")
    remote_address = request.get_header("REMOTE_ADDR")

    return {
        "comments": request.form.get("comments", ""),
        "remote_address": forwarded_for or real_ip or remote_address,
        "user_agent": request.get_header("HTTP_USER_AGENT"),
        "referer": request.get_header("HTTP_REFERER"),
    }


def get_object_metadata(obj, **kw):
    """Get object metadata

    :param obj: Content object
    :returns: Dictionary of extracted object metadata
    """

    # inject metadata of volatile data
    metadata = {
        "actor": get_user_id(),
        "roles": get_roles(),
        "action": "",
        "review_state": api.get_review_status(obj),
        "active": api.is_active(obj),
        "snapshot_created": DateTime().ISO(),
        "modified": api.get_modification_date(obj).ISO(),
        "remote_address": "",
        "user_agent": "",
        "referer": "",
        "comments": "",
    }

    # Update request data
    metadata.update(get_request_data())

    # allow metadata overrides
    metadata.update(kw)

    return metadata


def take_snapshot(obj, store=True, **kw):
    """Takes a snapshot of the passed in object

    :param obj: Content object
    :returns: New snapshot
    """
    logger.debug("📷 Take new snapshot for {}".format(repr(obj)))

    # get the object data
    snapshot = get_object_data(obj)

    # get the metadata
    metadata = get_object_metadata(obj, **kw)

    # store the metadata
    snapshot["__metadata__"] = metadata

    # convert the snapshot to JSON
    data = json.dumps(snapshot)

    # return immediately
    if not store:
        return snapshot

    # get the snapshot storage
    storage = get_storage(obj)

    # store the snapshot data
    storage.append(data)

    # Mark the content as auditable
    alsoProvides(obj, IAuditable)

    return snapshot


def compare_snapshots(snapshot_a, snapshot_b, raw=False):
    """Returns a diff of two given snapshots (dictionaries)

    :param snapshot_a: First snapshot
    :param snapshot_b: Second snapshot
    :param raw: True to compare the raw values, e.g. UIDs
    :returns: Dictionary of field/value pairs that differ
    """
    if not all(map(lambda x: isinstance(x, dict),
                   [snapshot_a, snapshot_b])):
        return {}

    diffs = {}
    for key_a, value_a in snapshot_a.iteritems():
        # skip fieds starting with _ or __
        if key_a.startswith("_"):
            continue
        # get the value of the second snapshot
        value_b = snapshot_b.get(key_a)
        # get the diff between the two values
        diff = diff_values(value_a, value_b, raw=raw)
        if diff is not None:
            diffs[key_a] = diff
    return diffs


def compare_last_two_snapshots(obj, raw=False):
    """Helper to compare the last two snapshots directly
    """

    if get_snapshot_count(obj) < 2:
        return {}

    version = get_version(obj)

    snap1 = get_snapshot_by_version(obj, version - 1)
    snap2 = get_snapshot_by_version(obj, version)

    return compare_snapshots(snap1, snap2, raw=raw)


def diff_values(value_a, value_b, raw=False):
    """Returns a human-readable diff between two values

    TODO: Provide an adapter per content type for this task to enable a more
          specific diff between the values

    :param value_a: First value to compare
    :param value_b: Second value to compare
    :param raw: True to compare the raw values, e.g. UIDs
    :returns a list of diff tuples
    """

    if not raw:
        value_a = _process_value(value_a)
        value_b = _process_value(value_b)

    # No changes
    if value_a == value_b:
        return None

    diffs = []
    # N.B.: the choice for the tuple data structure is to enable in the future
    # more granular diffs, e.g. the changed values within a dictionary etc.
    diffs.append((value_a, value_b))
    return diffs


def _process_value(value):
    """Convert the value into a human readable diff string
    """
    if not value:
        value = _("Not set")
    # handle strings
    elif isinstance(value, basestring):
        # XXX: bad data, e.g. in AS Method field
        if value == "None":
            value = _("Not set")
        # 0 is detected as the portal UID
        elif value == "0":
            value = "0"
        # handle physical paths
        elif value.startswith("/"):
            # remove the portal path to reduce noise in virtual hostings
            portal_path = api.get_path(api.get_portal())
            value = value.replace(portal_path, "", 1)
        elif api.is_uid(value):
            value = _get_title_or_id_from_uid(value)
    # handle dictionaries
    elif isinstance(value, (dict)):
        value = json.dumps(sorted(value.items()), indent=1)
    # handle lists and tuples
    elif isinstance(value, (list, tuple)):
        value = sorted(map(_process_value, value))
        value = "; ".join(value)
    # handle unicodes
    if isinstance(value, unicode):
        value = api.safe_unicode(value).encode("utf8")
    return str(value)


def _get_title_or_id_from_uid(uid):
    """Returns the title or ID from the given UID
    """
    try:
        obj = api.get_object_by_uid(uid)
    except api.APIError:
        return "<Deleted {}>".format(uid)
    title_or_id = api.get_title(obj) or api.get_id(obj)
    return title_or_id


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-2019 by it's authors.
19			# Some rights reserved, see README and LICENSE.
20
21			import json
22
23			from bika.lims import _
24			from bika.lims import api
25			from bika.lims import logger
26			from bika.lims.api.security import get_roles
27			from bika.lims.api.security import get_user_id
28			from bika.lims.interfaces import IAuditable
29			from bika.lims.interfaces import IDoNotSupportSnapshots
30			from DateTime import DateTime
31			from persistent.list import PersistentList
32			from plone.memoize.ram import cache
33			from senaite.core.supermodel import SuperModel
34			from zope.annotation.interfaces import IAnnotatable
35			from zope.annotation.interfaces import IAnnotations
36			from zope.interface import alsoProvides
37
38			SNAPSHOT_STORAGE = "senaite.core.snapshots"
39
40
41			def _objectdata_cache_key(func, obj):
42			"""Cache Key for object data
43			"""
44			uid = api.get_uid(obj)
45			modified = api.get_modification_date(obj).millis()
46			review_state = api.get_review_status(obj)
47			return "{}-{}-{}".format(uid, review_state, modified)
48
49
50			def supports_snapshots(obj):
51			"""Checks if the object supports snapshots
52
53			Only objects which can hold an annotation storage can be auditable
54
55			:param obj: Content object
56			:returns: True/False
57			"""
58			if IDoNotSupportSnapshots.providedBy(obj):
59			return False
60			return IAnnotatable.providedBy(obj)
61
62
63			def get_storage(obj):
64			"""Get or create the audit log storage for the given object
65
66			:param obj: Content object
67			:returns: PersistentList
68			"""
69			annotation = IAnnotations(obj)
70			if annotation.get(SNAPSHOT_STORAGE) is None:
			0 ignored issues – show Comprehensibility Best Practice introduced 2019-04-09 14:01 UTC by Report Bug Copy Issue Report The variable `SNAPSHOT_STORAGE` does not seem to be defined. Loading history...
71			annotation[SNAPSHOT_STORAGE] = PersistentList()
72			return annotation[SNAPSHOT_STORAGE]
73
74
75			def get_snapshots(obj):
76			"""Get all snapshots from the storage
77
78			:param obj: Content object
79			:returns: List of snapshot dictionaries
80			"""
81			snapshots = get_storage(obj)
82			return map(json.loads, snapshots)
83
84
85			def has_snapshots(obj):
86			"""Checks if the object has snapshots
87
88			:param obj: Content object
89			:returns: True/False
90			"""
91			storage = get_storage(obj)
92			return len(storage) > 0
93
94
95			def get_snapshot_count(obj):
96			"""Returns the number of snapsots
97
98			:param obj: Content object
99			:returns: Current snapshots in the storage
100			"""
101			try:
102			annotation = IAnnotations(obj)
103			except TypeError:
104			return 0
105			storage = annotation.get(SNAPSHOT_STORAGE, [])
106			return len(storage)
107
108
109			def get_version(obj):
110			"""Returns the version of the object
111
112			NOTE: Object versions start with 0!
113
114			:param obj: Content object
115			:returns: Current version of the object or -1
116			"""
117			count = get_snapshot_count(obj)
118			if count == 0:
119			return -1
120			return count - 1
121
122
123			def get_snapshot_by_version(obj, version=0):
124			"""Get a snapshot by version
125
126			Snapshot versions begin with `0`, because this is the first index of the
127			storage, which is a list.
128
129			:param obj: Content object
130			:param version: The index position of the snapshot in the storage
131			:returns: Snapshot at the given index position
132			"""
133			if version < 0:
134			return None
135			snapshots = get_snapshots(obj)
136			if version > len(snapshots) - 1:
137			return None
138			return snapshots[version]
139
140
141			def get_snapshot_version(obj, snapshot):
142			"""Returns the version of the given snapshot
143
144			:param obj: Content object
145			:param snapshot: Snapshot dictionary
146			:returns: Index where the object is lcated
147			"""
148			snapshots = get_snapshots(obj)
149			return snapshots.index(snapshot)
150
151
152			def get_last_snapshot(obj):
153			"""Get the last snapshot
154
155			:param obj: Content object
156			:returns: Last Snapshot or None
157			"""
158			version = get_version(obj)
159			return get_snapshot_by_version(obj, version)
160
161
162			def get_snapshot_metadata(snapshot):
163			"""Returns the snapshot metadata
164
165			:param snapshot: Snapshot dictionary
166			:returns: Metadata dictionary of the snapshot
167			"""
168			return snapshot.get("__metadata__", {})
169
170
171			@cache(_objectdata_cache_key)
172			def get_object_data(obj):
173			"""Get object schema data
174
175			NOTE: We RAM cache this data because it should only change when the object
176			was modified!
177
178			XXX: We need to set at least the modification date when we set fields in
179			Ajax Listing when we take a snapshot there!
180
181			:param obj: Content object
182			:returns: Dictionary of extracted schema data
183			"""
184
185			model = SuperModel(obj)
186			try:
187			data = model.to_dict()
188			except Exception as exc:
189			logger.error("Failed to get schema data for {}: {}"
190			.format(repr(obj), str(exc)))
191			data = {}
192
193			return data
194
195
196			def get_request_data(request=None):
197			"""Get request header/form data
198
199			A typical request behind NGINX looks like this:
200
201			{
202			'CONNECTION_TYPE': 'close',
203			'CONTENT_LENGTH': '52',
204			'CONTENT_TYPE': 'application/x-www-form-urlencoded; charset=UTF-8',
205			'GATEWAY_INTERFACE': 'CGI/1.1',
206			'HTTP_ACCEPT': 'application/json, text/javascript, /; q=0.01',
207			'HTTP_ACCEPT_ENCODING': 'gzip, deflate, br',
208			'HTTP_ACCEPT_LANGUAGE': 'de-DE,de;q=0.9,en-US;q=0.8,en;q=0.7',
209			'HTTP_COOKIE': '_ga=GA1.2.1058345096.1522506452; ...',
210			'HTTP_HOST': 'senaite.ridingbytes.com',
211			'HTTP_ORIGIN': 'https://senaite.ridingbytes.com',
212			'HTTP_REFERER': 'https://senaite.ridingbytes.com/clients/client-1/H2O-0054',
213			'HTTP_USER_AGENT': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/73.0.3683.103 Safari/537.36',
214			'HTTP_X_FORWARDED_FOR': '93.238.47.95',
215			'HTTP_X_REAL_IP': '93.238.47.95',
216			'HTTP_X_REQUESTED_WITH': 'XMLHttpRequest',
217			'PATH_INFO': '/VirtualHostBase/https/senaite.ridingbytes.com/senaite/VirtualHostRoot//@@API/update',
218			'PATH_TRANSLATED': '/VirtualHostBase/https/senaite.ridingbytes.com/senaite/VirtualHostRoot/@@API/update',
219			'QUERY_STRING': '',
220			'REMOTE_ADDR': '127.0.0.1',
221			'REQUEST_METHOD': 'POST',
222			'SCRIPT_NAME': '',
223			'SERVER_NAME': 'localhost',
224			'SERVER_PORT': '8081',
225			'SERVER_PROTOCOL': 'HTTP/1.0',
226			'SERVER_SOFTWARE': 'Zope/(2.13.28, python 2.7.12, linux2) ZServer/1.1',
227			'channel.creation_time': 1556086048
228			}
229
230			:param request: Request object
231			:returns: Dictionary of extracted request header/form data
232			""" # noqa
233
234			if request is None:
235			# get the request
236			request = api.get_request()
237
238			# Happens in the test runner
239			if not request:
240			return {}
241
242			# Try to obtain the real IP address of the client
243			forwarded_for = request.get_header("X_FORWARDED_FOR")
244			real_ip = request.get_header("X_REAL_IP")
245			remote_address = request.get_header("REMOTE_ADDR")
246
247			return {
248			"comments": request.form.get("comments", ""),
249			"remote_address": forwarded_for or real_ip or remote_address,
250			"user_agent": request.get_header("HTTP_USER_AGENT"),
251			"referer": request.get_header("HTTP_REFERER"),
252			}
253
254
255			def get_object_metadata(obj, **kw):
256			"""Get object metadata
257
258			:param obj: Content object
259			:returns: Dictionary of extracted object metadata
260			"""
261
262			# inject metadata of volatile data
263			metadata = {
264			"actor": get_user_id(),
265			"roles": get_roles(),
266			"action": "",
267			"review_state": api.get_review_status(obj),
268			"active": api.is_active(obj),
269			"snapshot_created": DateTime().ISO(),
270			"modified": api.get_modification_date(obj).ISO(),
271			"remote_address": "",
272			"user_agent": "",
273			"referer": "",
274			"comments": "",
275			}
276
277			# Update request data
278			metadata.update(get_request_data())
279
280			# allow metadata overrides
281			metadata.update(kw)
282
283			return metadata
284
285
286			def take_snapshot(obj, store=True, **kw):
287			"""Takes a snapshot of the passed in object
288
289			:param obj: Content object
290			:returns: New snapshot
291			"""
292			logger.debug("📷 Take new snapshot for {}".format(repr(obj)))
293
294			# get the object data
295			snapshot = get_object_data(obj)
296
297			# get the metadata
298			metadata = get_object_metadata(obj, **kw)
299
300			# store the metadata
301			snapshot["__metadata__"] = metadata
302
303			# convert the snapshot to JSON
304			data = json.dumps(snapshot)
305
306			# return immediately
307			if not store:
308			return snapshot
309
310			# get the snapshot storage
311			storage = get_storage(obj)
312
313			# store the snapshot data
314			storage.append(data)
315
316			# Mark the content as auditable
317			alsoProvides(obj, IAuditable)
318
319			return snapshot
320
321
322			def compare_snapshots(snapshot_a, snapshot_b, raw=False):
323			"""Returns a diff of two given snapshots (dictionaries)
324
325			:param snapshot_a: First snapshot
326			:param snapshot_b: Second snapshot
327			:param raw: True to compare the raw values, e.g. UIDs
328			:returns: Dictionary of field/value pairs that differ
329			"""
330			if not all(map(lambda x: isinstance(x, dict),
331			[snapshot_a, snapshot_b])):
332			return {}
333
334			diffs = {}
335			for key_a, value_a in snapshot_a.iteritems():
336			# skip fieds starting with _ or __
337			if key_a.startswith("_"):
338			continue
339			# get the value of the second snapshot
340			value_b = snapshot_b.get(key_a)
341			# get the diff between the two values
342			diff = diff_values(value_a, value_b, raw=raw)
343			if diff is not None:
344			diffs[key_a] = diff
345			return diffs
346
347
348			def compare_last_two_snapshots(obj, raw=False):
349			"""Helper to compare the last two snapshots directly
350			"""
351
352			if get_snapshot_count(obj) < 2:
353			return {}
354
355			version = get_version(obj)
356
357			snap1 = get_snapshot_by_version(obj, version - 1)
358			snap2 = get_snapshot_by_version(obj, version)
359
360			return compare_snapshots(snap1, snap2, raw=raw)
361
362
363			def diff_values(value_a, value_b, raw=False):
364			"""Returns a human-readable diff between two values
365
366			TODO: Provide an adapter per content type for this task to enable a more
367			specific diff between the values
368
369			:param value_a: First value to compare
370			:param value_b: Second value to compare
371			:param raw: True to compare the raw values, e.g. UIDs
372			:returns a list of diff tuples
373			"""
374
375			if not raw:
376			value_a = _process_value(value_a)
377			value_b = _process_value(value_b)
378
379			# No changes
380			if value_a == value_b:
381			return None
382
383			diffs = []
384			# N.B.: the choice for the tuple data structure is to enable in the future
385			# more granular diffs, e.g. the changed values within a dictionary etc.
386			diffs.append((value_a, value_b))
387			return diffs
388
389
390			def _process_value(value):
391			"""Convert the value into a human readable diff string
392			"""
393			if not value:
394			value = _("Not set")
395			# handle strings
396			elif isinstance(value, basestring):
397			# XXX: bad data, e.g. in AS Method field
398			if value == "None":
399			value = _("Not set")
400			# 0 is detected as the portal UID
401			elif value == "0":
402			value = "0"
403			# handle physical paths
404			elif value.startswith("/"):
405			# remove the portal path to reduce noise in virtual hostings
406			portal_path = api.get_path(api.get_portal())
407			value = value.replace(portal_path, "", 1)
408			elif api.is_uid(value):
409			value = _get_title_or_id_from_uid(value)
410			# handle dictionaries
411			elif isinstance(value, (dict)):
412			value = json.dumps(sorted(value.items()), indent=1)
413			# handle lists and tuples
414			elif isinstance(value, (list, tuple)):
415			value = sorted(map(_process_value, value))
416			value = "; ".join(value)
417			# handle unicodes
418			if isinstance(value, unicode):
419			value = api.safe_unicode(value).encode("utf8")
420			return str(value)
421
422
423			def _get_title_or_id_from_uid(uid):
424			"""Returns the title or ID from the given UID
425			"""
426			try:
427			obj = api.get_object_by_uid(uid)
428			except api.APIError:
429			return "<Deleted {}>".format(uid)
430			title_or_id = api.get_title(obj) or api.get_id(obj)
431			return title_or_id
432

senaite / senaite.core

Push — master ( f38d56...9d71bf )

build.bika.lims.api.snapshot._process_value() C

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like