transport.py 18 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684
  1. """
  2. This file implements data transport stuctures in order to allow data to be moved to and from anywhere
  3. We can thus read data from disk and write to the cloud,queue, or couchdb or SQL
  4. """
  5. from flask import request, session
  6. import os
  7. import pika
  8. import json
  9. import numpy as np
  10. from couchdbkit import Server
  11. import re
  12. from csv import reader
  13. from datetime import datetime
  14. """
  15. @TODO: Write a process by which the class automatically handles reading and creating a preliminary sample and discovers the meta data
  16. """
  17. class Reader:
  18. def __init__(self):
  19. self.nrows = 0
  20. self.xchar = None
  21. def row_count(self):
  22. content = self.read()
  23. return np.sum([1 for row in content])
  24. """
  25. This function determines the most common delimiter from a subset of possible delimiters. It uses a statistical approach to guage the distribution of columns for a given delimiter
  26. """
  27. def delimiter(self,sample):
  28. m = {',':[],'\t':[],'|':[],'\x3A':[]}
  29. delim = m.keys()
  30. for row in sample:
  31. for xchar in delim:
  32. if row.split(xchar) > 1:
  33. m[xchar].append(len(row.split(xchar)))
  34. else:
  35. m[xchar].append(0)
  36. #
  37. # The delimiter with the smallest variance, provided the mean is greater than 1
  38. # This would be troublesome if there many broken records sampled
  39. #
  40. m = {id: np.var(m[id]) for id in m.keys() if m[id] != [] and int(np.mean(m[id]))>1}
  41. index = m.values().index( min(m.values()))
  42. xchar = m.keys()[index]
  43. return xchar
  44. """
  45. This function determines the number of columns of a given sample
  46. @pre self.xchar is not None
  47. """
  48. def col_count(self,sample):
  49. m = {}
  50. i = 0
  51. for row in sample:
  52. row = self.format(row)
  53. id = str(len(row))
  54. #id = str(len(row.split(self.xchar)))
  55. if id not in m:
  56. m[id] = 0
  57. m[id] = m[id] + 1
  58. index = m.values().index( max(m.values()) )
  59. ncols = int(m.keys()[index])
  60. return ncols;
  61. """
  62. This function will clean records of a given row by removing non-ascii characters
  63. @pre self.xchar is not None
  64. """
  65. def format (self,row):
  66. if isinstance(row,list) == False:
  67. #
  68. # We've observed sometimes fields contain delimiter as a legitimate character, we need to be able to account for this and not tamper with the field values (unless necessary)
  69. cols = self.split(row)
  70. #cols = row.split(self.xchar)
  71. else:
  72. cols = row ;
  73. return [ re.sub('[^\x00-\x7F,\n,\r,\v,\b,]',' ',col.strip()).strip().replace('"','') for col in cols]
  74. #if isinstance(row,list) == False:
  75. # return (self.xchar.join(r)).format('utf-8')
  76. #else:
  77. # return r
  78. """
  79. This function performs a split of a record and tries to attempt to preserve the integrity of the data within i.e accounting for the double quotes.
  80. @pre : self.xchar is not None
  81. """
  82. def split (self,row):
  83. pattern = "".join(["(?:^|",self.xchar,")(\"(?:[^\"]+|\"\")*\"|[^",self.xchar,"]*)"])
  84. return re.findall(pattern,row.replace('\n',''))
  85. class Writer:
  86. def format(self,row,xchar):
  87. if xchar is not None and isinstance(row,list):
  88. return xchar.join(row)+'\n'
  89. elif xchar is None and isinstance(row,dict):
  90. row = json.dumps(row)
  91. return row
  92. """
  93. It is important to be able to archive data so as to insure that growth is controlled
  94. Nothing in nature grows indefinitely neither should data being handled.
  95. """
  96. def archive(self):
  97. pass
  98. def flush(self):
  99. pass
  100. """
  101. This class is designed to read data from an Http request file handler provided to us by flask
  102. The file will be heald in memory and processed accordingly
  103. NOTE: This is inefficient and can crash a micro-instance (becareful)
  104. """
  105. class HttpRequestReader(Reader):
  106. def __init__(self,**params):
  107. self.file_length = 0
  108. try:
  109. #self.file = params['file']
  110. #self.file.seek(0, os.SEEK_END)
  111. #self.file_length = self.file.tell()
  112. #print 'size of file ',self.file_length
  113. self.content = params['file'].readlines()
  114. self.file_length = len(self.content)
  115. except Exception, e:
  116. print "Error ... ",e
  117. pass
  118. def isready(self):
  119. return self.file_length > 0
  120. def read(self,size =-1):
  121. i = 1
  122. for row in self.content:
  123. i += 1
  124. if size == i:
  125. break
  126. yield row
  127. """
  128. This class is designed to write data to a session/cookie
  129. """
  130. class HttpSessionWriter(Writer):
  131. """
  132. @param key required session key
  133. """
  134. def __init__(self,**params):
  135. self.session = params['queue']
  136. self.session['sql'] = []
  137. self.session['csv'] = []
  138. self.tablename = re.sub('..+$','',params['filename'])
  139. self.session['uid'] = params['uid']
  140. #self.xchar = params['xchar']
  141. def format_sql(self,row):
  142. values = "','".join([col.replace('"','').replace("'",'') for col in row])
  143. return "".join(["INSERT INTO :table VALUES('",values,"');\n"]).replace(':table',self.tablename)
  144. def isready(self):
  145. return True
  146. def write(self,**params):
  147. label = params['label']
  148. row = params ['row']
  149. if label == 'usable':
  150. self.session['csv'].append(self.format(row,','))
  151. self.session['sql'].append(self.format_sql(row))
  152. """
  153. This class is designed to read data from disk (location on hard drive)
  154. @pre : isready() == True
  155. """
  156. class DiskReader(Reader) :
  157. """
  158. @param path absolute path of the file to be read
  159. """
  160. def __init__(self,**params):
  161. Reader.__init__(self)
  162. self.path = params['path'] ;
  163. def isready(self):
  164. return os.path.exists(self.path)
  165. """
  166. This function reads the rows from a designated location on disk
  167. @param size number of rows to be read, -1 suggests all rows
  168. """
  169. def read(self,size=-1):
  170. f = open(self.path,'rU')
  171. i = 1
  172. for row in f:
  173. i += 1
  174. if size == i:
  175. break
  176. yield row
  177. f.close()
  178. """
  179. This function writes output to disk in a designated location
  180. """
  181. class DiskWriter(Writer):
  182. def __init__(self,**params):
  183. if 'path' in params:
  184. self.path = params['path']
  185. else:
  186. self.path = None
  187. if 'name' in params:
  188. self.name = params['name'];
  189. else:
  190. self.name = None
  191. if os.path.exists(self.path) == False:
  192. os.mkdir(self.path)
  193. """
  194. This function determines if the class is ready for execution or not
  195. i.e it determines if the preconditions of met prior execution
  196. """
  197. def isready(self):
  198. p = self.path is not None and os.path.exists(self.path)
  199. q = self.name is not None
  200. return p and q
  201. """
  202. This function writes a record to a designated file
  203. @param label <passed|broken|fixed|stats>
  204. @param row row to be written
  205. """
  206. def write(self,**params):
  207. label = params['label']
  208. row = params['row']
  209. xchar = None
  210. if 'xchar' is not None:
  211. xchar = params['xchar']
  212. path = ''.join([self.path,os.sep,label])
  213. if os.path.exists(path) == False:
  214. os.mkdir(path) ;
  215. path = ''.join([path,os.sep,self.name])
  216. f = open(path,'a')
  217. row = self.format(row,xchar);
  218. f.write(row)
  219. f.close()
  220. """
  221. This class hierarchy is designed to handle interactions with a queue server using pika framework (our tests are based on rabbitmq)
  222. """
  223. class MessageQueue:
  224. def __init__(self,**params):
  225. self.host= params['host']
  226. self.uid = params['uid']
  227. self.qid = params['qid']
  228. def isready(self):
  229. #self.init()
  230. resp = self.connection is not None and self.connection.is_open
  231. self.close()
  232. return resp
  233. def close(self):
  234. if self.connection.is_closed == False :
  235. self.channel.close()
  236. self.connection.close()
  237. """
  238. This class is designed to publish content to an AMQP (Rabbitmq)
  239. The class will rely on pika to implement this functionality
  240. We will publish information to a given queue for a given exchange
  241. """
  242. class QueueWriter(MessageQueue,Writer):
  243. def __init__(self,**params):
  244. #self.host= params['host']
  245. #self.uid = params['uid']
  246. #self.qid = params['queue']
  247. MessageQueue.__init__(self,**params);
  248. def init(self,label=None):
  249. properties = pika.ConnectionParameters(host=self.host)
  250. self.connection = pika.BlockingConnection(properties)
  251. self.channel = self.connection.channel()
  252. self.info = self.channel.exchange_declare(exchange=self.uid,type='direct',durable=True)
  253. if label is None:
  254. self.qhandler = self.channel.queue_declare(queue=self.qid,durable=True)
  255. else:
  256. self.qhandler = self.channel.queue_declare(queue=label,durable=True)
  257. self.channel.queue_bind(exchange=self.uid,queue=self.qhandler.method.queue)
  258. """
  259. This function writes a stream of data to the a given queue
  260. @param object object to be written (will be converted to JSON)
  261. @TODO: make this less chatty
  262. """
  263. def write(self,**params):
  264. xchar = None
  265. if 'xchar' in params:
  266. xchar = params['xchar']
  267. object = self.format(params['row'],xchar)
  268. label = params['label']
  269. self.init(label)
  270. _mode = 2
  271. if isinstance(object,str):
  272. stream = object
  273. _type = 'text/plain'
  274. else:
  275. stream = json.dumps(object)
  276. if 'type' in params :
  277. _type = params['type']
  278. else:
  279. _type = 'application/json'
  280. self.channel.basic_publish(
  281. exchange=self.uid,
  282. routing_key=label,
  283. body=stream,
  284. properties=pika.BasicProperties(content_type=_type,delivery_mode=_mode)
  285. );
  286. self.close()
  287. def flush(self,label):
  288. self.init(label)
  289. _mode = 1 #-- Non persistent
  290. self.channel.queue_delete( queue=label);
  291. self.close()
  292. """
  293. This class will read from a queue provided an exchange, queue and host
  294. @TODO: Account for security and virtualhosts
  295. """
  296. class QueueReader(MessageQueue,Reader):
  297. """
  298. @param host host
  299. @param uid exchange identifier
  300. @param qid queue identifier
  301. """
  302. def __init__(self,**params):
  303. #self.host= params['host']
  304. #self.uid = params['uid']
  305. #self.qid = params['qid']
  306. MessageQueue.__init__(self,**params);
  307. if 'durable' in params :
  308. self.durable = True
  309. else:
  310. self.durable = False
  311. self.size = -1
  312. self.data = {}
  313. def init(self,qid):
  314. properties = pika.ConnectionParameters(host=self.host)
  315. self.connection = pika.BlockingConnection(properties)
  316. self.channel = self.connection.channel()
  317. self.channel.exchange_declare(exchange=self.uid,type='direct',durable=True)
  318. self.info = self.channel.queue_declare(queue=qid,durable=True)
  319. """
  320. This is the callback function designed to process the data stream from the queue
  321. """
  322. def callback(self,channel,method,header,stream):
  323. r = []
  324. if re.match("^\{|\[",stream) is not None:
  325. r = json.loads(stream)
  326. else:
  327. r = stream
  328. qid = self.info.method.queue
  329. if qid not in self.data :
  330. self.data[qid] = []
  331. self.data[qid].append(r)
  332. #
  333. # We stop reading when the all the messages of the queue are staked
  334. #
  335. if self.size == len(self.data[qid]) or len(self.data[qid]) == self.info.method.message_count:
  336. self.close()
  337. """
  338. This function will read, the first message from a queue
  339. @TODO:
  340. Implement channel.basic_get in order to retrieve a single message at a time
  341. Have the number of messages retrieved be specified by size (parameter)
  342. """
  343. def read(self,size=-1):
  344. r = {}
  345. self.size = size
  346. #
  347. # We enabled the reader to be able to read from several queues (sequentially for now)
  348. # The qid parameter will be an array of queues the reader will be reading from
  349. #
  350. if isinstance(self.qid,basestring) :
  351. self.qid = [self.qid]
  352. for qid in self.qid:
  353. self.init(qid)
  354. # r[qid] = []
  355. if self.info.method.message_count > 0:
  356. self.channel.basic_consume(self.callback,queue=qid,no_ack=False);
  357. self.channel.start_consuming()
  358. else:
  359. pass
  360. #self.close()
  361. # r[qid].append( self.data)
  362. return self.data
  363. class QueueListener(QueueReader):
  364. def init(self,qid):
  365. properties = pika.ConnectionParameters(host=self.host)
  366. self.connection = pika.BlockingConnection(properties)
  367. self.channel = self.connection.channel()
  368. self.channel.exchange_declare(exchange=self.uid,type='direct',durable=True )
  369. self.info = self.channel.queue_declare(passive=True,exclusive=True,queue=qid)
  370. self.channel.queue_bind(exchange=self.uid,queue=self.info.method.queue,routing_key=qid)
  371. #self.callback = callback
  372. def read(self):
  373. self.init(self.qid)
  374. self.channel.basic_consume(self.callback,queue=self.qid,no_ack=True);
  375. self.channel.start_consuming()
  376. """
  377. This class is designed to write output as sql insert statements
  378. The class will inherit from DiskWriter with minor adjustments
  379. @TODO: Include script to create the table if need be using the upper bound of a learner
  380. """
  381. class SQLDiskWriter(DiskWriter):
  382. def __init__(self,**args):
  383. DiskWriter.__init__(self,**args)
  384. self.tablename = re.sub('\..+$','',self.name).replace(' ','_')
  385. """
  386. @param label
  387. @param row
  388. @param xchar
  389. """
  390. def write(self,**args):
  391. label = args['label']
  392. row = args['row']
  393. if label == 'usable':
  394. values = "','".join([col.replace('"','').replace("'",'') for col in row])
  395. row = "".join(["INSERT INTO :table VALUES('",values,"');\n"]).replace(':table',self.tablename)
  396. args['row'] = row
  397. DiskWriter.write(self,**args)
  398. class Couchdb:
  399. """
  400. @param uri host & port reference
  401. @param uid user id involved
  402. @param dbname database name (target)
  403. """
  404. def __init__(self,**args):
  405. uri = args['uri']
  406. self.uid = args['uid']
  407. dbname = args['dbname']
  408. self.server = Server(uri=uri)
  409. self.dbase = self.server.get_db(dbname)
  410. if self.dbase.doc_exist(self.uid) == False:
  411. self.dbase.save_doc({"_id":self.uid})
  412. """
  413. Insuring the preconditions are met for processing
  414. """
  415. def isready(self):
  416. p = self.server.info() != {}
  417. if p == False or self.dbase.dbname not in self.server.all_dbs():
  418. return False
  419. #
  420. # At this point we are sure that the server is connected
  421. # We are also sure that the database actually exists
  422. #
  423. q = self.dbase.doc_exist(self.uid)
  424. if q == False:
  425. return False
  426. return True
  427. def view(self,id,**args):
  428. r =self.dbase.view(id,**args)
  429. r = r.all()
  430. return r[0]['value'] if len(r) > 0 else []
  431. """
  432. This function will read an attachment from couchdb and return it to calling code. The attachment must have been placed before hand (otherwise oops)
  433. @T: Account for security & access control
  434. """
  435. class CouchdbReader(Couchdb,Reader):
  436. """
  437. @param filename filename (attachment)
  438. """
  439. def __init__(self,**args):
  440. #
  441. # setting the basic parameters for
  442. Couchdb.__init__(self,**args)
  443. if 'filename' in args :
  444. self.filename = args['filename']
  445. else:
  446. self.filename = None
  447. def isready(self):
  448. #
  449. # Is the basic information about the database valid
  450. #
  451. p = Couchdb.isready(self)
  452. if p == False:
  453. return False
  454. #
  455. # The database name is set and correct at this point
  456. # We insure the document of the given user has the requested attachment.
  457. #
  458. doc = self.dbase.get(self.uid)
  459. if '_attachments' in doc:
  460. r = self.filename in doc['_attachments'].keys()
  461. else:
  462. r = False
  463. return r
  464. def stream(self):
  465. content = self.dbase.fetch_attachment(self.uid,self.filename).split('\n') ;
  466. i = 1
  467. for row in content:
  468. yield row
  469. if size > 0 and i == size:
  470. break
  471. i = i + 1
  472. def read(self,size=-1):
  473. if self.filename is not None:
  474. self.stream()
  475. else:
  476. return self.basic_read()
  477. def basic_read(self):
  478. document = self.dbase.get(self.uid)
  479. del document['_id'], document['_rev']
  480. return document
  481. """
  482. This class will write on a couchdb document provided a scope
  483. The scope is the attribute that will be on the couchdb document
  484. """
  485. class CouchdbWriter(Couchdb,Writer):
  486. """
  487. @param uri host & port reference
  488. @param uid user id involved
  489. @param filename filename (attachment)
  490. @param dbname database name (target)
  491. """
  492. def __init__(self,**args):
  493. uri = args['uri']
  494. self.uid = args['uid']
  495. if 'filename' in args:
  496. self.filename = args['filename']
  497. else:
  498. self.filename = None
  499. dbname = args['dbname']
  500. self.server = Server(uri=uri)
  501. self.dbase = self.server.get_db(dbname)
  502. #
  503. # If the document doesn't exist then we should create it
  504. #
  505. """
  506. write a given attribute to a document database
  507. @param label scope of the row repair|broken|fixed|stats
  508. @param row row to be written
  509. """
  510. def write(self,**params):
  511. document = self.dbase.get(self.uid)
  512. label = params['label']
  513. row = params['row']
  514. if label not in document :
  515. document[label] = []
  516. document[label].append(row)
  517. self.dbase.save_doc(document)
  518. def flush(self,**params) :
  519. size = params['size'] if 'size' in params else 0
  520. has_changed = False
  521. document = self.dbase.get(self.uid)
  522. for key in document:
  523. if key not in ['_id','_rev','_attachments'] :
  524. content = document[key]
  525. else:
  526. continue
  527. if isinstance(content,list) and size > 0:
  528. index = len(content) - size
  529. content = content[index:]
  530. document[key] = content
  531. else:
  532. document[key] = {}
  533. has_changed = True
  534. self.dbase.save_doc(document)
  535. def archive(self,params=None):
  536. document = self.dbase.get(self.uid)
  537. content = {}
  538. _doc = {}
  539. for id in document:
  540. if id in ['_id','_rev','_attachments'] :
  541. _doc[id] = document[id]
  542. else:
  543. content[id] = document[id]
  544. content = json.dumps(content)
  545. document= _doc
  546. now = str(datetime.today())
  547. name = '-'.join([document['_id'] , now,'.json'])
  548. self.dbase.save_doc(document)
  549. self.dbase.put_attachment(document,content,name,'application/json')
  550. """
  551. This class acts as a factory to be able to generate an instance of a Reader/Writer
  552. Against a Queue,Disk,Cloud,Couchdb
  553. The class doesn't enforce parameter validation, thus any error with the parameters sent will result in a null Object
  554. """
  555. class DataSourceFactory:
  556. def instance(self,**args):
  557. source = args['type']
  558. params = args['args']
  559. anObject = None
  560. if source in ['HttpRequestReader','HttpSessionWriter']:
  561. #
  562. # @TODO: Make sure objects are serializable, be smart about them !!
  563. #
  564. aClassName = ''.join([source,'(**params)'])
  565. else:
  566. stream = json.dumps(params)
  567. aClassName = ''.join([source,'(**',stream,')'])
  568. try:
  569. anObject = eval( aClassName)
  570. #setattr(anObject,'name',source)
  571. except Exception,e:
  572. print ['Error ',e]
  573. return anObject
  574. """
  575. This class implements a data-source handler that is intended to be used within the context of data processing, it allows to read/write anywhere transparently.
  576. The class is a facade to a heterogeneous class hierarchy and thus simplifies how the calling code interacts with the class hierarchy
  577. """
  578. class DataSource:
  579. def __init__(self,sourceType='Disk',outputType='Disk',params={}):
  580. self.Input = DataSourceFactory.instance(type=sourceType,args=params)
  581. self.Output= DataSourceFactory.instance(type=outputType,args=params)
  582. def read(self,size=-1):
  583. return self.Input.read(size)
  584. def write(self,**args):
  585. self.Output.write(**args)
  586. #p = {}
  587. #p['host'] = 'dev.the-phi.com'
  588. #p['uid'] = 'nyemba@gmail.com'
  589. #p['qid'] = 'repair'
  590. #factory = DataSourceFactory()
  591. #o = factory.instance(type='QueueReader',args=p)
  592. #print o is None
  593. #q = QueueWriter(host='dev.the-phi.com',uid='nyemba@gmail.com')
  594. #q.write(object='steve')
  595. #q.write(object='nyemba')
  596. #q.write(object='elon')