root/lang/python/pySchwartz/Schwartz.py @ 18629

#!/usr/bin/python

import sys
import time
import simplejson
from Job import Job
from Worker import Worker

class Schwartz(object):
  conns = []
  funcmap = {}
  workermap = {}

  def __init__(self, databases):
    for database in databases:
      m = __import__(database['driver'], globals(), locals(), [''])
      del database['driver']
      conn = m.connect(**database)
      self.conns.append(conn)
      if not self.funcmap.has_key(str(conn)):
        self.funcmap[str(conn)] = {}

  @staticmethod
  def funcname_to_id(conns, funcmap, funcname):
    funcid = ''
    for conn in conns:
      funcmap = funcmap[str(conn)]
      if funcmap.has_key(funcname):
        funcid = funcmap[funcname]
      if not funcid:
        cursor = None
        try:
          cursor = conn.cursor()
          cursor.execute('select funcid, funcname from funcmap')
          rows = cursor.fetchall()
          for row in rows:
            if row[1] == funcname:
              funcid = row[0]
              funcmap[funcname] = funcid
              break
          if not funcid:
            cursor.execute("insert into funcmap (funcname) values (%s)", funcname)
            cursor.execute('select funcid, funcname from funcmap')
            rows = cursor.fetchall()
            for row in rows:
              if row[1] == funcname:
                funcid = row[0]
                funcmap[funcname] = funcid
                break
          cursor.close()
        except:
          pass
        finally:
          if cursor:
            cursor.close()
      if funcid:
        break
    return funcid

  def can_do(self, funcname, worker):
    if isinstance(worker, Worker):
      if not self.workermap.has_key(funcname):
        self.workermap[funcname] = []
      self.workermap[funcname].append(worker)

  def work(self):
    while True:
      for funcname in self.workermap:
        funcid = Schwartz.funcname_to_id(self.conns, self.funcmap, funcname)
        for worker in self.workermap[funcname]:
          for conn in self.conns:
            cursor = None
            try:
              cursor = conn.cursor()
              cursor.execute('select arg from job where funcid = %s', funcid)
              rows = cursor.fetchall()
              for row in rows:
                worker.work(simplejson.loads(row[0]))
              cursor.execute('delete from job where funcid = %s', funcid)
              cursor.close()
              conn.commit()
            except Exception, e:
              print e
            finally:
              if cursor:
                cursor.close()
              conn.rollback()
      time.sleep(3)
      print "do working..."

  def __del__(self):
    for conn in self.conns:
      try:
        conn.close()
      except:
        pass