@ -8,27 +8,58 @@
""" """
import transport import transport
import os import os
from multiprocessing import Process from multiprocessing import Process , Lock
import numpy as np import numpy as np
import json import json
import pandas as pd
class Subject ( Process ) : class Subject ( Process ) :
cache = pd . DataFrame ( )
lock = Lock ( )
@staticmethod
def log ( _args ) :
Subject . lock . acquire ( )
try :
Subject . cache = Subject . cache . append ( pd . DataFrame ( [ _args ] ) )
except Exception as e :
print ( e )
finally :
Subject . lock . release ( )
def __init__ ( self , * * _args ) :
def __init__ ( self , * * _args ) :
super ( ) . __init__ ( )
super ( ) . __init__ ( )
self . observers = _args [ ' observers ' ]
self . observers = _args [ ' observers ' ]
self . index = 0
self . index = 0
self . name = _args [ ' name ' ]
self . name = _args [ ' name ' ]
self . table = self . observers [ 1 ] . table
self . table = self . observers [ 1 ] . table
self . m = { }
pass
pass
def run ( self ) :
def run ( self ) :
self . notify ( )
self . notify ( )
def notify ( self ) :
def notify ( self ) :
if self . index < len ( self . observers ) :
if self . index < len ( self . observers ) :
observer = self . observers [ self . index ]
observer = self . observers [ self . index ]
_observer = None if self . index == 0 else self . observers [ self . index - 1 ]
_observer = None if self . index == 0 else self . observers [ self . index - 1 ]
_invalues = None if not _observer else _observer . get ( )
_invalues = None if not _observer else _observer . get ( )
if _observer is None :
self . m [ ' table ' ] = self . name
observer . init ( caller = self , invalues = _invalues )
observer . init ( caller = self , invalues = _invalues )
self . index + = 1
self . index + = 1
observer . execute ( )
observer . execute ( )
print ( { " table " : self . table , " module " : observer . name ( ) , " status " : observer . status } )
# self.m[observer.name()] = observer.status
else :
pass
class Worker : class Worker :
def __init__ ( self , * * _args ) :
def __init__ ( self , * * _args ) :
@ -38,6 +69,7 @@ class Worker :
self . logs = [ ]
self . logs = [ ]
self . schema = _args [ ' schema ' ]
self . schema = _args [ ' schema ' ]
self . prefix = _args [ ' prefix ' ]
self . prefix = _args [ ' prefix ' ]
self . status = 0
def name ( self ) :
def name ( self ) :
return self . __class__ . __name__
return self . __class__ . __name__
@ -45,7 +77,7 @@ class Worker :
"""
"""
This function is designed to log to either the console or a data - store
This function is designed to log to either the console or a data - store
"""
"""
print ( _args
# print (_args
pass
pass
def init ( self , * * _args ) :
def init ( self , * * _args ) :
"""
"""
@ -60,9 +92,10 @@ class Worker :
try :
try :
self . _apply ( )
self . _apply ( )
except Exception as error :
except Exception as error :
print ( )
pass
print ( error )
# print ()
print ( )
# print (error)
# print ()
finally :
finally :
self . caller . notify ( )
self . caller . notify ( )
@ -101,10 +134,12 @@ class CreateSQL(Worker) :
writer . apply ( self . _sql . replace ( " :table " , sqltable ) )
writer . apply ( self . _sql . replace ( " :table " , sqltable ) )
writer . close ( )
writer . close ( )
log [ ' status ' ] = 1
log [ ' status ' ] = 1
self . status = 1
except Exception as e :
except Exception as e :
log [ ' status ' ] = 0
log [ ' status ' ] = 0
log [ ' info ' ] = { " error " : e . args [ 0 ] }
log [ ' info ' ] = { " error " : e . args [ 0 ] }
print ( e )
# print (e)
finally :
finally :
self . log ( * * log )
self . log ( * * log )
@ -141,25 +176,28 @@ class Reader(Worker):
self . rows = [ ]
self . rows = [ ]
def _apply ( self ) :
def _apply ( self ) :
self . reader = transport . factory . instance ( * * self . _info ) ;
try :
print ( )
self . reader = transport . factory . instance ( * * self . _info ) ;
print ( self . table )
self . rows = self . reader . read ( mongo = self . pipeline )
print ( json . dumps ( self . pipeline ) )
print ( )
self . rows = self . reader . read ( mongo = self . pipeline )
N = len ( self . rows ) / self . MAX_ROWS if len ( self . rows ) > self . MAX_ROWS else 1
N = len ( self . rows ) / self . MAX_ROWS if len ( self . rows ) > self . MAX_ROWS else 1
N = int ( N )
N = int ( N )
# self.rows = rows
# self.rows = rows
_log = { " context " : self . name ( ) , " args " : self . _info [ ' args ' ] [ ' db ' ] , " status " : 1 , " info " : { " rows " : len ( self . rows ) , " table " : self . table , " segments " : N } }
_log = { " context " : self . name ( ) , " args " : self . _info [ ' args ' ] [ ' db ' ] , " status " : 1 , " info " : { " rows " : len ( self . rows ) , " table " : self . table , " segments " : N } }
self . rows = np . array_split ( self . rows , N )
self . rows = np . array_split ( self . rows , N )
# self.get = lambda : rows #np.array_split(rows,N)
self . reader . close ( )
self . status = 1
#
except Exception as e :
log [ ' status ' ] = 0
log [ ' info ' ] = { " error " : e . args [ 0 ] }
# self.get = lambda : rows #np.array_split(rows,N)
self . reader . close ( )
#
self . log ( * * _log )
self . log ( * * _log )
# @TODO: Call the caller and notify it that this here is done
# @TODO: Call the caller and notify it that this here is done
def get ( self ) :
def get ( self ) :
return self . rows
return self . rows
@ -201,8 +239,8 @@ class Writer(Worker):
# for _e in rows :
# for _e in rows :
# writer.write(_e)
# writer.write(_e)
self . status = 1
else :
else :
print ( " No data was passed " )
print ( " No data was passed " )
