Metadata-Version: 2.1
Name: mp-manager
Version: 0.1
Summary: Simple python multiprocessing manager
Home-page: http://pypi.python.org/pypi/mp_manager
Author: Dmitriy Pomazunovskiy
Author-email: forestwheel@gmail.com
License: BSD
Platform: UNKNOWN
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: BSD License
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3.7
Classifier: Topic :: Software Development

======================================
Simple multiprocessing management tool
======================================

Handles big amounts of data using python multiprocessing

=========
Concept
=========

The main idea is to utilize multiprocessing, by building source queue and handle items to process in workers.

Multiprocess managing is implemented using `mp_manager.manager.py` with help of `mp_manager.process_helpers.py` that provides events, locks, shared values and stats, exception handler helper and `ExitStatus`

Workers for multiprocessing are built based on handlers classes.

========
Handlers
========

Handlers must inherit `mp_manager.base_handler.BaseHandler` and at least implement:

1) classmethod `build_items()` that must return iterable of items to handle by workers

2) `_handle_item()` which is used to handle current item

They also could implement:

1) `log_item()` that is used by manager as a source for logging

2) `_reset_stats()` is used internally to reset worker dict stats related to current item

============
Sample usage
============

::

  import math

  import random

  import time

  from mp_manager.base_handler import BaseHandler

  from mp_manager.process_helpers import Stats, ns_container

  from mp_manager.utils import build_batches


  class SampleHandler(BaseHandler):

    batch_size = 2

    @classmethod

    def build_items(cls, _args, __kwargs):

      total = 100

      batches_total = int(math.ceil(total / cls.batch_size))

      Stats.set_stats_val('items_total', total)

      Stats.set_stats_val('batches_total', batches_total)

      ns_container.set_shared_value('shared_data', cls._data_to_share(),

                      True)

      for i, val in enumerate(build_batches(range(total), cls.batch_size)):

        yield i, val

    @property

    def log_item(self):

      return f'{self.current_item[0]} / {Stats.stats.batches_total}'

    @classmethod

    def _data_to_share(cls):

      return dict(somekey="some value")

    def _reset_stats(self):

      self.stats = {

        'items_processed': 0,

        'items_positive': 0,

        'items_negative': 0,

      }

    def _handle_item(self):

      num, items = self.current_item

      for item in items:

        self._process_number(item)

      Stats.inc_stats_vals(self.stats)

      self.logger.info(Stats.get_stats())

    def _process_number(self, item):

      time.sleep(random.random())

      value = math.factorial(item) / math.cos(item)

      self.stats['items_processed'] += 1

      if value > 0:

        self.stats['items_positive'] += 1

      else:

        self.stats['items_negative'] += 1


  if __name__ == '__main__':

    import logging

    from mp_manager.manager import ImportManager

    logging.basicConfig(level=logging.DEBUG)

    status = ImportManager(SampleHandler, num_workers=8).run()

    print(status)

    exit(status.code)

