csv append files and csv pooler

Author: sunsingerus

src/pumper.py

@@ -27,7 +27,7 @@ def write_rows_event(self, event=None):
         pass
 
     def write_rows_event_each_row(self, event=None):
-        self.writer.insert(event_or_events=event)
+        self.writer.insert(event)
 
     def reader_idle_event(self):
         self.writer.flush()

src/reader/mysqlreader.py

@@ -74,6 +74,9 @@ def read(self):
                         event.table = mysql_event.table
                         self.fire('WriteRowsEvent', event=event)
                         for row in mysql_event.rows:
+                            event = Event()
+                            event.schema = mysql_event.schema
+                            event.table = mysql_event.table
                             event.row = row['values']
                             self.fire('WriteRowsEvent.EachRow', event=event)
                     else:

src/writer/csvwriter.py

@@ -4,13 +4,15 @@
 from .writer import Writer
 from ..event.event import Event
 import csv
+import os.path
 
 
 class CSVWriter(Writer):
 
     file = None
     path = None
     writer = None
+    header_written = False
 
     def __init__(self, csv_file_path):
         self.path = csv_file_path
@@ -20,7 +22,12 @@ def opened(self):
 
     def open(self):
         if not self.opened():
-            self.file = open(self.path, 'w')
+            # do not write header to already existing file
+            # assume it was written earlier
+            if os.path.isfile(self.path):
+                self.header_written = True
+            # open file for write-at-the-end mode
+            self.file = open(self.path, 'a+')
 
     def insert(self, event_or_events):
         # event_or_events = [
@@ -49,8 +56,9 @@ def insert(self, event_or_events):
             self.open()
 
         if not self.writer:
-            self.writer = csv.DictWriter(self.file, fieldnames=event_or_events[0].row.keys())
-            self.writer.writeheader()
+            self.writer = csv.DictWriter(self.file, fieldnames=sorted(event_or_events[0].row.keys()))
+            if not self.header_written:
+                self.writer.writeheader()
 
         for event in event_or_events:
             self.writer.writerow(event.row)