senaite.core.decorators.retriable() - Code Metrics - Inspection of "Added SENAITE maintenance scripts" - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — 2.x (#1933)

by Ramon

created 2022-02-09 20:21 UTC

senaite.core.decorators.retriable() C

↳ Parent: senaite.core.decorators

Complexity

Conditions

Size

Total Lines	29
Code Lines	26

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	26
dl	0
loc	29
rs	6.6666
c	0
b	0
f	0
cc	9
nop	4

# -*- coding: utf-8 -*-

from functools import wraps

import transaction
from bika.lims import api
from senaite.core import logger
from ZODB.POSException import ConflictError


def readonly_transaction(func):
    """Decorator to doom the current transaction

    https://transaction.readthedocs.io/en/latest/doom.html#dooming-transactions
    """
    @wraps(func)
    def decorator(self, *args, **kwargs):
        logger.info("*** READONLY TRANSACTION: '{}.{}' ***".
                    format(self.__class__.__name__, self.__name__))
        tx = transaction.get()
        tx.doom()
        return func(self, *args, **kwargs)
    return decorator


def retriable(count=3, sync=False, reraise=True, on_retry_exhausted=None):
    """Retries DB commits
    """

    def decorator(func):
        def wrapped(*args, **kwargs):
            retried = 0
            if sync:
                try:
                    api.get_portal()._p_jar.sync()
                except Exception:
                    pass
            while retried < count:
                try:
                    return func(*args, **kwargs)
                except ConflictError:
                    retried += 1
                    logger.warn("DB ConflictError: Retrying %s/%s"
                                % (retried, count))
                    try:
                        api.get_portal()._p_jar.sync()
                    except Exception:
                        if retried >= count:
                            if on_retry_exhausted is not None:
                                on_retry_exhausted(*args, **kwargs)
                            if reraise:
                                raise
        return wrapped
    return decorator


1			# -- coding: utf-8 --
2
3			from functools import wraps
4
5			import transaction
6			from bika.lims import api
7			from senaite.core import logger
8			from ZODB.POSException import ConflictError
9
10
11			def readonly_transaction(func):
12			"""Decorator to doom the current transaction
13
14			https://transaction.readthedocs.io/en/latest/doom.html#dooming-transactions
15			"""
16			@wraps(func)
17			def decorator(self, args, *kwargs):
18			logger.info("* READONLY TRANSACTION: '{}.{}' *".
19			format(self.__class__.__name__, self.__name__))
20			tx = transaction.get()
21			tx.doom()
22			return func(self, args, *kwargs)
23			return decorator
24
25
26			def retriable(count=3, sync=False, reraise=True, on_retry_exhausted=None):
27			"""Retries DB commits
28			"""
29
30			def decorator(func):
31			def wrapped(args, *kwargs):
32			retried = 0
33			if sync:
34			try:
35			api.get_portal()._p_jar.sync()
36			except Exception:
37			pass
38			while retried < count:
39			try:
40			return func(args, *kwargs)
41			except ConflictError:
42			retried += 1
43			logger.warn("DB ConflictError: Retrying %s/%s"
44			% (retried, count))
45			try:
46			api.get_portal()._p_jar.sync()
47			except Exception:
48			if retried >= count:
49			if on_retry_exhausted is not None:
50			on_retry_exhausted(args, *kwargs)
51			if reraise:
52			raise
53			return wrapped
54			return decorator
55

senaite / senaite.core

Pull Request — 2.x (#1933)

senaite.core.decorators.retriable() C

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like