Refactoring

.github/workflows/coverage.yml

@@ -11,22 +11,22 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
 
       - name: Set up Python 3.12
-        uses: actions/setup-python@v3
+        uses: actions/setup-python@v5
         with:
           python-version: "3.12"
 
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
-          pip install coverage codecov mock mock_alchemy
+          pip install coverage codecov mock mock_alchemy pytest
           pip install .
 
       - name: Run coverage
         run: |
-          coverage run --source seq_dbutils -m unittest discover
+          coverage run --source seq_dbutils -m pytest
           coverage report -m
           coverage xml
 

.github/workflows/publish-to-test-pypi.yml

@@ -13,10 +13,10 @@ jobs:
       id-token: write
 
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
 
       - name: Set up Python 3.12
-        uses: actions/setup-python@v3
+        uses: actions/setup-python@v5
         with:
           python-version: "3.12"
 

.github/workflows/python-versions.yml

@@ -15,16 +15,16 @@ jobs:
         python-version: ["3.9", "3.10", "3.11", "3.12"]
 
     steps:
-    - uses: actions/checkout@v3
+    - uses: actions/checkout@v4
     - name: Set up Python ${{ matrix.python-version }}
-      uses: actions/setup-python@v3
+      uses: actions/setup-python@v5
       with:
         python-version: ${{ matrix.python-version }}
 
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        pip install flake8 mock mock_alchemy
+        pip install flake8 mock mock_alchemy pytest
         pip install .
 
     - name: Lint with flake8
@@ -34,6 +34,6 @@ jobs:
         # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
         flake8 . --count --max-complexity=10 --statistics
 
-    - name: Test with unittest
+    - name: Test with pytest
       run: |
-        python -m unittest
+        pytest

seq_dbutils/_version.txt

@@ -1 +1 @@
-1.0.6
+2.0.0

seq_dbutils/config.py

@@ -1,6 +1,4 @@
 import logging
-import os
-import sys
 from configparser import ConfigParser
 
 logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
@@ -11,25 +9,20 @@ class Config:
 
     @classmethod
     def initialize(cls, config_file_path):
-        if not os.path.isfile(config_file_path):
-            logging.error(f'Config file {config_file_path} does not exist. Exiting...')
-            sys.exit(1)
-        cls.configParser.read(config_file_path)
+        try:
+            return cls.configParser.read(config_file_path)
+        except FileNotFoundError:
+            raise FileNotFoundError(f'Config file {config_file_path} does not exist. Exiting...')
 
     @classmethod
     def get_section_config(cls, required_section, required_key):
         return cls.configParser.get(required_section, required_key)
 
     @staticmethod
     def get_db_config(required_section):
-        assert required_section
-        try:
-            logging.info(f"Extracting config for '{required_section}'")
-            user = Config.get_section_config(required_section, 'user')
-            key = Config.get_section_config(required_section, 'key').encode()
-            host = Config.get_section_config(required_section, 'host')
-            db = Config.get_section_config(required_section, 'db')
-            return user, key, host, db
-        except Exception as ex:
-            logging.error(str(ex))
-            sys.exit(1)
+        logging.info(f"Extracting config for '{required_section}'")
+        user = Config.get_section_config(required_section, 'user')
+        key = Config.get_section_config(required_section, 'key').encode()
+        host = Config.get_section_config(required_section, 'host')
+        db = Config.get_section_config(required_section, 'db')
+        return user, key, host, db

seq_dbutils/connection.py

@@ -1,5 +1,4 @@
 import logging
-import sys
 
 import sqlalchemy
 
@@ -20,11 +19,7 @@ def __init__(self, user, pwd, host, db, connector_type='mysqlconnector'):
         self.connector_type = connector_type
 
     def create_sql_engine(self, sql_logging=False):
-        try:
-            logging.info(f'Connecting to {self.db} on host {self.host}')
-            conn_str = f'mysql+{self.connector_type}://{self.user}:{self.pwd}@{self.host}/{self.db}'
-            sql_engine = sqlalchemy.create_engine(conn_str, echo=sql_logging)
-            return sql_engine
-        except Exception as ex:
-            logging.error(str(ex))
-            sys.exit(1)
+        logging.info(f'Connecting to {self.db} on host {self.host}')
+        conn_str = f'mysql+{self.connector_type}://{self.user}:{self.pwd}@{self.host}/{self.db}'
+        sql_engine = sqlalchemy.create_engine(conn_str, echo=sql_logging)
+        return sql_engine

seq_dbutils/dataframe.py

@@ -1,6 +1,4 @@
 import logging
-import sys
-from datetime import datetime
 
 import pandas as pd
 from sqlalchemy.engine import Engine
@@ -26,19 +24,3 @@ def create_db_table_dataframe(self, df):
         df_db_table = df.filter(db_table_col_list, axis=1)
         df_db_table = df_db_table.dropna(subset=df_db_table.columns, how='all')
         return df_db_table
-
-    @staticmethod
-    def apply_date_format(input_date, format_date):
-        if input_date:
-            format_time = format_date + ' %H:%M:%S'
-            try:
-                input_date = datetime.strptime(input_date, format_date).date()
-            except ValueError as ex:
-                if 'unconverted data remains:' in ex.args[0]:
-                    input_date = datetime.strptime(input_date, format_time).date()
-                else:
-                    logging.error(str(ex))
-                    sys.exit(1)
-        else:
-            input_date = None
-        return input_date

seq_dbutils/decrypt.py

@@ -15,5 +15,5 @@ def initialize(cls, key):
             for line in file_object:
                 pwd_encrypted = line
         pwd_decrypted = cipher_suite.decrypt(pwd_encrypted)
-        pwd_plain_text = bytes(pwd_decrypted).decode("utf-8")
+        pwd_plain_text = bytes(pwd_decrypted).decode('utf-8')
         return pwd_plain_text

seq_dbutils/load.py

@@ -1,5 +1,4 @@
 import logging
-import sys
 
 import pandas as pd
 
@@ -22,11 +21,10 @@ def bulk_insert_df_table(self):
                 logging.info(
                     f"Bulk inserting into table '{self.table_subclass.__tablename__}'. Number of rows: {len(self.df_table)}")
                 self.session_instance.bulk_insert_mappings(self.table_subclass, self.df_table.to_dict(orient='records'))
-            except Exception as ex:
+            except Exception:
                 logging.error('Failed to load data into database. Rolling back...')
                 self.session_instance.rollback()
-                logging.error(str(ex))
-                sys.exit(1)
+                raise
         else:
             logging.info(f"Skipping bulk insert for table '{self.table_subclass.__tablename__}' and empty dataframe")
 
@@ -36,10 +34,9 @@ def bulk_update_df_table(self):
                 logging.info(
                     f"Bulk updating table '{self.table_subclass.__tablename__}'. Number of rows: {len(self.df_table)}")
                 self.session_instance.bulk_update_mappings(self.table_subclass, self.df_table.to_dict(orient='records'))
-            except Exception as ex:
+            except Exception:
                 logging.error('Failed to update data in database. Rolling back...')
                 self.session_instance.rollback()
-                logging.error(str(ex))
-                sys.exit(1)
+                raise
         else:
             logging.info(f"Skipping bulk update for table '{self.table_subclass.__tablename__}' and empty dataframe")

setup.cfg

@@ -30,7 +30,7 @@ test_suite = tests
 setup_requires =
     setuptools
 install_requires =
-    cryptography ~=41.0.7
+    cryptography ~=43.0.1
     mysql-connector-python ~=8.1.0
     pandas ~=2.1.1
     SQLAlchemy ~=2.0.22

tests/test_args.py

@@ -1,14 +1,14 @@
-from unittest import TestCase
+import pytest
 
 from seq_dbutils import Args
 
 
-class ArgsTestClass(TestCase):
+@pytest.fixture()
+def args():
+    return Args.initialize_args()
 
-    def setUp(self):
-        self.parser = Args.initialize_args()
 
-    def test_initialize_args(self):
-        parsed = self.parser.parse_args(['TEST'])
-        config = vars(parsed)['config'][0]
-        self.assertEqual(config, 'TEST')
+def test_initialize_args(args):
+    parsed = args.parse_args(['TEST'])
+    config = vars(parsed)['config'][0]
+    assert config == 'TEST'

tests/test_config.py

@@ -1,56 +1,67 @@
+from configparser import NoOptionError
 from os.path import abspath, dirname, join
-from unittest import TestCase
 
+import pytest
 from mock import patch
 
 from seq_dbutils import Config
 
 DATA_DIR = join(dirname(abspath(__file__)), 'data')
 
 
-class ArgsTestClass(TestCase):
-
-    @staticmethod
-    @patch('logging.error')
-    @patch('sys.exit')
-    def test_initialize_no_file(mock_exit, mock_error):
-        file = join(DATA_DIR, 'fake.ini')
-        Config.initialize(file)
-        mock_exit.assert_called_once()
-
-    @patch('logging.info')
-    @patch('configparser.ConfigParser.read')
-    def test_initialize_ok(self, mock_read, mock_info):
-        file = join(DATA_DIR, 'test_initialize_ok.ini')
-        Config.initialize(file)
-        mock_read.assert_called_once_with(file)
-
-    @patch('logging.info')
-    @patch('configparser.ConfigParser.get')
-    def test_get_section_config(self, mock_get, mock_info):
-        required_section = 'Mock'
-        required_key = 'mock'
-        Config.get_section_config(required_section, required_key)
-        mock_get.assert_called_once_with(required_section, required_key)
-
-    @patch('logging.info')
-    @patch('seq_dbutils.config.Config.get_section_config')
-    def test_get_db_config_ok(self, mock_get_section, mock_info):
-        my_str = 'mock'
-        args = 'TEST'
-        mock_get_section.return_value = my_str
-        mock_get_section.encode.return_value = my_str
-        user, key, host, db = Config.get_db_config(args)
-        self.assertEqual(mock_get_section.call_count, 4)
-        assert user == my_str
-        assert key == b'mock'
-        assert host == my_str
-        assert db == my_str
-
-    @staticmethod
-    @patch('logging.error')
-    @patch('logging.info')
-    @patch('sys.exit')
-    def test_get_db_config_fail(mock_exit, mock_info, mock_error):
-        Config.get_db_config('error')
-        mock_exit.assert_called_once()
+def test_initialize_no_file():
+    with patch('configparser.ConfigParser.read', side_effect=FileNotFoundError()):
+        with pytest.raises(FileNotFoundError):
+            Config.initialize('fake.ini')
+
+
+def test_initialize_exception():
+    with patch('configparser.ConfigParser.read', side_effect=Exception()):
+        with pytest.raises(Exception):
+            Config.initialize('fake.ini')
+
+
+def test_initialize_ok():
+    data = 'file_contents'
+    with patch('configparser.ConfigParser.read', return_value=data):
+        result = Config.initialize('fake.ini')
+        assert result == data
+
+
+def test_get_section_config_no_option():
+    required_section = 'SOME_SECTION'
+    required_key = 'some_key'
+    with patch('configparser.ConfigParser.get', side_effect=NoOptionError(required_section, required_key)):
+        with pytest.raises(NoOptionError):
+            Config.get_section_config(required_section, required_key)
+
+
+def test_get_section_config_exception():
+    required_section = 'SOME_SECTION'
+    required_key = 'some_key'
+    with patch('configparser.ConfigParser.get', side_effect=Exception()):
+        with pytest.raises(Exception):
+            Config.get_section_config(required_section, required_key)
+
+
+def test_get_section_config_ok():
+    required_section = 'SOME_SECTION'
+    required_key = 'some_key'
+    data = 'some_config'
+    with patch('configparser.ConfigParser.get', return_value=data):
+        result = Config.get_section_config(required_section, required_key)
+        assert result == data
+
+
+def test_get_db_config_ok():
+    result = 'some_config'
+    with patch('logging.info'):
+        with patch('seq_dbutils.config.Config.get_section_config') as mock_get_section:
+            mock_get_section.return_value = result
+            mock_get_section.encode.return_value = result
+            user, key, host, db = Config.get_db_config('SOME_SECTION')
+            assert mock_get_section.call_count == 4
+            assert user == result
+            assert key == b'some_config'
+            assert host == result
+            assert db == result