common.tasks.redis_lock() - Code Metrics - Inspection of ":sparkles: Structure ZOOMA calls into InjectTool" - cnr-ibba/IMAGE-InjectTool - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#51)

by Paolo

created 2019-10-10 14:36 UTC

common.tasks.redis_lock() B

↳ Parent: common.tasks

Complexity

Conditions

Size

Total Lines	52
Code Lines	18

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	18
dl	0
loc	52
rs	8.5666
c	0
b	0
f	0
cc	6
nop	3

How to fix Long Method

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on Tue Jan 15 16:42:24 2019

@author: Paolo Cozzi <[email protected]>
"""

import redis
import traceback

from contextlib import contextmanager
from celery.five import monotonic
from celery.utils.log import get_task_logger

from django.conf import settings
from django.core import management

from image.celery import app as celery_app

from .helpers import send_mail_to_admins

# Lock expires in 10 minutes
LOCK_EXPIRE = 60 * 10

# Get an instance of a logger
logger = get_task_logger(__name__)


class BaseTask(celery_app.Task):
    """Base class to celery tasks. Define logs for on_failure and debug_task"""

    name = None
    description = None
    action = None

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        if self.name is None:
            self.name = str(self.__class__)

        if self.action is None:
            self.action = str(self.__class__)

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        logger.error('{0!r} failed: {1!r}'.format(task_id, exc))

    def debug_task(self):
        # this doesn't throw an error when debugging a task called with run()
        if self.request_stack:
            logger.debug('Request: {0!r}'.format(self.request))


class NotifyAdminTaskMixin():
    """A mixin to send error message to admins"""

    action = None

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        """Override the default on_failure method"""

        # call base class
        super().on_failure(exc, task_id, args, kwargs, einfo)

        # get exception info
        einfo = traceback.format_exc()

        subject = "Error in %s" % (self.action)
        body = str(einfo)

        send_mail_to_admins(subject, body)


# https://stackoverflow.com/a/51429597
@celery_app.task(bind=True, base=BaseTask)
def clearsessions(self):
    """Cleanup expired sessions by using Django management command."""

    logger.info("Clearing session with celery...")

    # debugging instance
    self.debug_task()

    # calling management command
    management.call_command("clearsessions", verbosity=1)

    # debug
    logger.info("Sessions cleaned!")

    return "Session cleaned with success"


@contextmanager
def redis_lock(lock_id, blocking=False, expire=True):
    """
    This function get a lock relying on a lock name and other status. You
    can describe more process using the same lock name and give exclusive
    access to one of them.

    Args:
        lock_id (str): the name of the lock to take
        blocking (bool): if True, we wait until we have the block, if False
            we returns immediately False
        expire (bool): if True, lock will expire after LOCK_EXPIRE timeout,
            if False, it will persist until lock is released

    Returns:
        bool: True if lock acquired, False otherwise
    """

    # read parameters from settings
    REDIS_CLIENT = redis.StrictRedis(
        host=settings.REDIS_HOST,
        port=settings.REDIS_PORT,
        db=settings.REDIS_DB)

    # this will be the redis lock
    lock = None

    # timeout for the lock (if expire condition)
    timeout_at = monotonic() + LOCK_EXPIRE - 3

    if expire:
        lock = REDIS_CLIENT.lock(lock_id, timeout=LOCK_EXPIRE)

    else:
        lock = REDIS_CLIENT.lock(lock_id, timeout=None)

    status = lock.acquire(blocking=blocking)

    try:
        logger.debug("lock %s acquired is: %s" % (lock_id, status))
        yield status

    finally:
        # we take advantage of using add() for atomic locking
        # don't release the lock if we didn't acquire it
        if status and ((monotonic() < timeout_at and expire) or not expire):
            logger.debug("Releasing lock %s" % lock_id)
            # don't release the lock if we exceeded the timeout
            # to lessen the chance of releasing an expired lock
            # owned by someone else
            # if no timeout and lock is taken, release it
            lock.release()


class exclusive_task(object):
    """A class decorator to execute an exclusive task by decorating
    celery.tasks.Task.run (run this task once, others
    task calls will return already running message without calling task or
    will wait until other tasks of this type are completed)

    Args:
        task_name (str): task name used for debug
        lock_id (str): the task lock id
        blocking (bool): set task as blocking (wait until no other tasks
            are running. def. False)
        lock_expire (bool): define if lock will expire or not after a
            certain time (def. False)
    """

    def __init__(self, task_name, lock_id, blocking=False, block_expire=False):

        """
        If there are decorator arguments, the function
        to be decorated is not passed to the constructor!
        """
        logger.debug("Setting up ExclusiveTaskDecorator")

        self.task_name = task_name
        self.lock_id = lock_id
        self.blocking = blocking
        self.block_expire = block_expire

    def __call__(self, f):
        """
        If there are decorator arguments, __call__() is only called
        once, as part of the decoration process! You can only give
        it a single argument, which is the function object.
        """
        logger.debug("Decorating function")

        def wrapped_f(*args, **kwargs):
            with redis_lock(
                    self.lock_id,
                    self.blocking,
                    self.block_expire) as acquired:
                if acquired:
                    logger.debug("lock %s aquired")

                    # do stuff and return something
                    result = f(*args, **kwargs)

                    logger.debug("lock %s released")

                else:
                    # warn user and return a default message
                    result = "%s already running!" % (self.task_name)
                    logger.warning(result)

            return result

        return wrapped_f


1			#!/usr/bin/env python3
2			# -- coding: utf-8 --
3			"""
4			Created on Tue Jan 15 16:42:24 2019
5
6			@author: Paolo Cozzi <[email protected]>
7			"""
8
9			import redis
10			import traceback
11
12			from contextlib import contextmanager
13			from celery.five import monotonic
14			from celery.utils.log import get_task_logger
15
16			from django.conf import settings
17			from django.core import management
18
19			from image.celery import app as celery_app
20
21			from .helpers import send_mail_to_admins
22
23			# Lock expires in 10 minutes
24			LOCK_EXPIRE = 60 * 10
25
26			# Get an instance of a logger
27			logger = get_task_logger(__name__)
28
29
30			class BaseTask(celery_app.Task):
31			"""Base class to celery tasks. Define logs for on_failure and debug_task"""
32
33			name = None
34			description = None
35			action = None
36
37			def __init__(self, args, *kwargs):
38			super().__init__(args, *kwargs)
39
40			if self.name is None:
41			self.name = str(self.__class__)
42
43			if self.action is None:
44			self.action = str(self.__class__)
45
46			def on_failure(self, exc, task_id, args, kwargs, einfo):
47			logger.error('{0!r} failed: {1!r}'.format(task_id, exc))
48
49			def debug_task(self):
50			# this doesn't throw an error when debugging a task called with run()
51			if self.request_stack:
52			logger.debug('Request: {0!r}'.format(self.request))
53
54
55			class NotifyAdminTaskMixin():
56			"""A mixin to send error message to admins"""
57
58			action = None
59
60			def on_failure(self, exc, task_id, args, kwargs, einfo):
61			"""Override the default on_failure method"""
62
63			# call base class
64			super().on_failure(exc, task_id, args, kwargs, einfo)
65
66			# get exception info
67			einfo = traceback.format_exc()
68
69			subject = "Error in %s" % (self.action)
70			body = str(einfo)
71
72			send_mail_to_admins(subject, body)
73
74
75			# https://stackoverflow.com/a/51429597
76			@celery_app.task(bind=True, base=BaseTask)
77			def clearsessions(self):
78			"""Cleanup expired sessions by using Django management command."""
79
80			logger.info("Clearing session with celery...")
81
82			# debugging instance
83			self.debug_task()
84
85			# calling management command
86			management.call_command("clearsessions", verbosity=1)
87
88			# debug
89			logger.info("Sessions cleaned!")
90
91			return "Session cleaned with success"
92
93
94			@contextmanager
95			def redis_lock(lock_id, blocking=False, expire=True):
96			"""
97			This function get a lock relying on a lock name and other status. You
98			can describe more process using the same lock name and give exclusive
99			access to one of them.
100
101			Args:
102			lock_id (str): the name of the lock to take
103			blocking (bool): if True, we wait until we have the block, if False
104			we returns immediately False
105			expire (bool): if True, lock will expire after LOCK_EXPIRE timeout,
106			if False, it will persist until lock is released
107
108			Returns:
109			bool: True if lock acquired, False otherwise
110			"""
111
112			# read parameters from settings
113			REDIS_CLIENT = redis.StrictRedis(
114			host=settings.REDIS_HOST,
115			port=settings.REDIS_PORT,
116			db=settings.REDIS_DB)
117
118			# this will be the redis lock
119			lock = None
120
121			# timeout for the lock (if expire condition)
122			timeout_at = monotonic() + LOCK_EXPIRE - 3
123
124			if expire:
125			lock = REDIS_CLIENT.lock(lock_id, timeout=LOCK_EXPIRE)
126
127			else:
128			lock = REDIS_CLIENT.lock(lock_id, timeout=None)
129
130			status = lock.acquire(blocking=blocking)
131
132			try:
133			logger.debug("lock %s acquired is: %s" % (lock_id, status))
134			yield status
135
136			finally:
137			# we take advantage of using add() for atomic locking
138			# don't release the lock if we didn't acquire it
139			if status and ((monotonic() < timeout_at and expire) or not expire):
140			logger.debug("Releasing lock %s" % lock_id)
141			# don't release the lock if we exceeded the timeout
142			# to lessen the chance of releasing an expired lock
143			# owned by someone else
144			# if no timeout and lock is taken, release it
145			lock.release()
146
147
148			class exclusive_task(object):
149			"""A class decorator to execute an exclusive task by decorating
150			celery.tasks.Task.run (run this task once, others
151			task calls will return already running message without calling task or
152			will wait until other tasks of this type are completed)
153
154			Args:
155			task_name (str): task name used for debug
156			lock_id (str): the task lock id
157			blocking (bool): set task as blocking (wait until no other tasks
158			are running. def. False)
159			lock_expire (bool): define if lock will expire or not after a
160			certain time (def. False)
161			"""
162
163			def __init__(self, task_name, lock_id, blocking=False, block_expire=False):
164
165			"""
166			If there are decorator arguments, the function
167			to be decorated is not passed to the constructor!
168			"""
169			logger.debug("Setting up ExclusiveTaskDecorator")
170
171			self.task_name = task_name
172			self.lock_id = lock_id
173			self.blocking = blocking
174			self.block_expire = block_expire
175
176			def __call__(self, f):
177			"""
178			If there are decorator arguments, __call__() is only called
179			once, as part of the decoration process! You can only give
180			it a single argument, which is the function object.
181			"""
182			logger.debug("Decorating function")
183
184			def wrapped_f(args, *kwargs):
185			with redis_lock(
186			self.lock_id,
187			self.blocking,
188			self.block_expire) as acquired:
189			if acquired:
190			logger.debug("lock %s aquired")
191
192			# do stuff and return something
193			result = f(args, *kwargs)
194
195			logger.debug("lock %s released")
196
197			else:
198			# warn user and return a default message
199			result = "%s already running!" % (self.task_name)
200			logger.warning(result)
201
202			return result
203
204			return wrapped_f
205

cnr-ibba / IMAGE-InjectTool

Pull Request — master (#51)

common.tasks.redis_lock() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like