upstream/kallithea Files · pylons_app/lib/celerylib/__init__.py

Added some more tests, rewrite testing schema, to autogenerate fresh db, new index....

Added some more tests, rewrite testing schema, to autogenerate fresh db, new index. cleaned up some codes that involves testing.

marcink - - Load All Authors

File last commit:

r487:b12ea84f celery


                r491:fefffd6f

celery

Download file

             __init__.py
        
                    29 lines
            
             | 694 B
            
                | text/x-python
            
             |
                PythonLexer
            
             / pylons_app / lib / celerylib / __init__.py
          
                    History
                
                 |
                  Annotation
                 | Raw
                 |Copy content
                 |Copy permalink

      from vcs.utils.lazy import LazyProperty

      import logging

      import os

      import sys

      import traceback

      log = logging.getLogger(__name__)

      class ResultWrapper(object):

          def __init__(self, task):

              self.task = task

          @LazyProperty

          def result(self):

              return self.task

      def run_task(task, *args, **kwargs):

          try:

              t = task.delay(*args, **kwargs)

              log.info('running task %s', t.task_id)

              return t

          except Exception, e:

              if e.errno == 111:

                  log.debug('Unnable to connect. Sync execution')

              else:

                  log.error(traceback.format_exc())

              #pure sync version

              return ResultWrapper(task(*args, **kwargs))

	Site-wide shortcuts
/	Use quick search box
g h	Goto home page
g g	Goto my private gists page
g G	Goto my public gists page
g 0-9	Goto bookmarked items from 0-9
n r	New repository page
n g	New gist page

	Repositories
g s	Goto summary page
g c	Goto changelog page
g f	Goto files page
g F	Goto files page with file search activated
g p	Goto pull requests page
g o	Goto repository settings
g O	Goto repository access permissions settings
t s	Toggle sidebar on some pages

				from vcs.utils.lazy import LazyProperty
				import logging
				import os
				import sys
				import traceback

				log = logging.getLogger(__name__)

				class ResultWrapper(object):
				def __init__(self, task):
				self.task = task

				@LazyProperty
				def result(self):
				return self.task

				def run_task(task, args, *kwargs):
				try:
				t = task.delay(args, *kwargs)
				log.info('running task %s', t.task_id)
				return t
				except Exception, e:
				if e.errno == 111:
				log.debug('Unnable to connect. Sync execution')
				else:
				log.error(traceback.format_exc())
				#pure sync version
				return ResultWrapper(task(args, *kwargs))