acris 1.0.4

acris is a python library of programming patterns that we use, at acrisel, in Python projects and choose to contribute to Python community

Overview

acris is a python library providing useful programming patterns.

threaded

decorator for methods that can be executed as a thread.

example

from acris import threaded
from time import sleep

class ThreadedExample(object):
    @threaded
    def proc(self, id_, num, stall):
        s=num
        while num > 0:
            print("%s: %s" % (id_, s))
            num -= 1
            s += stall
            sleep(stall)
        print("%s: %s" % (id_, s))
        return s

class RetVal(object):
    def __init__(self, name):
        self.name=name

    def __call__(self, retval):
        print(self.name, ':', retval)

example output

te1=ThreadedExample().proc(1, 3, 1)
te2=ThreadedExample().proc(2, 3, 5)

te1.addCallback(RetVal('te1'))
te2.addCallback(RetVal('te2'))

will produce:

1: 3
2: 3
1: 4
1: 5
1: 6
te1 : 6
2: 8
2: 13
2: 18
te2 : 18

Singleton

meta class that creates singleton footprint of classes inheriting from it.

example

from acris import Singleton

class Sequence(Singleton):
    step_id=0

    def __call__(self):
        step_id=self.step_id
        self.step_id += 1
        return step_id

example output

A=Sequence()
print('A', A())
print('A', A())
B=Sequence()
print('B', B())

will produce:

A 0
A 1
B 2

Sequence

meta class to produce sequences. Sequence allows creating different sequences using name tags.

example

from acris import Sequence

A=Sequence('A')
print('A', A())
print('A', A())
B=Sequence('B')
print('B', B())

A=Sequence('A')
print('A', A())
print('A', A())
B=Sequence('B')
print('B', B())

example output

A 0
A 1
B 0
A 2
A 3
B 1

TimedSizedRotatingHandler

Use TimedSizedRotatingHandler is combining TimedRotatingFileHandler with RotatingFileHandler. Usage as handler with logging is as defined in Python’s logging how-to

example

import logging

# create logger
logger = logging.getLogger('simple_example')
logger.setLevel(logging.DEBUG)

# create console handler and set level to debug
ch = logging.TimedRotatingFileHandler()
ch.setLevel(logging.DEBUG)

# create formatter
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')

# add formatter to ch
ch.setFormatter(formatter)

# add ch to logger
logger.addHandler(ch)

# 'application' code
logger.debug('debug message')
logger.info('info message')
logger.warn('warn message')
logger.error('error message')
logger.critical('critical message')

MpLogger

Multiprocessor logger using QueueListener and QueueHandler It uses TimedSizedRotatingHandler as its logging handler

example

In main process:

   import logging
   import time

   logger=logging.getLogger(__name__)

   mplogger=MpLogger(logging_level=logging.DEBUG)
   mplogger.start()

   logger.debug("starting sub processes")
   # running processes
   module_logger.debug("joining sub processes")

   mplogger.stop()

Within individual process:

import logging

logger=logging.getLogger(__name__)
module_logger.debug("logging from sub process")