Fix #7

Author: jferard

sqliteondbf/converter.py

@@ -18,27 +18,44 @@
 # * A part of this tool was inspired by https://github.com/olemb/dbfread/blob/master/examples/dbf2sqlite by Ole Martin Bjørndalen / UiT The Arctic University of Norway (under MIT licence)
 # * The example files are adapted from https://www.census.gov/data/tables/2016/econ/stc/2016-annual.html (I didn't find a copyright, but this is fair use I believe)
 
-import sqlite3
+import dbfread
 import logging
 import os
-import dbfread
+import sqlite3
+
 
 class SQLiteConverter():
  """A converter from dbf to sqlite3"""
- def __init__(self, connection = sqlite3.connect(":memory:"), logger=logging.getLogger("sqliteondbf")):
+
+ def __init__(self, connection=sqlite3.connect(":memory:"),
+ logger=logging.getLogger("sqliteondbf")):
  self.__connection = connection
  self.__logger = logger
 
  def import_dbf(self, dbf_path, lowernames=True, encoding="cp850", char_decode_errors="strict"):
  """Import a dbf database to the current sqlite connection"""
+ self.__check_path(dbf_path)
  cursor = self.__connection.cursor()
 
+ file_count = 0
  for fpath in self.__dbf_files(dbf_path):
- self.__logger.info("import dbf file {}".format(fpath))
- dbf_table = dbfread.DBF(fpath, lowernames=lowernames, encoding=encoding, char_decode_errors=char_decode_errors)
+ file_count += 1
+ self.__logger.info("import dbf file #{}: {}".format(file_count, fpath))
+ dbf_table = dbfread.DBF(fpath, lowernames=lowernames, encoding=encoding,
+ char_decode_errors=char_decode_errors)
  SQLiteConverterWorker(self.__logger, cursor, dbf_table).import_dbf_file()
 
- self.__connection.commit()
+ if file_count:
+ self.__logger.info("{} file(s) imported".format(file_count))
+ self.__connection.commit()
+ else:
+ message = "no dbf file in {}".format(dbf_path)
+ self.__logger.warning(message)
+
+ def __check_path(self, dbf_path):
+ if not os.path.isdir(dbf_path):
+ raise Exception("{} is not a directory".format(dbf_path))
+
 
  def __dbf_files(self, dbf_path):
  for root, _, names in os.walk(dbf_path):
@@ -47,6 +64,7 @@ def __dbf_files(self, dbf_path):
  if lext == ".dbf":
  yield os.path.join(root, name)
 
+
 class SQLiteConverterWorker():
  """The worker: converts a dbf table and add the table to the current connection"""
  __TYPEMAP = {
@@ -93,7 +111,7 @@ def __field_type(self, f):
  return SQLiteConverterWorker.__TYPEMAP.get(f.type, 'TEXT')
 
  def __populate_table(self):
- placeholders = ", ".join(["?"]*len(self.__dbf_table.fields))
+ placeholders = ", ".join(["?"] * len(self.__dbf_table.fields))
  sql = 'INSERT INTO "{}" VALUES ({})'.format(self.__dbf_table.name, placeholders)
  self.__logger.debug("populate table SQL:\n{}".format(sql))
 

test/converter_test.py

@@ -135,9 +135,11 @@ def __verify_calls(self, expected, mock):
  self.assertEquals(e, c)
 
  @patch("os.walk")
+ @patch("os.path.isdir")
  @patch("dbfread.DBF")
- def test_converter(self, mock_DBF, mock_walk):
+ def test_converter(self, mock_DBF, mock_isdir, mock_walk):
  connection = Mock()
+ mock_isdir.return_value = True
  mock_walk.return_value = [("root", "dirs", ["dbf1.dbf"])]
  mock_DBF().name = "dbf"
  f1 = Mock()
@@ -154,7 +156,6 @@ def test_converter(self, mock_DBF, mock_walk):
  # verify
  self.__verify_calls([
  call("dir"),
- ANY, # call().__iter__()
  ], mock_walk)
  self.assertTrue(
  call('root/dbf1.dbf', lowernames=True, encoding="cp850", char_decode_errors="strict") in mock_DBF.mock_calls)
@@ -168,9 +169,53 @@ def test_converter(self, mock_DBF, mock_walk):
  call.commit()
  ], connection)
  self.__verify_calls([
- call.info('import dbf file root/dbf1.dbf'),
+ call.info('import dbf file #1: root/dbf1.dbf'),
  call.debug('drop table SQL:\nDROP TABLE IF EXISTS "dbf"'),
  call.debug('create table SQL:\nCREATE TABLE "dbf" ("f1" DATETIME)'),
  call.debug('populate table SQL:\nINSERT INTO "dbf" VALUES (?)'),
- call.debug("rowcount: 10")
+ call.debug("rowcount: 10"),
+ call.info('1 file(s) imported'),
+ ], self.__logger)
+
+ @patch("os.path.isdir")
+ @patch("dbfread.DBF")
+ def test_converter_non_existing_dir(self, mock_DBF, mock_isdir):
+ connection = Mock()
+ mock_isdir.return_value = False
+
+ # replay
+
+ self.assertRaises(Exception, cv.SQLiteConverter(connection, self.__logger).import_dbf, "dir")
+
+
+ @patch("os.walk")
+ @patch("os.path.isdir")
+ @patch("dbfread.DBF")
+ def test_converter_empty_dir(self, mock_DBF, mock_isdir, mock_walk):
+ connection = Mock()
+ mock_isdir.return_value = True
+ mock_walk.return_value = [("root", "dirs", [])]
+ mock_DBF().name = "dbf"
+ f1 = Mock()
+ mock_DBF().fields = [f1]
+ f1.type = "T"
+ f1.name = "f1"
+ m = Mock()
+ connection.cursor.return_value = m
+ m.rowcount = 10
+
+ # replay
+ cv.SQLiteConverter(connection, self.__logger).import_dbf("dir")
+
+ # verify
+ self.__verify_calls([
+ call("dir"),
+ ], mock_walk)
+ self.assertEquals(
+ "dbf", mock_DBF().name)
+ self.__verify_calls([
+ call.cursor(),
+ ], connection)
+ self.__verify_calls([
+ call.warning('no dbf file in dir'),
  ], self.__logger)