Changeset 8631 for main/waeup.kofa

Timestamp:

6 Jun 2012, 00:45:24 (12 years ago)

Author:

uli

Message:

Add a helper function to append one CSV file to another.

Location:

main/waeup.kofa/trunk/src/waeup/kofa/utils

Files:

• helpers.py (modified) (3 diffs)
• tests/test_helpers.py (modified) (2 diffs)

main/waeup.kofa/trunk/src/waeup/kofa/utils/helpers.py

@@ -18 +18 @@
 """General helper functions for Kofa.
 """
+import csv
 import datetime
 import imghdr
@@ -24 +25 @@
 import re
 import shutil
+import tempfile
 import grok
 from cStringIO import StringIO
@@ -639 +641 @@
 if check_pdf not in imghdr.tests:
     imghdr.tests.append(check_pdf)
+
+def merge_csv_files(path1, path2):
+    """Merge two CSV files into one (appending).
+
+    CSV data from `path2` will be merged into `path1` csv file. This
+    is a bit like 'appending' data from path2 to data from path1.
+
+    The path of the resulting temporary file will be returned.
+
+    In the result file data from `path2` will always come _after_ data
+    from `path1`.
+
+    **Caution**: It is the _callers_ responsibility to remove the
+    result file (which is created by tempfile.mkstemp) after usage.
+
+    This CSV file merging copes with different column orders in both
+    CSV files and even with different column sets in both files.
+    """
+    # sniff the col names
+    row10 = csv.DictReader(open(path1, 'rb')).next()
+    row20 = csv.DictReader(open(path2, 'rb')).next()
+    fieldnames = sorted(list(set(row10.keys() + row20.keys())))
+    # now read/write the real data
+    reader1 = csv.DictReader(open(path1, 'rb'))
+    reader2 = csv.DictReader(open(path2, 'rb'))
+    wp, tmp_path = tempfile.mkstemp()
+    writer = csv.DictWriter(os.fdopen(wp, 'wb'), fieldnames)
+    writer.writerow(dict((x,x) for x in fieldnames)) # header
+    for row in reader1:
+        writer.writerow(row)
+    for row in reader2:
+        writer.writerow(row)
+    return tmp_path

main/waeup.kofa/trunk/src/waeup/kofa/utils/tests/test_helpers.py

@@ -372 +372 @@
         return
 
+class MergeCSVFileTestCase(unittest.TestCase):
+
+    def setUp(self):
+        self.workdir = tempfile.mkdtemp()
+        self.path1 = os.path.join(self.workdir, 'myfile1')
+        self.path2 = os.path.join(self.workdir, 'myfile2')
+        self.result_path = None
+        return
+
+    def tearDown(self):
+        shutil.rmtree(self.workdir)
+        if self.result_path is not None and os.path.exists(self.result_path):
+            os.unlink(self.result_path)
+        return
+
+    def test_basic(self):
+        # we can merge very basic CSV files
+        open(self.path1, 'wb').write('name,age\nManfred,32\n')
+        open(self.path2, 'wb').write('name,age\nBarney,28\n')
+        self.result_path = helpers.merge_csv_files(self.path1, self.path2)
+        contents = open(self.result_path, 'r').read()
+        self.assertEqual(
+            contents,
+            'age,name\r\n'
+            '32,Manfred\r\n'
+            '28,Barney\r\n')
+        return
+
+    def test_different_col_order(self):
+        # if cols of both files have different order, that won't stop us
+        open(self.path1, 'wb').write('name,age\nManfred,32\n')
+        open(self.path2, 'wb').write('age,name\n28,Barney\n')
+        self.result_path = helpers.merge_csv_files(self.path1, self.path2)
+        contents = open(self.result_path, 'r').read()
+        self.assertEqual(
+            contents,
+            'age,name\r\n'
+            '32,Manfred\r\n'
+            '28,Barney\r\n')
+        return
+
+    def test_different_cols_at_all(self):
+        # also cols available only in one file will work.
+        open(self.path1, 'wb').write('name,age\nManfred,32\n')
+        open(self.path2, 'wb').write('name,age,buddy\nBarney,28,Manfred\n')
+        self.result_path = helpers.merge_csv_files(self.path1, self.path2)
+        contents = open(self.result_path, 'r').read()
+        self.assertEqual(
+            contents,
+            'age,buddy,name\r\n'
+            '32,,Manfred\r\n'
+            '28,Manfred,Barney\r\n')
+        return
+
+
 def test_suite():
     suite = unittest.TestSuite()
@@ -386 +441 @@
         DateTimeHelpersTestCase,
         GetFileFormatTestCase,
+        MergeCSVFileTestCase,
         ]:
         suite.addTests(