Source Code for Package pypln.workers

 1  # coding: utf-8 
 2   
 3  import sys 
 4  from os.path import dirname, basename 
 5  from glob import glob 
 6  from importlib import import_module 
 7   
 8   
 9  __all__ = ['available', 'wrapper'] 
10  current_dir = dirname(__file__) 
11  required_objects = ['__meta__', 'main'] 
12  required_meta = ['from', 'requires', 'to', 'provides'] 
13   
14  available = {} 
15  sys.path.insert(0, current_dir) 
16  for filename in glob('{}/*.py'.format(current_dir)): 
17      worker = basename(filename[:-3]) 
18      if worker != '__init__': 
19          worker_module = import_module(worker) 
20          worker_objects = dir(worker_module) 
21          for obj in required_objects: 
22              if obj not in worker_objects: 
23                  break 
24          else: 
25              meta_obj = getattr(worker_module, '__meta__') 
26              meta_keys = meta_obj.keys() 
27              for meta in required_meta: 
28                  if meta not in meta_keys: 
29                      break 
30              else: 
31                  __all__.append(worker) 
32                  available[worker] = {'main': getattr(worker_module, 'main'), 
33                                       'from': meta_obj['from'], 
34                                       'requires': meta_obj['requires'], 
35                                       'to': meta_obj['to'], 
36                                       'provides': meta_obj['provides'], 
37                                      } 
38   
39 -def wrapper(child_connection): 
40      #TODO: should receive the document or database's configuration? 
41      #      Note that if a worker should process a big document or an entire 
42      #      corpus, it's better to received database's configuration and pass to 
43      #      worker only an lazy iterator for the collection (pymongo's cursor) 
44      #TODO: create documentation about object type returned by worker (strings 
45      #      must be unicode) 
46      #TODO: add the possibility to create workers that are executables (they 
47      #      receive data as JSON in stdin and put the result as JSON in stdout), 
48      #      so we can create workers in C, Perl, Ruby etc. 
49      #TODO: should get any exception information and send it to broker signaling 
50      #      'job failed' and sending the traceback 
51      worker, document = child_connection.recv() 
52      result = available[worker]['main'](document) 
53      child_connection.send(result) 
54