bika.lims.idserver.slice() - Code Metrics - Inspection of "Sample cleanup (#1180)" - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Branch — master (9edffc)

by Jordi

created 2019-01-04 23:53 UTC

bika.lims.idserver.slice() A

↳ Parent: bika.lims.idserver

Complexity

Conditions

Size

Total Lines	15
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	7
dl	0
loc	15
rs	10
c	0
b	0
f	0
cc	1
nop	4

# -*- coding: utf-8 -*-
#
# This file is part of SENAITE.CORE
#
# Copyright 2018 by it's authors.
# Some rights reserved. See LICENSE.rst, CONTRIBUTORS.rst.

import re

import transaction
from bika.lims import api
from bika.lims import logger
from bika.lims.alphanumber import Alphanumber
from bika.lims.alphanumber import to_alpha
from bika.lims.browser.fields.uidreferencefield import \
    get_backreferences as get_backuidreferences
from bika.lims.interfaces import IIdServer
from bika.lims.numbergenerator import INumberGenerator
from DateTime import DateTime
from Products.ATContentTypes.utils import DT2dt
from zope.component import getAdapters
from zope.component import getUtility


def get_objects_in_sequence(brain_or_object, ctype, cref):
    """Return a list of items
    """
    obj = api.get_object(brain_or_object)
    if ctype == "backreference":
        return get_backreferences(obj, cref)
    if ctype == "contained":
        return get_contained_items(obj, cref)
    raise ValueError("Reference value is mandatory for sequence type counter")


def get_backreferences(obj, relationship):
    """Returns the backreferences
    """
    refs = get_backuidreferences(obj, relationship)

    # TODO remove after all ReferenceField get ported to UIDReferenceField
    # At this moment, there are still some content types that are using the
    # ReferenceField, so we need to fallback to traditional getBackReferences
    # for these cases.
    if not refs:
        refs = obj.getBackReferences(relationship)

    return refs


def get_contained_items(obj, spec):
    """Returns a list of (id, subobject) tuples of the current context.
    If 'spec' is specified, returns only objects whose meta_type match 'spec'
    """
    return obj.objectItems(spec)


def get_config(context, **kw):
    """Fetch the config dict from the Bika Setup for the given portal_type
    """
    # get the ID formatting config
    config_map = api.get_bika_setup().getIDFormatting()

    # allow portal_type override
    portal_type = kw.get("portal_type") or api.get_portal_type(context)

    # check if we have a config for the given portal_type
    for config in config_map:
        if config['portal_type'].lower() == portal_type.lower():
            return config

    # return a default config
    default_config = {
        'form': '%s-{seq}' % portal_type.lower(),
        'sequence_type': 'generated',
        'prefix': '%s' % portal_type.lower(),
    }
    return default_config


def get_variables(context, **kw):
    """Prepares a dictionary of key->value pairs usable for ID formatting
    """

    # allow portal_type override
    portal_type = kw.get("portal_type") or api.get_portal_type(context)

    # The variables map hold the values that might get into the constructed id
    variables = {
        'context': context,
        'id': api.get_id(context),
        'portal_type': portal_type,
        'year': get_current_year(),
        'parent': api.get_parent(context),
        'seq': 0,
        'alpha': Alphanumber(0),
    }

    # Augment the variables map depending on the portal type
    if portal_type == "AnalysisRequest":
        now = DateTime()
        sampling_date = context.getSamplingDate()
        sampling_date = sampling_date and DT2dt(sampling_date) or DT2dt(now)
        date_sampled = context.getDateSampled()
        date_sampled = date_sampled and DT2dt(date_sampled) or DT2dt(now)
        variables.update({
            'clientId': context.getClientID(),
            'dateSampled': date_sampled,
            'samplingDate': sampling_date,
            'sampleType': context.getSampleType().getPrefix(),
        })

    elif portal_type == "ARReport":
        variables.update({
            'clientId': context.aq_parent.getClientID(),
        })

    return variables


def split(string, separator="-"):
    """ split a string on the given separator
    """
    if not isinstance(string, basestring):
        return []
    return string.split(separator)


def to_int(thing, default=0):
    """Convert a thing to an integer
    """
    try:
        return int(thing)
    except (TypeError, ValueError):
        return default


def slice(string, separator="-", start=None, end=None):
    """Slice out a segment of a string, which is splitted on separator.
    """

    # split the given string at the given separator
    segments = split(string, separator)

    # get the start and endposition for slicing
    length = len(segments)
    start = to_int(start)
    end = to_int(end, length)

    # return the separator joined sliced segments
    sliced_parts = segments[start:end]
    return separator.join(sliced_parts)


def get_current_year():
    """Returns the current year as a two digit string
    """
    return DateTime().strftime("%Y")[2:]


def search_by_prefix(portal_type, prefix):
    """Returns brains which share the same portal_type and ID prefix
    """
    catalog = api.get_tool("uid_catalog")
    brains = catalog({"portal_type": portal_type})
    # Filter brains with the same ID prefix
    return filter(lambda brain: api.get_id(brain).startswith(prefix), brains)


def get_ids_with_prefix(portal_type, prefix):
    """Return a list of ids sharing the same portal type and prefix
    """
    brains = search_by_prefix(portal_type, prefix)
    ids = map(api.get_id, brains)
    return ids


def make_storage_key(portal_type, prefix=None):
    """Make a storage (dict-) key for the number generator
    """
    key = portal_type.lower()
    if prefix:
        key = "{}-{}".format(key, prefix)
    return key


def get_seq_number_from_id(id, id_template, prefix, **kw):
    """Return the sequence number of the given ID
    """
    separator = kw.get("separator", "-")
    postfix = id.replace(prefix, "").strip(separator)
    postfix_segments = postfix.split(separator)
    seq_number = 0
    possible_seq_nums = filter(lambda n: n.isalnum(), postfix_segments)
    if possible_seq_nums:
        seq_number = possible_seq_nums[-1]

    # Check if this id has to be expressed as an alphanumeric number
    seq_number = get_alpha_or_number(seq_number, id_template)
    seq_number = to_int(seq_number)
    return seq_number


def get_alpha_or_number(number, template):
    """Returns an Alphanumber that represents the number passed in, expressed
    as defined in the template. Otherwise, returns the number
    """
    match = re.match(r".*\{alpha:(\d+a\d+d)\}$", template.strip())
    if match and match.groups():
        format = match.groups()[0]
        return to_alpha(number, format)
    return number


def get_counted_number(context, config, variables, **kw):
    """Compute the number for the sequence type "Counter"
    """
    # This "context" is defined by the user in Bika Setup and can be actually
    # anything. However, we assume it is something like "sample" or similar
    ctx = config.get("context")

    # get object behind the context name (falls back to the current context)
    obj = variables.get(ctx, context)

    # get the counter type, which is either "backreference" or "contained"
    counter_type = config.get("counter_type")

    # the counter reference is either the "relationship" for
    # "backreference" or the meta type for contained objects
    counter_reference = config.get("counter_reference")

    # This should be a list of existing items, including the current context
    # object
    seq_items = get_objects_in_sequence(obj, counter_type, counter_reference)

    number = len(seq_items)
    return number


def get_generated_number(context, config, variables, **kw):
    """Generate a new persistent number with the number generator for the
    sequence type "Generated"
    """

    # separator where to split the ID
    separator = kw.get('separator', '-')

    # allow portal_type override
    portal_type = kw.get("portal_type") or api.get_portal_type(context)

    # The ID format for string interpolation, e.g. WS-{seq:03d}
    id_template = config.get("form", "")

    # The split length defines where the variable part of the ID template begins
    split_length = config.get("split_length", 1)

    # The prefix tempalte is the static part of the ID
    prefix_template = slice(id_template, separator=separator, end=split_length)

    # get the number generator
    number_generator = getUtility(INumberGenerator)

    # generate the key for the number generator storage
    prefix = prefix_template.format(**variables)

    # normalize out any unicode characters like Ö, É, etc. from the prefix
    prefix = api.normalize_filename(prefix)

    # The key used for the storage
    key = make_storage_key(portal_type, prefix)

    # Handle flushed storage
    if key not in number_generator:
        max_num = 0
        existing = get_ids_with_prefix(portal_type, prefix)
        numbers = map(lambda id: get_seq_number_from_id(id, id_template, prefix), existing)
        # figure out the highest number in the sequence
        if numbers:
            max_num = max(numbers)
        # set the number generator
        logger.info("*** SEEDING Prefix '{}' to {}".format(prefix, max_num))
        number_generator.set_number(key, max_num)

    if not kw.get("dry_run", False):
        # Generate a new number
        # NOTE Even when the number exceeds the given ID sequence format,
        #      it will overflow gracefully, e.g.
        #      >>> {sampleId}-R{seq:03d}'.format(sampleId="Water", seq=999999)
        #      'Water-R999999‘
        number = number_generator.generate_number(key=key)
    else:
        # => This allows us to "preview" the next generated ID in the UI
        # TODO Show the user the next generated number somewhere in the UI
        number = number_generator.get(key, 1)

    # Return an int or Alphanumber
    return get_alpha_or_number(number, id_template)


def generateUniqueId(context, **kw):
    """ Generate pretty content IDs.
    """

    # get the config for this portal type from the system setup
    config = get_config(context, **kw)

    # get the variables map for later string interpolation
    variables = get_variables(context, **kw)

    # The new generate sequence number
    number = 0

    # get the sequence type from the global config
    sequence_type = config.get("sequence_type", "generated")

    # Sequence Type is "Counter", so we use the length of the backreferences or
    # contained objects of the evaluated "context" defined in the config
    if sequence_type == 'counter':
        number = get_counted_number(context, config, variables, **kw)

    # Sequence Type is "Generated", so the ID is constructed according to the
    # configured split length
    if sequence_type == 'generated':
        number = get_generated_number(context, config, variables, **kw)

    # store the new sequence number to the variables map for str interpolation
    if isinstance(number, Alphanumber):
        variables["alpha"] = number
    variables["seq"] = int(number)

    # The ID formatting template from user config, e.g. {sampleId}-R{seq:02d}
    id_template = config.get("form", "")

    # Interpolate the ID template
    try:
        new_id = id_template.format(**variables)
    except KeyError, e:
        logger.error('KeyError: {} not in id_template {}'.format(
            e, id_template))
        raise
    normalized_id = api.normalize_filename(new_id)
    logger.info("generateUniqueId: {}".format(normalized_id))

    return normalized_id


def renameAfterCreation(obj):
    """Rename the content after it was created/added
    """
    # Check if the _bika_id was already set
    bika_id = getattr(obj, "_bika_id", None)
    if bika_id is not None:
        return bika_id
    # Can't rename without a subtransaction commit when using portal_factory
    transaction.savepoint(optimistic=True)
    # The id returned should be normalized already
    new_id = None
    # Checking if an adapter exists for this content type. If yes, we will
    # get new_id from adapter.
    for name, adapter in getAdapters((obj, ), IIdServer):
        if new_id:
            logger.warn(('More than one ID Generator Adapter found for'
                         'content type -> %s') % obj.portal_type)
        new_id = adapter.generate_id(obj.portal_type)
    if not new_id:
        new_id = generateUniqueId(obj)

    # TODO: This is a naive check just in current folder
    # -> this should check globally for duplicate objects with same prefix
    # N.B. a check like `search_by_prefix` each time would probably slow things
    # down too much!
    # -> A solution could be to store all IDs with a certain prefix in a storage
    parent = api.get_parent(obj)
    if new_id in parent.objectIds():
        # XXX We could do the check in a `while` loop and generate a new one.
        raise KeyError("The ID {} is already taken in the path {}".format(
            new_id, api.get_path(parent)))
    # rename the object to the new id
    parent.manage_renameObject(obj.id, new_id)

    return new_id


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			import re
9
10			import transaction
11			from bika.lims import api
12			from bika.lims import logger
13			from bika.lims.alphanumber import Alphanumber
14			from bika.lims.alphanumber import to_alpha
15			from bika.lims.browser.fields.uidreferencefield import \
16			get_backreferences as get_backuidreferences
17			from bika.lims.interfaces import IIdServer
18			from bika.lims.numbergenerator import INumberGenerator
19			from DateTime import DateTime
20			from Products.ATContentTypes.utils import DT2dt
21			from zope.component import getAdapters
22			from zope.component import getUtility
23
24
25			def get_objects_in_sequence(brain_or_object, ctype, cref):
26			"""Return a list of items
27			"""
28			obj = api.get_object(brain_or_object)
29			if ctype == "backreference":
30			return get_backreferences(obj, cref)
31			if ctype == "contained":
32			return get_contained_items(obj, cref)
33			raise ValueError("Reference value is mandatory for sequence type counter")
34
35
36			def get_backreferences(obj, relationship):
37			"""Returns the backreferences
38			"""
39			refs = get_backuidreferences(obj, relationship)
40
41			# TODO remove after all ReferenceField get ported to UIDReferenceField
42			# At this moment, there are still some content types that are using the
43			# ReferenceField, so we need to fallback to traditional getBackReferences
44			# for these cases.
45			if not refs:
46			refs = obj.getBackReferences(relationship)
47
48			return refs
49
50
51			def get_contained_items(obj, spec):
52			"""Returns a list of (id, subobject) tuples of the current context.
53			If 'spec' is specified, returns only objects whose meta_type match 'spec'
54			"""
55			return obj.objectItems(spec)
56
57
58			def get_config(context, **kw):
59			"""Fetch the config dict from the Bika Setup for the given portal_type
60			"""
61			# get the ID formatting config
62			config_map = api.get_bika_setup().getIDFormatting()
63
64			# allow portal_type override
65			portal_type = kw.get("portal_type") or api.get_portal_type(context)
66
67			# check if we have a config for the given portal_type
68			for config in config_map:
69			if config['portal_type'].lower() == portal_type.lower():
70			return config
71
72			# return a default config
73			default_config = {
74			'form': '%s-{seq}' % portal_type.lower(),
75			'sequence_type': 'generated',
76			'prefix': '%s' % portal_type.lower(),
77			}
78			return default_config
79
80
81			def get_variables(context, **kw):
82			"""Prepares a dictionary of key->value pairs usable for ID formatting
83			"""
84
85			# allow portal_type override
86			portal_type = kw.get("portal_type") or api.get_portal_type(context)
87
88			# The variables map hold the values that might get into the constructed id
89			variables = {
90			'context': context,
91			'id': api.get_id(context),
92			'portal_type': portal_type,
93			'year': get_current_year(),
94			'parent': api.get_parent(context),
95			'seq': 0,
96			'alpha': Alphanumber(0),
97			}
98
99			# Augment the variables map depending on the portal type
100			if portal_type == "AnalysisRequest":
101			now = DateTime()
102			sampling_date = context.getSamplingDate()
103			sampling_date = sampling_date and DT2dt(sampling_date) or DT2dt(now)
104			date_sampled = context.getDateSampled()
105			date_sampled = date_sampled and DT2dt(date_sampled) or DT2dt(now)
106			variables.update({
107			'clientId': context.getClientID(),
108			'dateSampled': date_sampled,
109			'samplingDate': sampling_date,
110			'sampleType': context.getSampleType().getPrefix(),
111			})
112
113			elif portal_type == "ARReport":
114			variables.update({
115			'clientId': context.aq_parent.getClientID(),
116			})
117
118			return variables
119
120
121			def split(string, separator="-"):
122			""" split a string on the given separator
123			"""
124			if not isinstance(string, basestring):
125			return []
126			return string.split(separator)
127
128
129			def to_int(thing, default=0):
130			"""Convert a thing to an integer
131			"""
132			try:
133			return int(thing)
134			except (TypeError, ValueError):
135			return default
136
137
138			def slice(string, separator="-", start=None, end=None):
139			"""Slice out a segment of a string, which is splitted on separator.
140			"""
141
142			# split the given string at the given separator
143			segments = split(string, separator)
144
145			# get the start and endposition for slicing
146			length = len(segments)
147			start = to_int(start)
148			end = to_int(end, length)
149
150			# return the separator joined sliced segments
151			sliced_parts = segments[start:end]
152			return separator.join(sliced_parts)
153
154
155			def get_current_year():
156			"""Returns the current year as a two digit string
157			"""
158			return DateTime().strftime("%Y")[2:]
159
160
161			def search_by_prefix(portal_type, prefix):
162			"""Returns brains which share the same portal_type and ID prefix
163			"""
164			catalog = api.get_tool("uid_catalog")
165			brains = catalog({"portal_type": portal_type})
166			# Filter brains with the same ID prefix
167			return filter(lambda brain: api.get_id(brain).startswith(prefix), brains)
168
169
170			def get_ids_with_prefix(portal_type, prefix):
171			"""Return a list of ids sharing the same portal type and prefix
172			"""
173			brains = search_by_prefix(portal_type, prefix)
174			ids = map(api.get_id, brains)
175			return ids
176
177
178			def make_storage_key(portal_type, prefix=None):
179			"""Make a storage (dict-) key for the number generator
180			"""
181			key = portal_type.lower()
182			if prefix:
183			key = "{}-{}".format(key, prefix)
184			return key
185
186
187			def get_seq_number_from_id(id, id_template, prefix, **kw):
188			"""Return the sequence number of the given ID
189			"""
190			separator = kw.get("separator", "-")
191			postfix = id.replace(prefix, "").strip(separator)
192			postfix_segments = postfix.split(separator)
193			seq_number = 0
194			possible_seq_nums = filter(lambda n: n.isalnum(), postfix_segments)
195			if possible_seq_nums:
196			seq_number = possible_seq_nums[-1]
197
198			# Check if this id has to be expressed as an alphanumeric number
199			seq_number = get_alpha_or_number(seq_number, id_template)
200			seq_number = to_int(seq_number)
201			return seq_number
202
203
204			def get_alpha_or_number(number, template):
205			"""Returns an Alphanumber that represents the number passed in, expressed
206			as defined in the template. Otherwise, returns the number
207			"""
208			match = re.match(r".*\{alpha:(\d+a\d+d)\}$", template.strip())
209			if match and match.groups():
210			format = match.groups()[0]
211			return to_alpha(number, format)
212			return number
213
214
215			def get_counted_number(context, config, variables, **kw):
216			"""Compute the number for the sequence type "Counter"
217			"""
218			# This "context" is defined by the user in Bika Setup and can be actually
219			# anything. However, we assume it is something like "sample" or similar
220			ctx = config.get("context")
221
222			# get object behind the context name (falls back to the current context)
223			obj = variables.get(ctx, context)
224
225			# get the counter type, which is either "backreference" or "contained"
226			counter_type = config.get("counter_type")
227
228			# the counter reference is either the "relationship" for
229			# "backreference" or the meta type for contained objects
230			counter_reference = config.get("counter_reference")
231
232			# This should be a list of existing items, including the current context
233			# object
234			seq_items = get_objects_in_sequence(obj, counter_type, counter_reference)
235
236			number = len(seq_items)
237			return number
238
239
240			def get_generated_number(context, config, variables, **kw):
241			"""Generate a new persistent number with the number generator for the
242			sequence type "Generated"
243			"""
244
245			# separator where to split the ID
246			separator = kw.get('separator', '-')
247
248			# allow portal_type override
249			portal_type = kw.get("portal_type") or api.get_portal_type(context)
250
251			# The ID format for string interpolation, e.g. WS-{seq:03d}
252			id_template = config.get("form", "")
253
254			# The split length defines where the variable part of the ID template begins
255			split_length = config.get("split_length", 1)
256
257			# The prefix tempalte is the static part of the ID
258			prefix_template = slice(id_template, separator=separator, end=split_length)
259
260			# get the number generator
261			number_generator = getUtility(INumberGenerator)
262
263			# generate the key for the number generator storage
264			prefix = prefix_template.format(**variables)
265
266			# normalize out any unicode characters like Ö, É, etc. from the prefix
267			prefix = api.normalize_filename(prefix)
268
269			# The key used for the storage
270			key = make_storage_key(portal_type, prefix)
271
272			# Handle flushed storage
273			if key not in number_generator:
274			max_num = 0
275			existing = get_ids_with_prefix(portal_type, prefix)
276			numbers = map(lambda id: get_seq_number_from_id(id, id_template, prefix), existing)
277			# figure out the highest number in the sequence
278			if numbers:
279			max_num = max(numbers)
280			# set the number generator
281			logger.info("*** SEEDING Prefix '{}' to {}".format(prefix, max_num))
282			number_generator.set_number(key, max_num)
283
284			if not kw.get("dry_run", False):
285			# Generate a new number
286			# NOTE Even when the number exceeds the given ID sequence format,
287			# it will overflow gracefully, e.g.
288			# >>> {sampleId}-R{seq:03d}'.format(sampleId="Water", seq=999999)
289			# 'Water-R999999‘
290			number = number_generator.generate_number(key=key)
291			else:
292			# => This allows us to "preview" the next generated ID in the UI
293			# TODO Show the user the next generated number somewhere in the UI
294			number = number_generator.get(key, 1)
295
296			# Return an int or Alphanumber
297			return get_alpha_or_number(number, id_template)
298
299
300			def generateUniqueId(context, **kw):
301			""" Generate pretty content IDs.
302			"""
303
304			# get the config for this portal type from the system setup
305			config = get_config(context, **kw)
306
307			# get the variables map for later string interpolation
308			variables = get_variables(context, **kw)
309
310			# The new generate sequence number
311			number = 0
312
313			# get the sequence type from the global config
314			sequence_type = config.get("sequence_type", "generated")
315
316			# Sequence Type is "Counter", so we use the length of the backreferences or
317			# contained objects of the evaluated "context" defined in the config
318			if sequence_type == 'counter':
319			number = get_counted_number(context, config, variables, **kw)
320
321			# Sequence Type is "Generated", so the ID is constructed according to the
322			# configured split length
323			if sequence_type == 'generated':
324			number = get_generated_number(context, config, variables, **kw)
325
326			# store the new sequence number to the variables map for str interpolation
327			if isinstance(number, Alphanumber):
328			variables["alpha"] = number
329			variables["seq"] = int(number)
330
331			# The ID formatting template from user config, e.g. {sampleId}-R{seq:02d}
332			id_template = config.get("form", "")
333
334			# Interpolate the ID template
335			try:
336			new_id = id_template.format(**variables)
337			except KeyError, e:
338			logger.error('KeyError: {} not in id_template {}'.format(
339			e, id_template))
340			raise
341			normalized_id = api.normalize_filename(new_id)
342			logger.info("generateUniqueId: {}".format(normalized_id))
343
344			return normalized_id
345
346
347			def renameAfterCreation(obj):
348			"""Rename the content after it was created/added
349			"""
350			# Check if the _bika_id was already set
351			bika_id = getattr(obj, "_bika_id", None)
352			if bika_id is not None:
353			return bika_id
354			# Can't rename without a subtransaction commit when using portal_factory
355			transaction.savepoint(optimistic=True)
356			# The id returned should be normalized already
357			new_id = None
358			# Checking if an adapter exists for this content type. If yes, we will
359			# get new_id from adapter.
360			for name, adapter in getAdapters((obj, ), IIdServer):
361			if new_id:
362			logger.warn(('More than one ID Generator Adapter found for'
363			'content type -> %s') % obj.portal_type)
364			new_id = adapter.generate_id(obj.portal_type)
365			if not new_id:
366			new_id = generateUniqueId(obj)
367
368			# TODO: This is a naive check just in current folder
369			# -> this should check globally for duplicate objects with same prefix
370			# N.B. a check like `search_by_prefix` each time would probably slow things
371			# down too much!
372			# -> A solution could be to store all IDs with a certain prefix in a storage
373			parent = api.get_parent(obj)
374			if new_id in parent.objectIds():
375			# XXX We could do the check in a `while` loop and generate a new one.
376			raise KeyError("The ID {} is already taken in the path {}".format(
377			new_id, api.get_path(parent)))
378			# rename the object to the new id
379			parent.manage_renameObject(obj.id, new_id)
380
381			return new_id
382

senaite / senaite.core

Branch — master (9edffc)

bika.lims.idserver.slice() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like