Repository: bluelabsio/records-mover (language: python)

Issue:
use composition in RM-87 
Per @Brunope 's comment here:
https://github.com/bluelabsio/records-mover/pull/245#discussion_r1346349348

I don't want to create some weird inheritance hierarchy, but thought it might make sense to use the spirit of [Composition Over Inheritance](https://en.wikipedia.org/wiki/Composition_over_inheritance) and just have some methods we can import to use in these functions. Of course, this isn't strictly composition since it doesn't use another class to implement those methods.


Discussion / hints:

Fix patch:
diff --git a/.circleci/config.yml b/.circleci/config.yml
index e4a799e4c..c4cdf28db 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -28,6 +28,10 @@ commands:
         type: string
         description: "Version of numpy to test against, if specific version required"
         default: ""
+      sqlalchemy_version:
+        type: string
+        description: "Version of sqlalchemy to test against, if specific version required"
+        default: ""
     steps:
       - restore_cache:
           key: deps-v9-<<parameters.python_version>>-<<parameters.pandas_version>>-<<parameters.extras>>-<<parameters.include_dev_dependencies>>-{{ .Branch }}-{{ checksum "requirements.txt" }}-{{ checksum "setup.py" }}
@@ -35,7 +39,7 @@ commands:
           name: Install python deps in venv
           environment:
             PYENV_VERSION: <<parameters.python_version>>
-            SQLALCHEMY_SILENCE_UBER_WARNING: 1
+            SQLALCHEMY_WARN_20: 1
           command: |
             if [ -f venv/bin/activate ]
             then
@@ -62,6 +66,11 @@ commands:
               else
                 pip uninstall -y pandas
               fi
+              if [ '<<parameters.sqlalchemy_version>>' != '' ]
+              then
+                pip cache purge
+                pip install --force-reinstall -v --progress-bar=off 'sqlalchemy<<parameters.sqlalchemy_version>>'
+              fi
             fi
       - save_cache:
           key: deps-v9-<<parameters.python_version>>-<<parameters.pandas_version>>-<<parameters.extras>>-<<parameters.include_dev_dependencies>>-{{ .Branch }}-{{ checksum "requirements.txt" }}-{{ checksum "setup.py" }}
@@ -105,6 +114,10 @@ jobs:
         type: string
         description: "Version of pandas to test against"
         default: '>=1.5.3,<2'
+      sqlalchemy_version:
+        type: string
+        description: "Version of sqlalchemy to test against"
+        default: ''
       coverage:
         type: boolean
         # The python and pandas versions can affect the mypy and test
@@ -115,7 +128,7 @@ jobs:
     docker:
       - image: cimg/python:<<parameters.python_version>>
     environment:
-            SQLALCHEMY_SILENCE_UBER_WARNING: 1
+            SQLALCHEMY_WARN_20: 1
     steps:
       - checkout
       - add_ssh_keys:
@@ -126,6 +139,7 @@ jobs:
           extras: <<parameters.extras>>
           python_version: <<parameters.python_version>>
           pandas_version: <<parameters.pandas_version>>
+          sqlalchemy_version: <<parameters.sqlalchemy_version>>
           # includes numpy-stubs - should be obsolete and unneeded
           # starting with numpy 1.20
           include_dev_dependencies: true
@@ -185,11 +199,15 @@ jobs:
         type: string
         description: "Version of pandas to test against"
         default: '>=1.5.3,<2'
+      sqlalchemy_version:
+        type: string
+        description: "Version of sqlalchemy to test against"
+        default: ''
       command:
         type: string
         description: "Command to run in records-mover venv"
     environment:
-            SQLALCHEMY_SILENCE_UBER_WARNING: 1
+            SQLALCHEMY_WARN_20: 1
     docker:
       - image: cimg/python:<<parameters.python_version>>
       - image: jbfavre/vertica:8.1.1-16_centos-7
@@ -219,6 +237,7 @@ jobs:
           extras: <<parameters.extras>>
           python_version: <<parameters.python_version>>
           pandas_version: <<parameters.pandas_version>>
+          sqlalchemy_version: <<parameters.sqlalchemy_version>>
       - run:
           name: Install gcloud
           command: |
@@ -304,7 +323,7 @@ jobs:
         description: "Any flags you'd like to add to the pytest call"
         default: ""
     environment:
-            SQLALCHEMY_SILENCE_UBER_WARNING: 1
+            SQLALCHEMY_WARN_20: 1
     docker:
       - image: cimg/python:<<parameters.python_version>>
     steps:
@@ -337,7 +356,7 @@ jobs:
               # This is set by default in the CircleCI environment
               unset SCRATCH_GCS_URL
             fi
-            with-db <<parameters.db_name>> pytest <<parameters.pytest_flags>>
+            with-db <<parameters.db_name>> pytest -vv <<parameters.pytest_flags>>
       - store_test_results:
           path: test-reports
       - store_artifacts:
@@ -362,7 +381,7 @@ jobs:
     docker:
       - image: cimg/python:<<parameters.python_version>>
     environment:
-            SQLALCHEMY_SILENCE_UBER_WARNING: 1
+            SQLALCHEMY_WARN_20: 1
     steps:
       - checkout
       - installvenv:
@@ -413,7 +432,7 @@ jobs:
     docker:
       - image: cimg/python:3.9
     environment:
-      SQLALCHEMY_SILENCE_UBER_WARNING: 1
+      SQLALCHEMY_WARN_20: 1
     steps:
       - checkout
       - installvenv:
@@ -448,15 +467,6 @@ workflows:
       # We try to test against all non-end-of-life Python versions:
       #
       # https://devguide.python.org/devcycle/#end-of-life-branches
-      - test:
-          name: test-3.7
-          extras: '[unittest,typecheck]'
-          python_version: "3.7"
-          pandas_version: ">=1.3.5,<2"
-          context: slack-secrets
-          filters:
-            tags:
-              only: /v\d+\.\d+\.\d+(-[\w]+)?/
       - test:
           name: test-3.8
           extras: '[unittest,typecheck]'
@@ -492,6 +502,16 @@ workflows:
           filters:
             tags:
               only: /v\d+\.\d+\.\d+(-[\w]+)?/
+      - test:
+          name: test-3.11-old-sqlalchemy
+          extras: '[unittest,typecheck]'
+          python_version: "3.11"
+          sqlalchemy_version: "==1.4"
+          context: slack-secrets
+          coverage: true
+          filters:
+            tags:
+              only: /v\d+\.\d+\.\d+(-[\w]+)?/
       - integration_test_with_dbs:
           name: vertica-no-s3-itest
           extras: '[vertica,itest]'
@@ -509,7 +529,7 @@ workflows:
             # scratch bucket config.
             #
             unset SCRATCH_S3_URL AWS_SESSION_TOKEN AWS_SECRET_ACCESS_KEY AWS_ACCESS_KEY_ID
-            with-db dockerized-vertica pytest
+            with-db dockerized-vertica pytest -vv
           filters:
             tags:
               only: /v\d+\.\d+\.\d+(-[\w]+)?/
@@ -525,7 +545,7 @@ workflows:
             export RECORDS_MOVER_SESSION_TYPE=env
             mkdir -p test-reports/itest
             cd tests/integration/records/single_db
-            with-db dockerized-postgres pytest
+            with-db dockerized-postgres pytest -vv
           filters:
             tags:
               only: /v\d+\.\d+\.\d+(-[\w]+)?/
@@ -541,7 +561,7 @@ workflows:
             export RECORDS_MOVER_SESSION_TYPE=env
             mkdir -p test-reports/itest
             cd tests/integration/records/single_db
-            with-db dockerized-mysql pytest
+            with-db dockerized-mysql pytest -vv
           filters:
             tags:
               only: /v\d+\.\d+\.\d+(-[\w]+)?/
@@ -557,7 +577,7 @@ workflows:
             export RECORDS_MOVER_SESSION_TYPE=env
             mkdir -p test-reports/itest
             cd tests/integration/records/single_db
-            with-db dockerized-vertica pytest
+            with-db dockerized-vertica pytest -vv
           filters:
             tags:
               only: /v\d+\.\d+\.\d+(-[\w]+)?/
@@ -655,6 +675,33 @@ workflows:
           extras: '[literally_every_single_database_binary,itest]'
           python_version: "3.9"
           context: slack-secrets
+          command: |
+            . venv/bin/activate
+            export PATH=${PATH}:${PWD}/tests/integration/bin:/opt/vertica/bin
+            export DB_FACTS_PATH=${PWD}/tests/integration/circleci-dbfacts.yml
+            export RECORDS_MOVER_SESSION_TYPE=env
+            mkdir -p test-reports/itest
+            cd tests/integration/records/multi_db
+            pytest -vvv test_records_table2table.py
+      - integration_test_with_dbs:
+          name: tbl2tbl-itest-old-sqlalchemy
+          extras: '[literally_every_single_database_binary,itest]'
+          python_version: "3.9"
+          sqlalchemy_version: "==1.4"
+          context: slack-secrets
+          command: |
+            . venv/bin/activate
+            export PATH=${PATH}:${PWD}/tests/integration/bin:/opt/vertica/bin
+            export DB_FACTS_PATH=${PWD}/tests/integration/circleci-dbfacts.yml
+            export RECORDS_MOVER_SESSION_TYPE=env
+            mkdir -p test-reports/itest
+            cd tests/integration/records/multi_db
+            pytest -vvv test_records_table2table.py
+      - integration_test_with_dbs:
+          name: redshift-test-dataframe-schema-sql-creation
+          extras: '[redshift-binary,itest]'
+          python_version: "3.9"
+          context: slack-secrets
           command: |
             . venv/bin/activate
             export PATH=${PATH}:${PWD}/tests/integration/bin:/opt/vertica/bin
@@ -662,7 +709,7 @@ workflows:
             export RECORDS_MOVER_SESSION_TYPE=env
             mkdir -p test-reports/itest
             cd tests/integration
-            python3 -m records.multi_db.test_records_table2table
+            python3 -m records.multi_db.test_dataframe_schema_sql_creation
       - integration_test:
           name: bigquery-no-gcs-itest
           extras: '[bigquery,itest]'
@@ -690,7 +737,6 @@ workflows:
             - PyPI
             - slack-secrets
           requires:
-            - test-3.7
             - test-3.8
             - test-3.9
             - cli-extra-test
diff --git a/metrics/coverage_high_water_mark b/metrics/coverage_high_water_mark
index a19ddd68f..ea853819e 100644
--- a/metrics/coverage_high_water_mark
+++ b/metrics/coverage_high_water_mark
@@ -1,1 +1,1 @@
-93.4900
+92.0000
diff --git a/metrics/flake8_high_water_mark b/metrics/flake8_high_water_mark
index d00491fd7..573541ac9 100644
--- a/metrics/flake8_high_water_mark
+++ b/metrics/flake8_high_water_mark
@@ -1,1 +1,1 @@
-1
+0
diff --git a/metrics/mypy_high_water_mark b/metrics/mypy_high_water_mark
index c4c82ddc2..28e9eddba 100644
--- a/metrics/mypy_high_water_mark
+++ b/metrics/mypy_high_water_mark
@@ -1,1 +1,1 @@
-89.7000
+89.0000
diff --git a/records_mover/airflow/hooks/records_hook.py b/records_mover/airflow/hooks/records_hook.py
index 729989087..ff858c64c 100644
--- a/records_mover/airflow/hooks/records_hook.py
+++ b/records_mover/airflow/hooks/records_hook.py
@@ -8,6 +8,7 @@
 from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
 from typing import Optional, Union, List, TYPE_CHECKING
 import sqlalchemy
+from ...check_db_conn_engine import check_db_conn_engine
 
 try:
     # Works with Airflow 1
@@ -50,12 +51,15 @@ def _url_resolver(self) -> UrlResolver:
                            gcs_client_getter=lambda: None,
                            gcp_credentials_getter=lambda: None)
 
-    def _db_driver(self, db: Union[sqlalchemy.engine.Engine,
-                                   sqlalchemy.engine.Connection]) -> DBDriver:
+    def _db_driver(self, db: Optional[Union[sqlalchemy.engine.Engine,
+                                      sqlalchemy.engine.Connection]] = None,
+                   db_conn: Optional[sqlalchemy.engine.Connection] = None,
+                   db_engine: Optional[sqlalchemy.engine.Engine] = None) -> DBDriver:
         s3_temp_base_loc = (self._url_resolver.directory_url(self._s3_temp_base_url)
                             if self._s3_temp_base_url else None)
-
-        return db_driver(db=db, url_resolver=self._url_resolver, s3_temp_base_loc=s3_temp_base_loc)
+        db, db_conn, db_engine = check_db_conn_engine(db=db, db_conn=db_conn, db_engine=db_engine)
+        return db_driver(db=db, db_conn=db_conn, db_engine=db_engine,
+                         url_resolver=self._url_resolver, s3_temp_base_loc=s3_temp_base_loc)
 
     @property
     def _s3_temp_base_url(self) -> Optional[str]:
diff --git a/records_mover/check_db_conn_engine.py b/records_mover/check_db_conn_engine.py
new file mode 100644
index 000000000..236d8f690
--- /dev/null
+++ b/records_mover/check_db_conn_engine.py
@@ -0,0 +1,52 @@
+# flake8: noqa
+
+import sqlalchemy
+from typing import Union, Optional, Tuple
+import warnings
+
+
+def check_db_conn_engine(db: Optional[Union[sqlalchemy.engine.Engine,
+                                            sqlalchemy.engine.Connection]] = None,
+                         db_conn: Optional[sqlalchemy.engine.Connection] = None,
+                         db_engine: Optional[sqlalchemy.engine.Engine] = None) -> \
+    Tuple[Optional[Union[sqlalchemy.engine.Engine,
+                         sqlalchemy.engine.Connection]],
+          Optional[sqlalchemy.engine.Connection],
+          sqlalchemy.engine.Engine]:
+    """
+    This function was added because, in previous versions of SQL Alchemy, you could execute SQL against either
+    Engines or Connections and not worry very much about the difference between the two.  Since version 2, this
+    is no longer the case.
+
+    Because of the interchangeability between Engines and Connections, most of our code accepted a rather ambiguous
+    'db' argument, which could be either an Engine or a Connection.  This function is intended to be used to disabiguate
+    the two, create engines from connections where possible, and warn users to stop using the 'db' argument.
+
+    Args:
+        db: Deprecated argument. Use db_conn for Connection objects and db_engine for Engine objects.
+        db_conn: Optional SQLAlchemy Connection object.
+        db_engine: Optional SQLAlchemy Engine object.
+
+    Returns:
+        A tuple containing the Connection or Engine 'db' object provided as input, the Connection object if provided,
+        and the Engine object.
+
+    Raises:
+        ValueError: If none of db, db_conn, or db_engine are provided as arguments.
+    """
+    if db:
+        warnings.warn("The db argument is deprecated and will be"
+                      "removed in future releases.\n"
+                      "Please use db_conn for Connection objects and db_engine for Engine"
+                      "objects.",
+                      DeprecationWarning)
+    if not (db or db_conn or db_engine):
+        raise ValueError("Either db, db_conn, or db_engine must be provided as arguments")
+    if isinstance(db, sqlalchemy.engine.Connection) and not db_conn:
+        db_conn = db
+    if isinstance(db, sqlalchemy.engine.Engine) and not db_engine:
+        db_engine = db
+    if not db_engine:
+        print("db_engine is not provided, so we're assigning db_engine to db_conn.engine")
+        db_engine = db_conn.engine  # type: ignore[union-attr]
+    return (db, db_conn, db_engine)  # type: ignore[return-value]
diff --git a/records_mover/db/__init__.py b/records_mover/db/__init__.py
index 9ea299eb7..aca8c1638 100644
--- a/records_mover/db/__init__.py
+++ b/records_mover/db/__init__.py
@@ -4,6 +4,6 @@
     'create_sqlalchemy_url',
 ]
 
-from .driver import DBDriver  # noqa
+from .driver import DBDriver # noqa
 from .errors import LoadError  # noqa
 from .connect import create_sqlalchemy_url
diff --git a/records_mover/db/bigquery/bigquery_db_driver.py b/records_mover/db/bigquery/bigquery_db_driver.py
index b8646aae1..851ee3b35 100644
--- a/records_mover/db/bigquery/bigquery_db_driver.py
+++ b/records_mover/db/bigquery/bigquery_db_driver.py
@@ -19,17 +19,23 @@
 
 class BigQueryDBDriver(DBDriver):
     def __init__(self,
-                 db: Union[sqlalchemy.engine.Connection, sqlalchemy.engine.Engine],
+                 db: Optional[Union[sqlalchemy.engine.Connection, sqlalchemy.engine.Engine]],
                  url_resolver: UrlResolver,
                  gcs_temp_base_loc: Optional[BaseDirectoryUrl] = None,
+                 db_conn: Optional[sqlalchemy.engine.Connection] = None,
+                 db_engine: Optional[sqlalchemy.engine.Engine] = None,
                  **kwargs: object) -> None:
-        super().__init__(db)
+        super().__init__(db, db_conn, db_engine)
         self._bigquery_loader =\
-            BigQueryLoader(db=self.db,
+            BigQueryLoader(db=db,
+                           db_conn=db_conn,
+                           db_engine=db_engine,
                            url_resolver=url_resolver,
                            gcs_temp_base_loc=gcs_temp_base_loc)
         self._bigquery_unloader =\
-            BigQueryUnloader(db=self.db,
+            BigQueryUnloader(db=db,
+                             db_conn=db_conn,
+                             db_engine=db_engine,
                              url_resolver=url_resolver,
                              gcs_temp_base_loc=gcs_temp_base_loc)
 
diff --git a/records_mover/db/bigquery/loader.py b/records_mover/db/bigquery/loader.py
index 66c9b58cb..7aadd9bd4 100644
--- a/records_mover/db/bigquery/loader.py
+++ b/records_mover/db/bigquery/loader.py
@@ -19,19 +19,26 @@
 import logging
 from ..loader import LoaderFromFileobj
 from ..errors import NoTemporaryBucketConfiguration
+from ...check_db_conn_engine import check_db_conn_engine
+from ..db_conn_mixin import DBConnMixin
 
 logger = logging.getLogger(__name__)
 
 
-class BigQueryLoader(LoaderFromFileobj):
+class BigQueryLoader(DBConnMixin, LoaderFromFileobj):
     def __init__(self,
-                 db: Union[sqlalchemy.engine.Connection, sqlalchemy.engine.Engine],
+                 db: Optional[Union[sqlalchemy.engine.Connection, sqlalchemy.engine.Engine]],
                  url_resolver: UrlResolver,
-                 gcs_temp_base_loc: Optional[BaseDirectoryUrl])\
-            -> None:
+                 gcs_temp_base_loc: Optional[BaseDirectoryUrl],
+                 db_conn: Optional[sqlalchemy.engine.Connection] = None,
+                 db_engine: Optional[sqlalchemy.engine.Engine] = None) -> None:
+        db, db_conn, db_engine = check_db_conn_engine(db=db, db_conn=db_conn, db_engine=db_engine)
         self.db = db
+        self._db_conn = db_conn
+        self.db_engine = db_engine
         self.url_resolver = url_resolver
         self.gcs_temp_base_loc = gcs_temp_base_loc
+        self.conn_opened_here = False
 
     def best_scheme_to_load_from(self) -> str:
         return 'gs'
@@ -77,7 +84,7 @@ def load_from_fileobj(self, schema: str, table: str,
         logger.info("Loading from fileobj into BigQuery")
         # https://googleapis.github.io/google-cloud-python/latest/bigquery/usage/tables.html#creating-a-table
         connection: Connection =\
-            self.db.engine.raw_connection().connection
+            self.db_engine.raw_connection().connection
         # https://google-cloud.readthedocs.io/en/latest/bigquery/generated/google.cloud.bigquery.client.Client.html
         client: Client = connection._client
         project_id, dataset_id = self._parse_bigquery_schema_name(schema)
@@ -122,7 +129,7 @@ def load(self,
 
         logger.info("Loading from records directory into BigQuery")
         # https://googleapis.github.io/google-cloud-python/latest/bigquery/usage/tables.html#creating-a-table
-        connection: Connection = self.db.engine.raw_connection().connection
+        connection: Connection = self.db_engine.raw_connection().connection
         # https://google-cloud.readthedocs.io/en/latest/bigquery/generated/google.cloud.bigquery.client.Client.html
         client: Client = connection._client
         project_id, dataset_id = self._parse_bigquery_schema_name(schema)
@@ -187,3 +194,6 @@ def known_supported_records_formats_for_load(self) -> List[BaseRecordsFormat]:
             ParquetRecordsFormat(),
             AvroRecordsFormat()
         ]
+
+    def __del__(self) -> None:
+        self.del_db_conn()
diff --git a/records_mover/db/bigquery/unloader.py b/records_mover/db/bigquery/unloader.py
index 9c6339347..7d8c3ba79 100644
--- a/records_mover/db/bigquery/unloader.py
+++ b/records_mover/db/bigquery/unloader.py
@@ -12,20 +12,22 @@
 from records_mover.records.unload_plan import RecordsUnloadPlan
 from records_mover.records.records_directory import RecordsDirectory
 from records_mover.db.errors import NoTemporaryBucketConfiguration
+from ...check_db_conn_engine import check_db_conn_engine
 
 logger = logging.getLogger(__name__)
 
 
 class BigQueryUnloader(Unloader):
     def __init__(self,
-                 db: Union[sqlalchemy.engine.Connection, sqlalchemy.engine.Engine],
+                 db: Optional[Union[sqlalchemy.engine.Connection, sqlalchemy.engine.Engine]],
                  url_resolver: UrlResolver,
-                 gcs_temp_base_loc: Optional[BaseDirectoryUrl])\
-            -> None:
-        self.db = db
+                 gcs_temp_base_loc: Optional[BaseDirectoryUrl],
+                 db_conn: Optional[sqlalchemy.engine.Connection] = None,
+                 db_engine: Optional[sqlalchemy.engine.Engine] = None) -> None:
+        db, db_conn, db_engine = check_db_conn_engine(db=db, db_conn=db_conn, db_engine=db_engine)
         self.url_resolver = url_resolver
         self.gcs_temp_base_loc = gcs_temp_base_loc
-        super().__init__(db=db)
+        super().__init__(db=db, db_conn=db_conn, db_engine=db_engine)
 
     def can_unload_format(self, target_records_format: BaseRecordsFormat) -> bool:
         if isinstance(target_records_format, AvroRecordsFormat):
@@ -93,7 +95,7 @@ def unload(self,
         logger.info("Loading from records directory into BigQuery")
         # https://googleapis.github.io/google-cloud-python/latest/bigquery/usage/tables.html#creating-a-table
         connection: Connection =\
-            self.db.engine.raw_connection().connection
+            self.db_engine.raw_connection().connection
         # https://google-cloud.readthedocs.io/en/latest/bigquery/generated/google.cloud.bigquery.client.Client.html
         client: Client = connection._client
         project_id, dataset_id = self._parse_bigquery_schema_name(schema)
diff --git a/records_mover/db/db_conn_mixin.py b/records_mover/db/db_conn_mixin.py
new file mode 100644
index 000000000..da0668402
--- /dev/null
+++ b/records_mover/db/db_conn_mixin.py
@@ -0,0 +1,31 @@
+import logging
+import sqlalchemy
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+
+class DBConnMixin():
+    _db_conn: Optional[sqlalchemy.engine.Connection]
+    conn_opened_here: bool
+    db_conn: sqlalchemy.engine.Connection
+    db_engine: sqlalchemy.engine.Engine
+
+    def get_db_conn(self) -> sqlalchemy.engine.Connection:
+        if self._db_conn is None:
+            self._db_conn = self.db_engine.connect()
+            self.conn_opened_here = True
+            logger.debug(f"Opened connection to database within {self} because none was provided.")
+        return self._db_conn
+
+    def set_db_conn(self,
+                    db_conn: Optional[sqlalchemy.engine.Connection]) -> None:
+        self._db_conn = db_conn
+
+    def del_db_conn(self) -> None:
+        if self.conn_opened_here:
+            self.db_conn.close()
+            self.db_conn = None  # type: ignore[assignment]  # noqa: F821
+            self.conn_opened_here = False
+
+    db_conn = property(get_db_conn, set_db_conn, del_db_conn)  # type: ignore[assignment]  # noqa: F821,E501
diff --git a/records_mover/db/driver.py b/records_mover/db/driver.py
index 35bb60404..466618d0c 100644
--- a/records_mover/db/driver.py
+++ b/records_mover/db/driver.py
@@ -1,3 +1,4 @@
+from ..check_db_conn_engine import check_db_conn_engine
 from sqlalchemy.schema import CreateTable
 from ..records.records_format import BaseRecordsFormat
 from .loader import LoaderFromFileobj, LoaderFromRecordsDirectory
@@ -10,27 +11,34 @@
 from abc import ABCMeta, abstractmethod
 from records_mover.records import RecordsSchema
 from typing import Union, Dict, List, Tuple, Optional, TYPE_CHECKING
+from .db_conn_mixin import DBConnMixin
 if TYPE_CHECKING:
     from typing_extensions import Literal  # noqa
 
 logger = logging.getLogger(__name__)
 
 
-class DBDriver(metaclass=ABCMeta):
+class DBDriver(DBConnMixin, metaclass=ABCMeta):
     def __init__(self,
-                 db: Union[sqlalchemy.engine.Engine,
-                           sqlalchemy.engine.Connection], **kwargs) -> None:
+                 db: Optional[Union[sqlalchemy.engine.Engine,
+                              sqlalchemy.engine.Connection]],
+                 db_conn: Optional[sqlalchemy.engine.Connection] = None,
+                 db_engine: Optional[sqlalchemy.engine.Engine] = None,
+                 **kwargs) -> None:
+        db, db_conn, db_engine = check_db_conn_engine(db=db, db_conn=db_conn, db_engine=db_engine)
         self.db = db
-        self.db_engine = db.engine
+        self.db_engine = db_engine
+        self._db_conn = db_conn
+        self.conn_opened_here = False
         self.meta = MetaData()
 
     def has_table(self, schema: str, table: str) -> bool:
-        return self.db.dialect.has_table(self.db, table_name=table, schema=schema)
+        return sqlalchemy.inspect(self.db_engine).has_table(table, schema=schema)
 
     def table(self,
               schema: str,
               table: str) -> Table:
-        return Table(table, self.meta, schema=schema, autoload=True, autoload_with=self.db_engine)
+        return Table(table, self.meta, schema=schema, autoload_with=self.db_engine)
 
     def schema_sql(self,
                    schema: str,
@@ -42,7 +50,7 @@ def schema_sql(self,
         """
         # http://docs.sqlalchemy.org/en/latest/core/reflection.html
         table_obj = self.table(schema, table)
-        return str(CreateTable(table_obj, bind=self.db))
+        return str(CreateTable(table_obj))
 
     def varchar_length_is_in_chars(self) -> bool:
         """True if the 'n' in VARCHAR(n) is represented in natural language
@@ -50,35 +58,55 @@ def varchar_length_is_in_chars(self) -> bool:
         override it to control"""
         return False
 
-    def set_grant_permissions_for_groups(self, schema_name: str, table: str,
+    def set_grant_permissions_for_groups(self,
+                                         schema_name: str,
+                                         table: str,
                                          groups: Dict[str, List[str]],
-                                         db: Union[sqlalchemy.engine.Engine,
-                                                   sqlalchemy.engine.Connection]) -> None:
-        schema_and_table: str = quote_schema_and_table(self.db.engine, schema_name, table)
+                                         db: Optional[Union[sqlalchemy.engine.Engine,
+                                                      sqlalchemy.engine.Connection]],
+                                         db_conn: Optional[sqlalchemy.engine.Connection] = None,
+                                         db_engine: Optional[sqlalchemy.engine.Engine] = None
+                                         ) -> None:
+        db, db_conn, db_engine = check_db_conn_engine(db=db, db_conn=db_conn, db_engine=db_engine)
+        schema_and_table: str = quote_schema_and_table(None, schema_name, table,
+                                                       db_engine=self.db_engine)
         for perm_type in groups:
             groups_list = groups[perm_type]
             for group in groups_list:
-                group_name: str = quote_group_name(self.db.engine, group)
+                group_name: str = quote_group_name(None, group, db_engine=self.db_engine)
                 if not perm_type.isalpha():
                     raise TypeError("Please make sure your permission types"
                                     " are an acceptable value.")
                 perms_sql = f'GRANT {perm_type} ON TABLE {schema_and_table} TO {group_name}'
-                db.execute(perms_sql)
+                if db_conn:
+                    db_conn.execute(perms_sql)
+                else:
+                    with db_engine.connect() as conn:
+                        conn.execute(perms_sql)
 
     def set_grant_permissions_for_users(self, schema_name: str, table: str,
                                         users: Dict[str, List[str]],
-                                        db: Union[sqlalchemy.engine.Engine,
-                                                  sqlalchemy.engine.Connection]) -> None:
-        schema_and_table: str = quote_schema_and_table(self.db.engine, schema_name, table)
+                                        db: Optional[Union[sqlalchemy.engine.Engine,
+                                                     sqlalchemy.engine.Connection]],
+                                        db_conn: Optional[sqlalchemy.engine.Connection] = None,
+                                        db_engine: Optional[sqlalchemy.engine.Engine] = None
+                                        ) -> None:
+        db, db_conn, db_engine = check_db_conn_engine(db=db, db_conn=db_conn, db_engine=db_engine)
+        schema_and_table: str = quote_schema_and_table(None, schema_name, table,
+                                                       db_engine=self.db_engine)
         for perm_type in users:
             user_list = users[perm_type]
             for user in user_list:
-                user_name: str = quote_user_name(self.db.engine, user)
+                user_name: str = quote_user_name(self.db_engine, user)
                 if not perm_type.isalpha():
                     raise TypeError("Please make sure your permission types"
                                     " are an acceptable value.")
                 perms_sql = f'GRANT {perm_type} ON TABLE {schema_and_table} TO {user_name}'
-                db.execute(perms_sql)
+                if db_conn:
+                    db_conn.execute(perms_sql)
+                else:
+                    with db_engine.connect() as conn:
+                        conn.execute(perms_sql)
 
     def supports_time_type(self) -> bool:
         return True
@@ -193,6 +221,9 @@ def tweak_records_schema_after_unload(self,
                                           records_format: BaseRecordsFormat) -> RecordsSchema:
         return records_schema
 
+    def __del__(self) -> None:
+        self.del_db_conn()
+
 
 class GenericDBDriver(DBDriver):
     def loader_from_fileobj(self) -> None:
diff --git a/records_mover/db/factory.py b/records_mover/db/factory.py
index 69d35351d..48cf6b2d2 100644
--- a/records_mover/db/factory.py
+++ b/records_mover/db/factory.py
@@ -1,33 +1,38 @@
+# flake8: noqa
+
 from .driver import GenericDBDriver, DBDriver
 import sqlalchemy
-from typing import Union
+from typing import Union, Optional
+from ..check_db_conn_engine import check_db_conn_engine
 
 
-def db_driver(db: Union[sqlalchemy.engine.Engine,
-                        sqlalchemy.engine.Connection],
+def db_driver(db: Optional[Union[sqlalchemy.engine.Engine,
+                           sqlalchemy.engine.Connection]],
+              db_conn: Optional[sqlalchemy.engine.Connection] = None,
+              db_engine: Optional[sqlalchemy.engine.Engine] = None,
               **kwargs) -> DBDriver:
-    engine: sqlalchemy.engine.Engine = db.engine
-    engine_name: str = engine.name
+    db, db_conn, db_engine = check_db_conn_engine(db=db, db_conn=db_conn, db_engine=db_engine)
+    engine_name: str = db_engine.name
 
     if engine_name == 'vertica':
         from .vertica.vertica_db_driver import VerticaDBDriver
 
-        return VerticaDBDriver(db, **kwargs)
+        return VerticaDBDriver(db=db, db_conn=db_conn, db_engine=db_engine, **kwargs)
     elif engine_name == 'redshift':
         from .redshift.redshift_db_driver import RedshiftDBDriver
 
-        return RedshiftDBDriver(db, **kwargs)
+        return RedshiftDBDriver(db=db, db_conn=db_conn, db_engine=db_engine, **kwargs)
     elif engine_name == 'bigquery':
         from .bigquery.bigquery_db_driver import BigQueryDBDriver
 
-        return BigQueryDBDriver(db, **kwargs)
+        return BigQueryDBDriver(db=db, db_conn=db_conn, db_engine=db_engine, **kwargs)
     elif engine_name == 'postgresql':
         from .postgres.postgres_db_driver import PostgresDBDriver
 
-        return PostgresDBDriver(db, **kwargs)
+        return PostgresDBDriver(db=db, db_conn=db_conn, db_engine=db_engine, **kwargs)
     elif engine_name == 'mysql':
         from .mysql.mysql_db_driver import MySQLDBDriver
 
-        return MySQLDBDriver(db, **kwargs)
+        return MySQLDBDriver(db=db, db_conn=db_conn, db_engine=db_engine, **kwargs)
     else:
-        return GenericDBDriver(db, **kwargs)
+        return GenericDBDriver(db=db, db_conn=db_conn, db_engine=db_engine, **kwargs)
diff --git a/records_mover/db/mysql/loader.py b/records_mover/db/mysql/loader.py
index 374f5625b..3cf3e754e 100644
--- a/records_mover/db/mysql/loader.py
+++ b/records_mover/db/mysql/loader.py
@@ -9,18 +9,26 @@
 from .load_options import mysql_load_options
 from ...records.delimited import complain_on_unhandled_hints
 from ...url.resolver import UrlResolver
-from typing import Union, List
+from typing import Union, List, Optional
 import logging
 import tempfile
+from ...check_db_conn_engine import check_db_conn_engine
+from ..db_conn_mixin import DBConnMixin
 
 logger = logging.getLogger(__name__)
 
 
-class MySQLLoader(LoaderFromRecordsDirectory):
+class MySQLLoader(DBConnMixin, LoaderFromRecordsDirectory):
     def __init__(self,
-                 db: Union[sqlalchemy.engine.Engine, sqlalchemy.engine.Connection],
-                 url_resolver: UrlResolver) -> None:
+                 db: Optional[Union[sqlalchemy.engine.Engine, sqlalchemy.engine.Connection]],
+                 url_resolver: UrlResolver,
+                 db_conn: Optional[sqlalchemy.engine.Connection] = None,
+                 db_engine: Optional[sqlalchemy.engine.Engine] = None) -> None:
+        db, db_conn, db_engine = check_db_conn_engine(db=db, db_conn=db_conn, db_engine=db_engine)
+        self.conn_opened_here = False
         self.db = db
+        self._db_conn = db_conn
+        self.db_engine = db_engine
         self.url_resolver = url_resolver
 
     def load(self,
@@ -61,7 +69,7 @@ def load(self,
                                                       schema_name=schema)
             logger.info(f"Loading to MySQL with options: {load_options}")
             logger.info(str(sql))
-            self.db.execute(sql)
+            self.db_conn.execute(sql)
             logger.info("MySQL LOAD DATA complete.")
         return None
 
@@ -101,3 +109,6 @@ def known_supported_records_formats_for_load(self) -> List[BaseRecordsFormat]:
                                        'compression': None
                                    }),
         ]
+
+    def __del__(self) -> None:
+        self.del_db_conn()
diff --git a/records_mover/db/mysql/mysql_db_driver.py b/records_mover/db/mysql/mysql_db_driver.py
index f50822166..277bcd292 100644
--- a/records_mover/db/mysql/mysql_db_driver.py
+++ b/records_mover/db/mysql/mysql_db_driver.py
@@ -17,11 +17,15 @@
 
 class MySQLDBDriver(DBDriver):
     def __init__(self,
-                 db: Union[sqlalchemy.engine.Engine, sqlalchemy.engine.Connection],
                  url_resolver: UrlResolver,
+                 db: Optional[Union[sqlalchemy.engine.Engine, sqlalchemy.engine.Connection]] = None,
+                 db_conn: Optional[sqlalchemy.engine.Connection] = None,
+                 db_engine: Optional[sqlalchemy.engine.Engine] = None,
                  **kwargs) -> None:
-        super().__init__(db)
+        super().__init__(db=db, db_conn=db_conn, db_engine=db_engine)
         self._mysql_loader = MySQLLoader(db=db,
+                                         db_conn=db_conn,
+                                         db_engine=db_engine,
                                          url_resolver=url_resolver)
 
     def loader(self) -> Optional[LoaderFromRecordsDirectory]:
diff --git a/records_mover/db/postgres/loader.py b/records_mover/db/postgres/loader.py
index babe5a4a0..ef218b70c 100644
--- a/records_mover/db/postgres/loader.py
+++ b/records_mover/db/postgres/loader.py
@@ -1,5 +1,6 @@
 import sqlalchemy
 from sqlalchemy import MetaData
+from sqlalchemy import text
 from sqlalchemy.schema import Table
 from ..quoting import quote_value
 from ...url.resolver import UrlResolver
@@ -9,21 +10,29 @@
 from ...records.processing_instructions import ProcessingInstructions
 from .sqlalchemy_postgres_copy import copy_from
 from .copy_options import postgres_copy_from_options
-from typing import IO, Union, List, Iterable
+from typing import IO, Union, List, Iterable, Optional
 from ..loader import LoaderFromFileobj
 import logging
+from ...check_db_conn_engine import check_db_conn_engine
+from ..db_conn_mixin import DBConnMixin
 
 logger = logging.getLogger(__name__)
 
 
-class PostgresLoader(LoaderFromFileobj):
+class PostgresLoader(DBConnMixin, LoaderFromFileobj):
     def __init__(self,
                  url_resolver: UrlResolver,
                  meta: MetaData,
-                 db: Union[sqlalchemy.engine.Connection, sqlalchemy.engine.Engine]) -> None:
+                 db: Optional[Union[sqlalchemy.engine.Connection, sqlalchemy.engine.Engine]],
+                 db_conn: Optional[sqlalchemy.engine.Connection] = None,
+                 db_engine: Optional[sqlalchemy.engine.Engine] = None) -> None:
+        db, db_conn, db_engine = check_db_conn_engine(db=db, db_conn=db_conn, db_engine=db_engine)
         self.url_resolver = url_resolver
         self.db = db
+        self._db_conn = db_conn
+        self.db_engine = db_engine
         self.meta = meta
+        self.conn_opened_here = False
 
     def load_from_fileobj(self,
                           schema: str,
@@ -58,32 +67,30 @@ def load_from_fileobjs(self,
         table_obj = Table(table,
                           self.meta,
                           schema=schema,
-                          autoload=True,
-                          autoload_with=self.db)
+                          autoload_with=self.db_engine)
 
-        with self.db.engine.begin() as conn:
-            # https://www.postgresql.org/docs/8.3/sql-set.html
-            #
-            # The effects of SET LOCAL last only till the end of the
-            # current transaction, whether committed or not. A special
-            # case is SET followed by SET LOCAL within a single
-            # transaction: the SET LOCAL value will be seen until the end
-            # of the transaction, but afterwards (if the transaction is
-            # committed) the SET value will take effect.
-            date_style = f"ISO, {date_order_style}"
-            sql = f"SET LOCAL DateStyle = {quote_value(conn, date_style)}"
-            logger.info(sql)
-            conn.execute(sql)
+        # https://www.postgresql.org/docs/8.3/sql-set.html
+        #
+        # The effects of SET LOCAL last only till the end of the
+        # current transaction, whether committed or not. A special
+        # case is SET followed by SET LOCAL within a single
+        # transaction: the SET LOCAL value will be seen until the end
+        # of the transaction, but afterwards (if the transaction is
+        # committed) the SET value will take effect.
+        date_style = f"ISO, {date_order_style}"
+        sql = f"SET LOCAL DateStyle = {quote_value(None, date_style, db_engine=self.db_engine)}"
+        logger.info(sql)
+        self.db_conn.execute(text(sql))
 
-            for fileobj in fileobjs:
-                # Postgres COPY FROM defaults to appending data--we
-                # let the records Prep class decide what to do about
-                # the existing table, so it's safe to call this
-                # multiple times and append until done:
-                copy_from(fileobj,
-                          table_obj,
-                          conn,
-                          **postgres_options)
+        for fileobj in fileobjs:
+            # Postgres COPY FROM defaults to appending data--we
+            # let the records Prep class decide what to do about
+            # the existing table, so it's safe to call this
+            # multiple times and append until done:
+            copy_from(fileobj,
+                      table_obj,
+                      self.db_conn,
+                      **postgres_options)
         logger.info('Copy complete')
 
     def can_load_this_format(self, source_records_format: BaseRecordsFormat) -> bool:
@@ -130,3 +137,6 @@ def known_supported_records_formats_for_load(self) -> List[BaseRecordsFormat]:
             DelimitedRecordsFormat(variant='bigquery',
                                    hints={'compression': None}),
         ]
+
+    def __del__(self) -> None:
+        self.del_db_conn()
diff --git a/records_mover/db/postgres/postgres_db_driver.py b/records_mover/db/postgres/postgres_db_driver.py
index 3ca394209..dbc4e8c68 100644
--- a/records_mover/db/postgres/postgres_db_driver.py
+++ b/records_mover/db/postgres/postgres_db_driver.py
@@ -20,14 +20,20 @@
 
 class PostgresDBDriver(DBDriver):
     def __init__(self,
-                 db: Union[sqlalchemy.engine.Engine, sqlalchemy.engine.Connection],
+                 db: Optional[Union[sqlalchemy.engine.Engine, sqlalchemy.engine.Connection]],
                  url_resolver: UrlResolver,
+                 db_conn: Optional[sqlalchemy.engine.Connection] = None,
+                 db_engine: Optional[sqlalchemy.engine.Engine] = None,
                  **kwargs) -> None:
-        super().__init__(db)
+        super().__init__(db=db, db_conn=db_conn, db_engine=db_engine)
         self._postgres_loader = PostgresLoader(url_resolver=url_resolver,
                                                meta=self.meta,
-                                               db=self.db)
-        self._postgres_unloader = PostgresUnloader(db=self.db)
+                                               db=db,
+                                               db_conn=db_conn,
+                                               db_engine=db_engine)
+        self._postgres_unloader = PostgresUnloader(db=db,
+                                                   db_engine=db_engine,
+                                                   db_conn=db_conn)
 
     def loader(self) -> Optional[LoaderFromRecordsDirectory]:
         return self._postgres_loader
diff --git a/records_mover/db/postgres/unloader.py b/records_mover/db/postgres/unloader.py
index 9afba3d00..4392935aa 100644
--- a/records_mover/db/postgres/unloader.py
+++ b/records_mover/db/postgres/unloader.py
@@ -1,4 +1,5 @@
 from contextlib import contextmanager
+from sqlalchemy import text
 from sqlalchemy.schema import Table
 from .sqlalchemy_postgres_copy import copy_to
 from ..quoting import quote_value
@@ -43,30 +44,28 @@ def unload(self,
         table_obj = Table(table,
                           self.meta,
                           schema=schema,
-                          autoload=True,
-                          autoload_with=self.db)
+                          autoload_with=self.db_engine)
 
-        with self.db.engine.begin() as conn:
-            # https://www.postgresql.org/docs/8.3/sql-set.html
-            #
-            # The effects of SET LOCAL last only till the end of the
-            # current transaction, whether committed or not. A special
-            # case is SET followed by SET LOCAL within a single
-            # transaction: the SET LOCAL value will be seen until the end
-            # of the transaction, but afterwards (if the transaction is
-            # committed) the SET value will take effect.
-            date_style = f"{date_output_style}, {date_order_style}"
-            sql = f"SET LOCAL DateStyle = {quote_value(conn, date_style)}"
-            logger.info(sql)
-            conn.execute(sql)
+        # https://www.postgresql.org/docs/8.3/sql-set.html
+        #
+        # The effects of SET LOCAL last only till the end of the
+        # current transaction, whether committed or not. A special
+        # case is SET followed by SET LOCAL within a single
+        # transaction: the SET LOCAL value will be seen until the end
+        # of the transaction, but afterwards (if the transaction is
+        # committed) the SET value will take effect.
+        date_style = f"{date_output_style}, {date_order_style}"
+        sql = f"SET LOCAL DateStyle = {quote_value(None, date_style, db_engine=self.db_engine)}"
+        logger.info(sql)
+        self.db_conn.execute(text(sql))
 
-            filename = unload_plan.records_format.generate_filename('data')
-            loc = directory.loc.file_in_this_directory(filename)
-            with loc.open(mode='wb') as fileobj:
-                copy_to(table_obj.select(),
-                        fileobj,
-                        conn,
-                        **postgres_options)
+        filename = unload_plan.records_format.generate_filename('data')
+        loc = directory.loc.file_in_this_directory(filename)
+        with loc.open(mode='wb') as fileobj:
+            copy_to(table_obj.select(),
+                    fileobj,
+                    self.db_conn,
+                    **postgres_options)
 
         logger.info('Copy complete')
         directory.save_preliminary_manifest()
diff --git a/records_mover/db/quoting.py b/records_mover/db/quoting.py
index b322a5ac3..b651ff6af 100644
--- a/records_mover/db/quoting.py
+++ b/records_mover/db/quoting.py
@@ -1,10 +1,12 @@
 from sqlalchemy.engine import Engine, Connection
-from typing import Union
+from typing import Union, Optional
+from ..check_db_conn_engine import check_db_conn_engine
 
 
-def quote_schema_and_table(db: Union[Connection, Engine],
+def quote_schema_and_table(db: Optional[Union[Connection, Engine]],
                            schema: str,
-                           table: str) -> str:
+                           table: str,
+                           db_engine: Optional[Engine] = None) -> str:
     """
     Prevent SQL injection when we're not able to use bind
     variables (e.g., passing a table name in SQL).
@@ -18,22 +20,32 @@ def quote_schema_and_table(db: Union[Connection, Engine],
     quoted_bobby: "Robert'); DROP TABLE Students;--"
 
     """
-    dialect = db.dialect
+    db, _, db_engine = check_db_conn_engine(db=db, db_conn=None, db_engine=db_engine)
+    dialect = db_engine.dialect
     return (dialect.preparer(dialect).quote(schema) + '.' +
             dialect.preparer(dialect).quote(table))
 
 
-def quote_table_only(db: Union[Connection, Engine], table: str) -> str:
-    dialect = db.dialect
+def quote_table_only(db: Optional[Union[Connection, Engine]],
+                     table: str,
+                     db_engine: Optional[Engine] = None) -> str:
+    db, _, db_engine = check_db_conn_engine(db=db, db_conn=None, db_engine=db_engine)
+
+    dialect = db_engine.dialect
     return dialect.preparer(dialect).quote(table)
 
 
-def quote_column_name(db: Union[Connection, Engine], column_name: str) -> str:
-    dialect = db.dialect
+def quote_column_name(db: Optional[Union[Connection, Engine]],
+                      column_name: str,
+                      db_engine: Optional[Engine] = None) -> str:
+    db, _, db_engine = check_db_conn_engine(db=db, db_conn=None, db_engine=db_engine)
+    dialect = db_engine.dialect
     return dialect.preparer(dialect).quote(column_name)
 
 
-def quote_value(db: Union[Connection, Engine], value: str) -> str:
+def quote_value(db: Optional[Union[Connection, Engine]],
+                value: str,
+                db_engine: Optional[Engine] = None) -> str:
     """
     Prevent SQL injection on literal string values in places when we're
     not able to use bind variables (e.g., using weird DB-specific
@@ -48,15 +60,22 @@ def quote_value(db: Union[Connection, Engine], value: str) -> str:
     quoted_bobby: 'Robert''); DROP TABLE Students;--'
 
     """
-    dialect = db.dialect
+    db, _, db_engine = check_db_conn_engine(db=db, db_conn=None, db_engine=db_engine)
+    dialect = db_engine.dialect
     return dialect.preparer(dialect, initial_quote="'").quote(value)
 
 
-def quote_user_name(db: Union[Connection, Engine], user_name: str) -> str:
-    dialect = db.dialect
+def quote_user_name(db: Optional[Union[Connection, Engine]],
+                    user_name: str,
+                    db_engine: Optional[Engine] = None) -> str:
+    db, _, db_engine = check_db_conn_engine(db=db, db_conn=None, db_engine=db_engine)
+    dialect = db_engine.dialect
     return dialect.preparer(dialect).quote_identifier(user_name)
 
 
-def quote_group_name(db: Union[Connection, Engine], group_name: str) -> str:
-    dialect = db.dialect
+def quote_group_name(db: Optional[Union[Connection, Engine]],
+                     group_name: str,
+                     db_engine: Optional[Engine] = None) -> str:
+    db, _, db_engine = check_db_conn_engine(db=db, db_conn=None, db_engine=db_engine)
+    dialect = db_engine.dialect
     return dialect.preparer(dialect).quote_identifier(group_name)
diff --git a/records_mover/db/redshift/loader.py b/records_mover/db/redshift/loader.py
index 33e2abb13..80ccdd250 100644
--- a/records_mover/db/redshift/loader.py
+++ b/records_mover/db/redshift/loader.py
@@ -6,6 +6,7 @@
 from ...records.records_format import BaseRecordsFormat, DelimitedRecordsFormat, AvroRecordsFormat
 from ...records.processing_instructions import ProcessingInstructions
 import sqlalchemy
+from sqlalchemy import text
 from sqlalchemy.schema import Table
 import logging
 from .records_copy import redshift_copy_options
@@ -15,19 +16,26 @@
 from ...url import BaseDirectoryUrl
 from botocore.credentials import Credentials
 from ...records.delimited import complain_on_unhandled_hints
+from ...check_db_conn_engine import check_db_conn_engine
+from ..db_conn_mixin import DBConnMixin
 
 logger = logging.getLogger(__name__)
 
 
-class RedshiftLoader(LoaderFromRecordsDirectory):
+class RedshiftLoader(DBConnMixin, LoaderFromRecordsDirectory):
     def __init__(self,
-                 db: Union[sqlalchemy.engine.Engine, sqlalchemy.engine.Connection],
                  meta: sqlalchemy.MetaData,
-                 s3_temp_base_loc: Optional[BaseDirectoryUrl])\
-            -> None:
+                 s3_temp_base_loc: Optional[BaseDirectoryUrl],
+                 db: Optional[Union[sqlalchemy.engine.Engine, sqlalchemy.engine.Connection]],
+                 db_conn: Optional[sqlalchemy.engine.Connection] = None,
+                 db_engine: Optional[sqlalchemy.engine.Engine] = None) -> None:
+        db, db_conn, db_engine = check_db_conn_engine(db=db, db_conn=db_conn, db_engine=db_engine)
         self.db = db
+        self._db_conn = db_conn
+        self.db_engine = db_engine
         self.meta = meta
         self.s3_temp_base_loc = s3_temp_base_loc
+        self.conn_opened_here = False
 
     @contextmanager
     def temporary_s3_directory_loc(self) -> Iterator[BaseDirectoryUrl]:
@@ -91,9 +99,9 @@ def load(self,
                                empty_as_null=True,
                                **redshift_options)  # type: ignore
             logger.info(f"Starting Redshift COPY from {directory}...")
-            redshift_pid: int = self.db.execute("SELECT pg_backend_pid();").scalar()
+            redshift_pid: int = self.db_conn.execute(text("SELECT pg_backend_pid();")).scalar()
             try:
-                self.db.execute(copy)
+                self.db_conn.execute(copy)
             except sqlalchemy.exc.InternalError:
                 # Upon a load erorr, we receive:
                 #
@@ -173,3 +181,6 @@ def temporary_loadable_directory_loc(self) -> Iterator[BaseDirectoryUrl]:
 
     def has_temporary_loadable_directory_loc(self) -> bool:
         return self.s3_temp_base_loc is not None
+
+    def __del__(self) -> None:
+        self.del_db_conn()
diff --git a/records_mover/db/redshift/redshift_db_driver.py b/records_mover/db/redshift/redshift_db_driver.py
index 3635a408e..3846237fa 100644
--- a/records_mover/db/redshift/redshift_db_driver.py
+++ b/records_mover/db/redshift/redshift_db_driver.py
@@ -19,6 +19,7 @@
 from ..unloader import Unloader
 from .loader import RedshiftLoader
 from ..loader import LoaderFromRecordsDirectory
+from ...check_db_conn_engine import check_db_conn_engine
 
 
 logger = logging.getLogger(__name__)
@@ -26,22 +27,28 @@
 
 class RedshiftDBDriver(DBDriver):
     def __init__(self,
-                 db: Union[sqlalchemy.engine.Engine, sqlalchemy.engine.Connection],
+                 db: Optional[Union[sqlalchemy.engine.Engine, sqlalchemy.engine.Connection]],
+                 db_conn: Optional[sqlalchemy.engine.Connection] = None,
+                 db_engine: Optional[sqlalchemy.engine.Engine] = None,
                  s3_temp_base_loc: Optional[BaseDirectoryUrl] = None,
                  **kwargs) -> None:
-        super().__init__(db)
+        super().__init__(db=db, db_conn=db_conn, db_engine=db_engine)
         self.s3_temp_base_loc = s3_temp_base_loc
         self._redshift_loader =\
             RedshiftLoader(db=db,
+                           db_conn=db_conn,
+                           db_engine=db_engine,
                            meta=self.meta,
                            s3_temp_base_loc=s3_temp_base_loc)
         self._redshift_unloader =\
             RedshiftUnloader(db=db,
+                             db_conn=db_conn,
+                             db_engine=db_engine,
                              table=self.table,
                              s3_temp_base_loc=s3_temp_base_loc)
 
     def schema_sql(self, schema: str, table: str) -> str:
-        out = schema_sql_from_admin_views(schema, table, self.db)
+        out = schema_sql_from_admin_views(schema, table, None, db_conn=self.db_conn)
         if out is None:
             return super().schema_sql(schema=schema, table=table)
         else:
@@ -52,7 +59,7 @@ def schema_sql(self, schema: str, table: str) -> str:
     # tables and columns filled up memory in the job.
     @timeout_decorator.timeout(80)
     def table(self, schema: str, table: str) -> Table:
-        with self.db.engine.connect() as conn:
+        with self.db_engine.connect() as conn:
             with conn.begin():
                 # The code in the Redshift SQLAlchemy driver relies on 'SET
                 # LOCAL search_path' in order to do reflection and pull back
@@ -66,22 +73,32 @@ def table(self, schema: str, table: str) -> Table:
                 # There was some hang-wringing when they added it originally:
                 #  https://github.com/sqlalchemy-redshift/sqlalchemy-redshift/commit/
                 #      94723ec6437c5e5197fcf785845499e81640b167
-                return Table(table, self.meta, schema=schema, autoload=True, autoload_with=conn)
+                return Table(table, self.meta, schema=schema, autoload_with=conn)
 
-    def set_grant_permissions_for_groups(self, schema_name: str, table: str,
+    def set_grant_permissions_for_groups(self,
+                                         schema_name: str,
+                                         table: str,
                                          groups: Dict[str, List[str]],
-                                         db: Union[sqlalchemy.engine.Engine,
-                                                   sqlalchemy.engine.Connection]) -> None:
-        schema_and_table = quote_schema_and_table(self.db.engine, schema_name, table)
+                                         db: Optional[Union[sqlalchemy.engine.Engine,
+                                                      sqlalchemy.engine.Connection]],
+                                         db_conn: Optional[sqlalchemy.engine.Connection] = None,
+                                         db_engine: Optional[sqlalchemy.engine.Engine] = None
+                                         ) -> None:
+        db, db_conn, db_engine = check_db_conn_engine(db=db, db_conn=db_conn, db_engine=db_engine)
+        schema_and_table = quote_schema_and_table(None, schema_name, table, db_engine=db_engine)
         for perm_type in groups:
             groups_list = groups[perm_type]
             for group in groups_list:
-                group_name: str = quote_group_name(self.db.engine, group)
+                group_name: str = quote_group_name(None, group, db_engine=self.db_engine)
                 if not perm_type.isalpha():
                     raise TypeError("Please make sure your permission types"
                                     " are an acceptable value.")
                 perms_sql = f'GRANT {perm_type} ON TABLE {schema_and_table} TO GROUP {group_name}'
-                db.execute(perms_sql)
+                if db_conn:
+                    db_conn.execute(perms_sql)
+                else:
+                    with db_engine.connect() as conn:
+                        conn.execute(perms_sql)
         return None
 
     def supports_time_type(self) -> bool:
diff --git a/records_mover/db/redshift/sql.py b/records_mover/db/redshift/sql.py
index 9fd2ac1c1..63e25789d 100644
--- a/records_mover/db/redshift/sql.py
+++ b/records_mover/db/redshift/sql.py
@@ -1,7 +1,10 @@
+# flake8: noqa
+
 import sqlalchemy
 import logging
 from typing import Union, Optional
 from sqlalchemy.sql import text
+from ...check_db_conn_engine import check_db_conn_engine
 
 logger = logging.getLogger(__name__)
 
@@ -19,11 +22,15 @@
 
 def schema_sql_from_admin_views(schema: str,
                                 table: str,
-                                db: Union[sqlalchemy.engine.Engine, sqlalchemy.engine.Connection])\
+                                db: Optional[Union[sqlalchemy.engine.Engine,
+                                             sqlalchemy.engine.Connection]],
+                                db_conn: Optional[sqlalchemy.engine.Connection] = None,
+                                db_engine: Optional[sqlalchemy.engine.Engine] = None)\
         -> Optional[str]:
     # The default behavior in current sqlalchemy driver seems to
     # map "DOUBLE PRECISION" to "DOUBLE_PRECISION", so lean back
     # on admin views for now:
+    db, db_conn, db_engine = check_db_conn_engine(db=db, db_conn=db_conn, db_engine=db_engine)
     sql = text("""
 SELECT ddl
 FROM admin.v_generate_tbl_ddl
@@ -32,12 +39,14 @@ def schema_sql_from_admin_views(schema: str,
 """)
     out: Optional[str]
     try:
-        result = db.execute(sql, schema_name=schema, table_name=table).fetchall()
+        result = db_conn.execute(sql,  # type: ignore[union-attr]
+                                 {'schema_name': schema,
+                                  'table_name': table}).fetchall()
         if len(result) == 0:
             out = None
         else:
             # First line is a commented-out drop table, don't need that.
-            out = "\n".join(map(lambda r: r['ddl'], result[1:]))
+            out = "\n".join(map(lambda r: r.ddl, result[1:]))
     except sqlalchemy.exc.ProgrammingError:
         logger.debug("Error while generating SQL", exc_info=True)
         out = None
diff --git a/records_mover/db/redshift/unloader.py b/records_mover/db/redshift/unloader.py
index 3c6dae130..e6963ba9f 100644
--- a/records_mover/db/redshift/unloader.py
+++ b/records_mover/db/redshift/unloader.py
@@ -18,6 +18,7 @@
 from ..errors import CredsDoNotSupportS3Export, NoTemporaryBucketConfiguration
 from ...records.delimited import complain_on_unhandled_hints
 from ..unloader import Unloader
+from ...check_db_conn_engine import check_db_conn_engine
 
 
 logger = logging.getLogger(__name__)
@@ -25,11 +26,14 @@
 
 class RedshiftUnloader(Unloader):
     def __init__(self,
-                 db: Union[sqlalchemy.engine.Engine, sqlalchemy.engine.Connection],
+                 db: Optional[Union[sqlalchemy.engine.Engine, sqlalchemy.engine.Connection]],
                  table: Callable[[str, str], Table],
                  s3_temp_base_loc: Optional[BaseDirectoryUrl],
+                 db_conn: Optional[sqlalchemy.engine.Connection] = None,
+                 db_engine: Optional[sqlalchemy.engine.Engine] = None,
                  **kwargs) -> None:
-        super().__init__(db=db)
+        check_db_conn_engine(db=db, db_conn=db_conn, db_engine=db_engine)
+        super().__init__(db=db, db_conn=db_conn, db_engine=db_engine)
         self.table = table
         self.s3_temp_base_loc = s3_temp_base_loc
 
@@ -80,15 +84,17 @@ def unload_to_s3_directory(self,
             #
             register_secret(aws_creds.token)
             register_secret(aws_creds.secret_key)
-            select = text(f"SELECT * FROM {quote_schema_and_table(self.db, schema, table)}")
+            select = text(
+                "SELECT * FROM "
+                f"{quote_schema_and_table(None, schema, table, db_engine=self.db_engine)}")
             unload = UnloadFromSelect(select=select,
                                       access_key_id=aws_creds.access_key,
                                       secret_access_key=aws_creds.secret_key,
                                       session_token=aws_creds.token, manifest=True,
                                       unload_location=directory.loc.url, **redshift_options)
             try:
-                self.db.execute(unload)
-                out = self.db.execute(text("SELECT pg_last_unload_count()"))
+                self.db_conn.execute(unload)
+                out = self.db_conn.execute(text("SELECT pg_last_unload_count()"))
                 rows: Optional[int] = out.scalar()
                 assert rows is not None
                 logger.info(f"Just unloaded {rows} rows")
diff --git a/records_mover/db/unloader.py b/records_mover/db/unloader.py
index 7f12b960e..b678dcc41 100644
--- a/records_mover/db/unloader.py
+++ b/records_mover/db/unloader.py
@@ -7,12 +7,24 @@
 from typing import Union, List, Optional, Iterator
 from abc import ABCMeta, abstractmethod
 import sqlalchemy
+from ..check_db_conn_engine import check_db_conn_engine
+import logging
+from .db_conn_mixin import DBConnMixin
 
 
-class Unloader(metaclass=ABCMeta):
+logger = logging.getLogger(__name__)
+
+
+class Unloader(DBConnMixin, metaclass=ABCMeta):
     def __init__(self,
-                 db: Union[sqlalchemy.engine.Connection, sqlalchemy.engine.Engine]) -> None:
+                 db: Optional[Union[sqlalchemy.engine.Connection, sqlalchemy.engine.Engine]],
+                 db_conn: Optional[sqlalchemy.engine.Connection] = None,
+                 db_engine: Optional[sqlalchemy.engine.Engine] = None) -> None:
+        db, db_conn, db_engine = check_db_conn_engine(db=db, db_conn=db_conn, db_engine=db_engine)
         self.db = db
+        self._db_conn = db_conn
+        self.db_engine = db_engine
+        self.conn_opened_here = False
         self.meta = MetaData()
 
     @abstractmethod
@@ -65,3 +77,6 @@ def best_records_format(self) -> Optional[BaseRecordsFormat]:
         if len(supported_formats) == 0:
             return None
         return supported_formats[0]
+
+    def __del__(self) -> None:
+        self.del_db_conn()
diff --git a/records_mover/db/vertica/export_sql.py b/records_mover/db/vertica/export_sql.py
index f58947b76..9ab4824d8 100644
--- a/records_mover/db/vertica/export_sql.py
+++ b/records_mover/db/vertica/export_sql.py
@@ -21,7 +21,7 @@ def vertica_export_sql(db_engine: Engine,
     """
 
     def quote(value: str) -> str:
-        return quote_value(db_engine, value)
+        return quote_value(None, value, db_engine=db_engine)
 
     params_data = {
         "url": quote(s3_url + 'records.csv'),
@@ -32,7 +32,7 @@ def quote(value: str) -> str:
     }
 
     params = ", ".join([f"{key}={value}" for key, value in params_data.items()])
-    schema_and_table = quote_schema_and_table(db_engine, schema, table)
+    schema_and_table = quote_schema_and_table(None, schema, table, db_engine=db_engine)
     sql = template.format(params=params,
                           schema_and_table=schema_and_table)
 
diff --git a/records_mover/db/vertica/import_sql.py b/records_mover/db/vertica/import_sql.py
index da7662043..b4005dd6f 100644
--- a/records_mover/db/vertica/import_sql.py
+++ b/records_mover/db/vertica/import_sql.py
@@ -55,14 +55,14 @@ def quote(value: str) -> str:
         return quote_value(db_engine, value)
 
     if rejected_data_table is not None and rejected_data_schema is not None:
-        rejected_target = quote_schema_and_table(db_engine,
-                                                 rejected_data_schema, rejected_data_table)
+        rejected_target = quote_schema_and_table(None, rejected_data_schema, rejected_data_table,
+                                                 db_engine=db_engine)
         rejected_data = f"REJECTED DATA AS TABLE {rejected_target}"
     else:
         rejected_data = ''
 
     import_sql = import_sql_template.format(
-        schema_and_table=quote_schema_and_table(db_engine, schema, table),
+        schema_and_table=quote_schema_and_table(None, schema, table, db_engine=db_engine),
         # https://forum.vertica.com/discussion/238556/reading-gzip-files-from-s3-into-vertica
         gzip='GZIP' if gzip else '',
         # The capital E in the next line specifies a string literal
diff --git a/records_mover/db/vertica/loader.py b/records_mover/db/vertica/loader.py
index b7f4b7ed0..c93e2ad17 100644
--- a/records_mover/db/vertica/loader.py
+++ b/records_mover/db/vertica/loader.py
@@ -10,8 +10,9 @@
 from ...records.records_format import DelimitedRecordsFormat, BaseRecordsFormat
 from ...records.processing_instructions import ProcessingInstructions
 from ..loader import LoaderFromFileobj
-from typing import IO, Union, List, Type
+from typing import IO, Union, List, Type, Optional
 import logging
+from ...check_db_conn_engine import check_db_conn_engine
 
 logger = logging.getLogger(__name__)
 
@@ -19,9 +20,14 @@
 class VerticaLoader(LoaderFromFileobj):
     def __init__(self,
                  url_resolver: UrlResolver,
-                 db: Union[sqlalchemy.engine.Connection, sqlalchemy.engine.Engine]) -> None:
+                 db: Optional[Union[sqlalchemy.engine.Connection, sqlalchemy.engine.Engine]],
+                 db_conn: Optional[sqlalchemy.engine.Connection] = None,
+                 db_engine: Optional[sqlalchemy.engine.Engine] = None) -> None:
+        db, db_conn, db_engine = check_db_conn_engine(db=db, db_conn=db_conn, db_engine=db_engine)
         self.url_resolver = url_resolver
         self.db = db
+        self.db_conn = db_conn
+        self.db_engine = db_engine
 
     def load_from_fileobj(self,
                           schema: str,
@@ -41,11 +47,11 @@ def load_from_fileobj(self,
 
         # vertica_options isn't yet a TypedDict that matches the
         # vertica_import_sql options, so suppress type checking
-        import_sql = vertica_import_sql(db_engine=self.db.engine, table=table,
+        import_sql = vertica_import_sql(db_engine=self.db_engine, table=table,
                                         schema=schema, **vertica_options)   # type: ignore
         rawconn = None
         try:
-            rawconn = self.db.engine.raw_connection()
+            rawconn = self.db_engine.raw_connection()
             cursor = rawconn.cursor()
             logger.info(import_sql)
             if isinstance(fileobj, urllib.response.addinfourl):
diff --git a/records_mover/db/vertica/unloader.py b/records_mover/db/vertica/unloader.py
index fb980c1be..718879f95 100644
--- a/records_mover/db/vertica/unloader.py
+++ b/records_mover/db/vertica/unloader.py
@@ -1,5 +1,7 @@
+from ...check_db_conn_engine import check_db_conn_engine
 from records_mover.db.quoting import quote_value
 import sqlalchemy
+from sqlalchemy import text
 from contextlib import contextmanager
 from ...records.unload_plan import RecordsUnloadPlan
 from ...records.records_format import BaseRecordsFormat, DelimitedRecordsFormat
@@ -22,9 +24,12 @@
 
 class VerticaUnloader(Unloader):
     def __init__(self,
-                 db: Union[sqlalchemy.engine.Connection, sqlalchemy.engine.Engine],
-                 s3_temp_base_loc: Optional[BaseDirectoryUrl]) -> None:
-        super().__init__(db=db)
+                 db: Optional[Union[sqlalchemy.engine.Connection, sqlalchemy.engine.Engine]],
+                 s3_temp_base_loc: Optional[BaseDirectoryUrl],
+                 db_conn: Optional[sqlalchemy.engine.Connection] = None,
+                 db_engine: Optional[sqlalchemy.engine.Engine] = None) -> None:
+        db, db_conn, db_engine = check_db_conn_engine(db=db, db_conn=db_conn, db_engine=db_engine)
+        super().__init__(db=db, db_conn=db_conn, db_engine=db_engine)
         self.s3_temp_base_loc = s3_temp_base_loc
 
     @contextmanager
@@ -40,8 +45,8 @@ def aws_creds_sql(self, aws_id: str, aws_secret: str) -> str:
         return """
             ALTER SESSION SET UDPARAMETER FOR awslib aws_id={aws_id};
             ALTER SESSION SET UDPARAMETER FOR awslib aws_secret={aws_secret};
-        """.format(aws_id=quote_value(self.db.engine, aws_id),
-                   aws_secret=quote_value(self.db.engine, aws_secret))
+        """.format(aws_id=quote_value(None, aws_id, db_engine=self.db_engine),
+                   aws_secret=quote_value(None, aws_secret, db_engine=self.db_engine))
 
     def unload(self,
                schema: str,
@@ -69,7 +74,8 @@ def s3_temp_bucket_available(self) -> bool:
         return self.s3_temp_base_loc is not None
 
     def s3_export_available(self) -> bool:
-        out = self.db.execute("SELECT lib_name from user_libraries where lib_name = 'awslib'")
+        out = self.db_conn.execute(
+            text("SELECT lib_name from user_libraries where lib_name = 'awslib'"))
         available = len(list(out.fetchall())) == 1
         if not available:
             logger.info("Not attempting S3 export - no access to awslib in Vertica")
@@ -97,7 +103,11 @@ def unload_to_s3_directory(self,
         processing_instructions = unload_plan.processing_instructions
         try:
             s3_sql = self.aws_creds_sql(aws_creds.access_key, aws_creds.secret_key)
-            self.db.execute(s3_sql)
+            if not self.db_conn.in_transaction():
+                with self.db_conn.begin():
+                    self.db_conn.execute(text(s3_sql))
+            else:
+                self.db_conn.execute(text(s3_sql))
         except sqlalchemy.exc.ProgrammingError as e:
             raise DatabaseDoesNotSupportS3Export(str(e))
 
@@ -106,13 +116,13 @@ def unload_to_s3_directory(self,
                                     unhandled_hints,
                                     unload_plan.records_format.hints)
 
-        export_sql = vertica_export_sql(db_engine=self.db.engine,
+        export_sql = vertica_export_sql(db_engine=self.db_engine,
                                         table=table,
                                         schema=schema,
                                         s3_url=directory.loc.url,
                                         **vertica_options)
         logger.info(export_sql)
-        export_result = self.db.execute(export_sql).fetchall()
+        export_result = self.db_conn.execute(text(export_sql)).fetchall()
         directory.save_preliminary_manifest()
         export_count = 0
         for record in export_result:
diff --git a/records_mover/db/vertica/vertica_db_driver.py b/records_mover/db/vertica/vertica_db_driver.py
index 6e83b82d9..bc61b2548 100644
--- a/records_mover/db/vertica/vertica_db_driver.py
+++ b/records_mover/db/vertica/vertica_db_driver.py
@@ -14,6 +14,7 @@
 from ...utils.limits import (INT64_MIN, INT64_MAX,
                              FLOAT64_SIGNIFICAND_BITS,
                              num_digits)
+from ...check_db_conn_engine import check_db_conn_engine
 
 
 logger = logging.getLogger(__name__)
@@ -21,13 +22,18 @@
 
 class VerticaDBDriver(DBDriver):
     def __init__(self,
-                 db: Union[sqlalchemy.engine.Connection, sqlalchemy.engine.Engine],
+                 db: Optional[Union[sqlalchemy.engine.Connection, sqlalchemy.engine.Engine]],
                  url_resolver: UrlResolver,
                  s3_temp_base_loc: Optional[BaseDirectoryUrl] = None,
+                 db_conn: Optional[sqlalchemy.engine.Connection] = None,
+                 db_engine: Optional[sqlalchemy.engine.Engine] = None,
                  **kwargs: object) -> None:
-        super().__init__(db)
-        self._vertica_loader = VerticaLoader(url_resolver=url_resolver, db=self.db)
-        self._vertica_unloader = VerticaUnloader(s3_temp_base_loc=s3_temp_base_loc, db=db)
+        db, db_conn, db_engine = check_db_conn_engine(db=db, db_conn=db_conn, db_engine=db_engine)
+        super().__init__(db=db, db_conn=db_conn, db_engine=db_engine)
+        self._vertica_loader = VerticaLoader(url_resolver=url_resolver, db=db, db_conn=db_conn,
+                                             db_engine=db_engine)
+        self._vertica_unloader = VerticaUnloader(s3_temp_base_loc=s3_temp_base_loc, db=db,
+                                                 db_conn=db_conn, db_engine=db_engine)
         self.url_resolver = url_resolver
 
     def loader(self) -> Optional[LoaderFromRecordsDirectory]:
@@ -41,15 +47,17 @@ def unloader(self) -> Optional[Unloader]:
 
     def has_table(self, schema: str, table: str) -> bool:
         try:
-            sql = f"SELECT 1 from {quote_schema_and_table(self.db, schema, table)} limit 0;"
-            self.db.execute(sql)
+            sql = ("SELECT 1 "
+                   f"from {quote_schema_and_table(None, schema, table, db_engine=self.db_engine)} "
+                   "limit 0;")
+            self.db_conn.execute(text(sql))
             return True
         except sqlalchemy.exc.ProgrammingError:
             return False
 
     def schema_sql(self, schema: str, table: str) -> str:
-        sql = text("SELECT EXPORT_OBJECTS('', :schema_and_table, false)")
-        result = self.db.execute(sql, schema_and_table=f"{schema}.{table}").fetchall()
+        sql = text("SELECT EXPORT_OBJECTS('', :schema_and_table , false)")
+        result = self.db_conn.execute(sql, {'schema_and_table': f"{schema}.{table}"}).fetchall()
 
         if len(result) == 1:
             return result[0].EXPORT_OBJECTS
@@ -72,7 +80,7 @@ def table(self,
                           nullable=colinfo['nullable'],
                           default=colinfo['default'],
                           **colinfo.get('attrs', {}))
-                   for colinfo in self.db_engine.dialect.get_columns(self.db_engine,
+                   for colinfo in self.db_engine.dialect.get_columns(self.db_conn,
                                                                      table,
                                                                      schema=schema)]
         t = Table(table, self.meta, schema=schema, *columns)
diff --git a/records_mover/records/delimited/sniff.py b/records_mover/records/delimited/sniff.py
index e52d5ceb8..c0ba6d029 100644
--- a/records_mover/records/delimited/sniff.py
+++ b/records_mover/records/delimited/sniff.py
@@ -1,3 +1,4 @@
+# flake8: noqa
 import chardet
 from contextlib import contextmanager
 from . import PartialRecordsHints
@@ -240,7 +241,7 @@ def sniff_hints(fileobj: IO[bytes],
         # csv.Sniffer() can teach us some things about how each field
         # is encoded and whether there's a header:
         #
-        other_inferred_csv_hints = {}
+        other_inferred_csv_hints = PartialRecordsHints()
         if record_terminator_hint is not None:
             other_inferred_csv_hints['record-terminator'] = record_terminator_hint
             python_inferred_hints = csv_hints_from_python(fileobj,
@@ -268,7 +269,7 @@ def sniff_hints(fileobj: IO[bytes],
         #
         out: PartialRecordsHints = {
             'compression': compression_hint,
-            **pandas_inferred_hints,  # type: ignore
+            **pandas_inferred_hints,
             **python_inferred_hints,
             'encoding': final_encoding_hint,
             **other_inferred_csv_hints,
diff --git a/records_mover/records/mover.py b/records_mover/records/mover.py
index 5a4aa71c6..8829da9bc 100644
--- a/records_mover/records/mover.py
+++ b/records_mover/records/mover.py
@@ -117,7 +117,8 @@ def move(records_source: RecordsSource,
     elif isinstance(records_source, SupportsToFileobjsSource):
         # Incompatible types in assignment (expression has type "Optional[Any]",
         #     variable has type "BaseRecordsFormat")
-        target_records_format: BaseRecordsFormat = getattr(records_target, "records_format", None)  # type: ignore
+        target_records_format: BaseRecordsFormat \
+          = getattr(records_target, "records_format", None)  # type: ignore
         logger.info(f"Mover: copying from {records_source} to {records_target} "
                     f"by first writing {records_source} to {target_records_format} "
                     "records format (if easy to rewrite)...")
diff --git a/records_mover/records/prep.py b/records_mover/records/prep.py
index 947694d9e..d3a5caecd 100644
--- a/records_mover/records/prep.py
+++ b/records_mover/records/prep.py
@@ -5,6 +5,7 @@
 from records_mover.db import DBDriver
 from records_mover.records.table import TargetTableDetails
 import logging
+from sqlalchemy import text
 
 logger = logging.getLogger(__name__)
 
@@ -14,16 +15,18 @@ def __init__(self, target_table_details: TargetTableDetails) -> None:
         self.tbl = target_table_details
 
     def add_permissions(self, conn: Connection, driver: DBDriver) -> None:
-        schema_and_table: str = quote_schema_and_table(driver.db,
+        schema_and_table: str = quote_schema_and_table(None,
                                                        self.tbl.schema_name,
-                                                       self.tbl.table_name)
+                                                       self.tbl.table_name,
+                                                       db_engine=driver.db_engine)
         if self.tbl.add_group_perms_for is not None:
             logger.info(f"Adding permissions for {schema_and_table} "
                         f"to group {self.tbl.add_group_perms_for}")
             driver.set_grant_permissions_for_groups(self.tbl.schema_name,
                                                     self.tbl.table_name,
                                                     self.tbl.add_group_perms_for,
-                                                    conn)
+                                                    None,
+                                                    db_conn=conn)
         if self.tbl.add_user_perms_for is not None:
             logger.info(f"Adding permissions for {schema_and_table} "
                         f"to {self.tbl.add_user_perms_for}")
@@ -31,11 +34,12 @@ def add_permissions(self, conn: Connection, driver: DBDriver) -> None:
                 set_grant_permissions_for_users(self.tbl.schema_name,
                                                 self.tbl.table_name,
                                                 self.tbl.add_user_perms_for,
-                                                conn)
+                                                None,
+                                                db_conn=conn)
 
     def create_table(self, schema_sql: str, conn: Connection, driver: DBDriver) -> None:
         logger.info('Creating table...')
-        conn.execute(schema_sql)
+        conn.execute(text(schema_sql))
         logger.info(f"Just ran {schema_sql}")
         self.add_permissions(conn, driver)
         logger.info("Table prepped")
@@ -45,28 +49,31 @@ def prep_table_for_load(self,
                             existing_table_handling: ExistingTableHandling,
                             driver: DBDriver) -> None:
         logger.info("Looking for existing table..")
-        db = driver.db
+        db_engine = driver.db_engine
+        db_conn = driver.db_conn
 
         if driver.has_table(table=self.tbl.table_name, schema=self.tbl.schema_name):
             logger.info("Table already exists.")
             how_to_prep = existing_table_handling
-            schema_and_table: str = quote_schema_and_table(db,
+            schema_and_table: str = quote_schema_and_table(None,
                                                            self.tbl.schema_name,
-                                                           self.tbl.table_name)
+                                                           self.tbl.table_name,
+                                                           db_engine=db_engine,)
             if (how_to_prep == ExistingTableHandling.TRUNCATE_AND_OVERWRITE):
                 logger.info("Truncating...")
-                db.execute(f"TRUNCATE TABLE {schema_and_table}")
+                db_conn.execute(text(f"TRUNCATE TABLE {schema_and_table}"))
                 logger.info("Truncated.")
             elif (how_to_prep == ExistingTableHandling.DELETE_AND_OVERWRITE):
                 logger.info("Deleting rows...")
-                db.execute(f"DELETE FROM {schema_and_table} WHERE true")
+                db_conn.execute(text(f"DELETE FROM {schema_and_table} WHERE true"))
                 logger.info("Deleted")
             elif (how_to_prep == ExistingTableHandling.DROP_AND_RECREATE):
-                with db.engine.connect() as conn:
+                with db_engine.connect() as conn:
                     with conn.begin():
+                        logger.info(f"The connection object is: {conn}")
                         logger.info("Dropping and recreating...")
                         drop_table_sql = f"DROP TABLE {schema_and_table}"
-                        conn.execute(drop_table_sql)
+                        conn.execute(text(drop_table_sql))
                         logger.info(f"Just ran {drop_table_sql}")
                         self.create_table(schema_sql, conn, driver)
             elif (how_to_prep == ExistingTableHandling.APPEND):
@@ -74,7 +81,7 @@ def prep_table_for_load(self,
             else:
                 raise ValueError(f"Don't know how to handle {how_to_prep}")
         else:
-            with db.engine.connect() as conn:
+            with db_engine.connect() as conn:
                 with conn.begin():
                     self.create_table(schema_sql, conn, driver)
 
diff --git a/records_mover/records/prep_and_load.py b/records_mover/records/prep_and_load.py
index 3f099a9af..d009d49d4 100644
--- a/records_mover/records/prep_and_load.py
+++ b/records_mover/records/prep_and_load.py
@@ -16,26 +16,29 @@ def prep_and_load(tbl: TargetTableDetails,
                   load_exception_type: Type[Exception],
                   reset_before_reload: Callable[[], None] = lambda: None) -> MoveResult:
     logger.info("Connecting to database...")
-    with tbl.db_engine.begin() as db:
-        driver = tbl.db_driver(db)
-        prep.prep(schema_sql=schema_sql, driver=driver)
-    with tbl.db_engine.begin() as db:
-        # This second transaction ensures the table has been created
-        # before non-transactional statements like Redshift's COPY
-        # take place.  Otherwise you'll get an error like:
-        #
-        #  Cannot COPY into nonexistent table
-        driver = tbl.db_driver(db)
-        try:
-            import_count = load(driver)
-        except load_exception_type:
-            if not tbl.drop_and_recreate_on_load_error:
-                raise
-            reset_before_reload()
-            with tbl.db_engine.begin() as db:
-                driver = tbl.db_driver(db)
-                prep.prep(schema_sql=schema_sql,
-                          driver=driver,
-                          existing_table_handling=ExistingTableHandling.DROP_AND_RECREATE)
+    with tbl.db_engine.connect() as db_conn:
+        with db_conn.begin():
+            driver = tbl.db_driver(db=None, db_conn=db_conn)
+            prep.prep(schema_sql=schema_sql, driver=driver)
+    with tbl.db_engine.connect() as db_conn:
+        with db_conn.begin():
+            # This second transaction ensures the table has been created
+            # before non-transactional statements like Redshift's COPY
+            # take place.  Otherwise you'll get an error like:
+            #
+            #  Cannot COPY into nonexistent table
+            driver = tbl.db_driver(db=None, db_conn=db_conn)
+            try:
                 import_count = load(driver)
-        return MoveResult(move_count=import_count, output_urls=None)
+            except load_exception_type:
+                if not tbl.drop_and_recreate_on_load_error:
+                    raise
+                reset_before_reload()
+                with tbl.db_engine.connect() as db_conn:
+                    with db_conn.begin():
+                        driver = tbl.db_driver(db=None, db_conn=db_conn)
+                        prep.prep(schema_sql=schema_sql,
+                                  driver=driver,
+                                  existing_table_handling=ExistingTableHandling.DROP_AND_RECREATE)
+                        import_count = load(driver)
+            return MoveResult(move_count=import_count, output_urls=None)
diff --git a/records_mover/records/records.py b/records_mover/records/records.py
index 04de3ce7e..190d03a25 100644
--- a/records_mover/records/records.py
+++ b/records_mover/records/records.py
@@ -4,10 +4,10 @@
 from .targets import RecordsTargets
 from .mover import move
 from enum import Enum
-from typing import Callable, Union, TYPE_CHECKING
+from typing import Callable, Union, Optional, TYPE_CHECKING
 if TYPE_CHECKING:
     from sqlalchemy.engine import Engine, Connection  # noqa
-    from ..db import DBDriver  # noqa
+    from ..db import DBDriver # noqa
 
     from records_mover import Session  # noqa
 
@@ -60,7 +60,9 @@ class Records:
     "Alias of :meth:`records_mover.records.move`"
 
     def __init__(self,
-                 db_driver: Union[Callable[[Union['Engine', 'Connection']], 'DBDriver'],
+                 db_driver: Union[Callable[[Union['Engine', 'Connection', None],
+                                            Optional['Connection'],
+                                            Optional['Engine']], 'DBDriver'],
                                   PleaseInfer] = PleaseInfer.token,
                  url_resolver: Union[UrlResolver, PleaseInfer] = PleaseInfer.token,
                  session: Union['Session', PleaseInfer] = PleaseInfer.token) -> None:
diff --git a/records_mover/records/schema/field/sqlalchemy.py b/records_mover/records/schema/field/sqlalchemy.py
index 0f4b61977..444594c47 100644
--- a/records_mover/records/schema/field/sqlalchemy.py
+++ b/records_mover/records/schema/field/sqlalchemy.py
@@ -107,7 +107,7 @@ def field_from_sqlalchemy_column(column: Column,
                                  driver=driver)
     representations = {
         'origin': RecordsSchemaFieldRepresentation.
-        from_sqlalchemy_column(column, driver.db.dialect,
+        from_sqlalchemy_column(column, driver.db_engine.dialect,
                                rep_type)
     }
 
diff --git a/records_mover/records/schema/schema/known_representation.py b/records_mover/records/schema/schema/known_representation.py
index 72f86ce39..52aa802b8 100644
--- a/records_mover/records/schema/schema/known_representation.py
+++ b/records_mover/records/schema/schema/known_representation.py
@@ -60,7 +60,7 @@ def from_dataframe(df: 'DataFrame',
     def from_db_driver(driver: 'DBDriver',
                        schema_name: str,
                        table_name: str) -> 'RecordsSchemaSqlKnownRepresentation':
-        type = f"sql/{driver.db.dialect.name}"
+        type = f"sql/{driver.db_engine.dialect.name}"
         ddl = driver.schema_sql(schema_name, table_name)
         return RecordsSchemaSqlKnownRepresentation(type=type, table_ddl=ddl)
 
diff --git a/records_mover/records/schema/schema/sqlalchemy.py b/records_mover/records/schema/schema/sqlalchemy.py
index 9f2aa0045..082ca68d8 100644
--- a/records_mover/records/schema/schema/sqlalchemy.py
+++ b/records_mover/records/schema/schema/sqlalchemy.py
@@ -21,10 +21,7 @@ def schema_to_schema_sql(records_schema: 'RecordsSchema',
     table = Table(table_name, meta,
                   *columns,
                   schema=schema_name)
-    db = None
-    if driver is not None:
-        db = driver.db
-    return str(CreateTable(table, bind=db))
+    return str(CreateTable(table).compile(driver.db_engine))
 
 
 def schema_from_db_table(schema_name: str,
diff --git a/records_mover/records/sources/factory.py b/records_mover/records/sources/factory.py
index 19cf5326f..8228f4ece 100644
--- a/records_mover/records/sources/factory.py
+++ b/records_mover/records/sources/factory.py
@@ -1,3 +1,5 @@
+# flake8: noqa
+
 import pathlib
 from ..records_format import BaseRecordsFormat
 from ..schema import RecordsSchema
@@ -15,7 +17,7 @@
 if TYPE_CHECKING:
     # see the 'gsheets' extras_require option in setup.py - needed for this!
     import google.auth.credentials  # noqa
-    from sqlalchemy.engine import Engine  # noqa
+    from sqlalchemy.engine import Engine, Connection  # noqa
     from ...db import DBDriver  # noqa
     from .google_sheets import GoogleSheetsRecordsSource  # noqa ditto
     # with pandas, which an optional addition for clients of this
@@ -50,7 +52,9 @@ class RecordsSources(object):
     """
 
     def __init__(self,
-                 db_driver: Callable[['Engine'], 'DBDriver'],
+                 db_driver: Callable[[Optional[Union['Engine', 'Connection']],
+                                      Optional['Connection'],
+                                      Optional['Engine']], 'DBDriver'],
                  url_resolver: UrlResolver) -> None:
         self.db_driver = db_driver
         self.url_resolver = url_resolver
@@ -158,18 +162,22 @@ def data_url(self,
     def table(self,
               db_engine: 'Engine',
               schema_name: str,
-              table_name: str) -> 'TableRecordsSource':
+              table_name: str,
+              db_conn: Optional['Connection'] = None) -> 'TableRecordsSource':
         """Represents a SQLALchemy-accessible database table as as a source.
 
         :param db_engine: SQLAlchemy database engine to pull data from.
         :param schema_name: Schema name of a table to get data from.
         :param table_name: Table name of a table to get data from.
+        :param db_conn: SQLAlchemy database connection to use to pull data from.
         """
         from .table import TableRecordsSource  # noqa
-        return TableRecordsSource(schema_name=schema_name,
-                                  table_name=table_name,
-                                  url_resolver=self.url_resolver,
-                                  driver=self.db_driver(db_engine))
+        return TableRecordsSource(
+            schema_name=schema_name,
+            table_name=table_name,
+            url_resolver=self.url_resolver,
+            driver=self.db_driver(None, db_engine=db_engine,  # type: ignore[call-arg]
+                                  db_conn=db_conn))
 
     def directory_from_url(self,
                            url: str,
diff --git a/records_mover/records/sources/table.py b/records_mover/records/sources/table.py
index 9dd7ca889..4d7826e95 100644
--- a/records_mover/records/sources/table.py
+++ b/records_mover/records/sources/table.py
@@ -8,6 +8,7 @@
 from ..records_format import BaseRecordsFormat
 from ..unload_plan import RecordsUnloadPlan
 from ..results import MoveResult
+from sqlalchemy import text
 from sqlalchemy.engine import Engine
 from contextlib import contextmanager
 from ..schema import RecordsSchema
@@ -74,15 +75,11 @@ def to_dataframes_source(self,
         from .dataframes import DataframesRecordsSource  # noqa
         import pandas
 
-        db = self.driver.db
+        db_conn = self.driver.db_conn
+        db_engine = self.driver.db_engine
         records_schema = self.pull_records_schema()
 
-        if isinstance(db, Engine):
-            connection = db.connect()
-            columns = db.dialect.get_columns(connection, self.table_name, schema=self.schema_name)
-            connection.close()
-        else:
-            columns = db.dialect.get_columns(db, self.table_name, schema=self.schema_name)
+        columns = db_engine.dialect.get_columns(db_conn, self.table_name, schema=self.schema_name)
 
         num_columns = len(columns)
         if num_columns == 0:
@@ -91,10 +88,11 @@ def to_dataframes_source(self,
         chunksize = int(entries_per_chunk / num_columns)
         logger.info(f"Exporting in chunks of up to {chunksize} rows by {num_columns} columns")
 
-        quoted_table = quote_schema_and_table(db, self.schema_name, self.table_name)
+        quoted_table = quote_schema_and_table(None, self.schema_name,
+                                              self.table_name, db_engine=db_engine,)
         chunks: Generator['DataFrame', None, None] = \
-            pandas.read_sql(f"SELECT * FROM {quoted_table}",
-                            con=db,
+            pandas.read_sql(text(f"SELECT * FROM {quoted_table}"),
+                            con=db_conn,
                             chunksize=chunksize)
         try:
             yield DataframesRecordsSource(dfs=self.with_cast_dataframe_types(records_schema,
diff --git a/records_mover/records/table.py b/records_mover/records/table.py
index a10f47342..c94bb6958 100644
--- a/records_mover/records/table.py
+++ b/records_mover/records/table.py
@@ -24,5 +24,8 @@ class TargetTableDetails(metaclass=ABCMeta):
     #
     # https://github.com/python/mypy/issues/5485
     # @abstractmethod
-    def db_driver(self, db: Union[Engine, Connection]) -> DBDriver:  # type: ignore
+    def db_driver(self,  # type: ignore
+                  db: Optional[Union[Engine, Connection]],
+                  db_engine: Optional[Engine] = None,
+                  db_conn: Optional[Connection] = None) -> DBDriver:
         ...
diff --git a/records_mover/records/targets/factory.py b/records_mover/records/targets/factory.py
index 8c0ecec93..2d1872c5f 100644
--- a/records_mover/records/targets/factory.py
+++ b/records_mover/records/targets/factory.py
@@ -43,7 +43,9 @@ class RecordsTargets(object):
 
     def __init__(self,
                  url_resolver: UrlResolver,
-                 db_driver: Callable[[Union['Engine', 'Connection']], 'DBDriver']) -> None:
+                 db_driver: Callable[[Optional[Union['Engine', 'Connection']],
+                                      Optional['Connection'],
+                                      Optional['Engine']], 'DBDriver']) -> None:
         self.url_resolver = url_resolver
         self.db_driver = db_driver
 
@@ -73,7 +75,8 @@ def table(self,
               ExistingTableHandling.DELETE_AND_OVERWRITE,
               drop_and_recreate_on_load_error: bool = False,
               add_user_perms_for: Optional[Dict[str, List[str]]] = None,
-              add_group_perms_for: Optional[Dict[str, List[str]]] = None) -> \
+              add_group_perms_for: Optional[Dict[str, List[str]]] = None,
+              db_conn: Optional['Connection'] = None) -> \
             'TableRecordsTarget':
         """Represents a SQLALchemy-accessible database table as as a target.
 
@@ -97,6 +100,9 @@ def table(self,
         :param add_group_perms_for: If specified, a table's permissions will be set for the
            specified group. Format should be like {'all': ['group1', 'group2'], 'select': ['group3',
            'group4']}
+
+        :param db_conn: SQLAlchemy database connection to write data to.  If not specified, one
+           will be created from the db_engine.
         """
         from .table import TableRecordsTarget  # noqa
         return TableRecordsTarget(schema_name=schema_name,
@@ -106,7 +112,8 @@ def table(self,
                                   existing_table_handling=existing_table_handling,
                                   drop_and_recreate_on_load_error=drop_and_recreate_on_load_error,
                                   add_user_perms_for=add_user_perms_for,
-                                  add_group_perms_for=add_group_perms_for)
+                                  add_group_perms_for=add_group_perms_for,
+                                  db_conn=db_conn)
 
     def google_sheet(self,
                      spreadsheet_id: str,
diff --git a/records_mover/records/targets/spectrum.py b/records_mover/records/targets/spectrum.py
index ccb0dc0a4..aab01cf01 100644
--- a/records_mover/records/targets/spectrum.py
+++ b/records_mover/records/targets/spectrum.py
@@ -1,3 +1,5 @@
+# flake8: noqa
+
 from .base import SupportsRecordsDirectory
 from records_mover.db.quoting import quote_schema_and_table
 from ...db import DBDriver
@@ -11,6 +13,7 @@
 from typing import Optional, Callable, Union
 import logging
 import sqlalchemy
+from sqlalchemy import text
 
 
 logger = logging.getLogger(__name__)
@@ -21,14 +24,19 @@ def __init__(self,
                  schema_name: str,
                  table_name: str,
                  db_engine: Engine,
-                 db_driver: Callable[[Union[Engine, Connection]], DBDriver],
+                 db_driver: Callable[[Optional[Union['Engine', 'Connection']],
+                                      Optional[Connection],
+                                      Optional[Engine]], DBDriver],
                  url_resolver: UrlResolver,
                  spectrum_base_url: Optional[str],
                  spectrum_rdir_url: Optional[str],
                  existing_table_handling: ExistingTableHandling =
                  ExistingTableHandling.TRUNCATE_AND_OVERWRITE) -> None:
-        self.db = db_engine
-        self.driver = db_driver(db_engine)
+        self.db = None
+        self.db_engine = db_engine
+        self.driver = db_driver(db=None,  # type: ignore[call-arg]
+                                db_conn=None,
+                                db_engine=db_engine)
         self.schema_name = schema_name
         self.table_name = table_name
         self.url_resolver = url_resolver
@@ -70,14 +78,15 @@ def prep_bucket(self) -> None:
             if self.existing_table_handling == ExistingTableHandling.DELETE_AND_OVERWRITE:
                 logger.warning('Redshift Spectrum does not support transactional delete.')
             if self.existing_table_handling == ExistingTableHandling.DROP_AND_RECREATE:
-                schema_and_table: str = quote_schema_and_table(self.db,
+                schema_and_table: str = quote_schema_and_table(None,
                                                                self.schema_name,
-                                                               self.table_name)
+                                                               self.table_name,
+                                                               db_engine=self.db_engine)
                 logger.info(f"Dropping external table {schema_and_table}...")
-                with self.db.connect() as cursor:
+                with self.db_engine.connect() as conn:
                     # See below note about fix from Spectrify
-                    cursor.execution_options(isolation_level='AUTOCOMMIT')
-                    cursor.execute(f"DROP TABLE IF EXISTS {schema_and_table}")
+                    conn.execution_options(isolation_level='AUTOCOMMIT')
+                    conn.execute(text(f"DROP TABLE IF EXISTS {schema_and_table}"))
 
             logger.info(f"Deleting files in {self.output_loc}...")
             self.output_loc.purge_directory()
diff --git a/records_mover/records/targets/table/move_from_dataframes_source.py b/records_mover/records/targets/table/move_from_dataframes_source.py
index a288eded1..ee848bcc8 100644
--- a/records_mover/records/targets/table/move_from_dataframes_source.py
+++ b/records_mover/records/targets/table/move_from_dataframes_source.py
@@ -71,21 +71,22 @@ def records_schema(self) -> RecordsSchema:
 
     def load(self, driver: DBDriver) -> int:
         rows_loaded = 0
-        with self.tbl.db_engine.begin() as db:
-            for df in self.dfs_source.dfs:
-                df = purge_unnamed_unused_columns(df)
-                df = self.records_schema.\
-                    assign_dataframe_names(include_index=self.dfs_source.include_index, df=df)
-                df.to_sql(name=self.tbl.table_name,
-                          con=db,
-                          schema=self.tbl.schema_name,
-                          index=self.dfs_source.include_index,
-                          if_exists='append')
-                rows_loaded += len(df.index)
+        with self.tbl.db_engine.connect() as conn:
+            with conn.begin():
+                for df in self.dfs_source.dfs:
+                    df = purge_unnamed_unused_columns(df)
+                    df = self.records_schema.\
+                        assign_dataframe_names(include_index=self.dfs_source.include_index, df=df)
+                    df.to_sql(name=self.tbl.table_name,
+                              con=conn,
+                              schema=self.tbl.schema_name,
+                              index=self.dfs_source.include_index,
+                              if_exists='append')
+                    rows_loaded += len(df.index)
         return rows_loaded
 
     def move_from_dataframes_source_via_insert(self) -> MoveResult:
-        driver = self.tbl.db_driver(self.tbl.db_engine)
+        driver = self.tbl.db_driver(db=None, db_engine=self.tbl.db_engine)
         schema_sql = self.records_schema.to_schema_sql(driver,
                                                        self.tbl.schema_name,
                                                        self.tbl.table_name)
diff --git a/records_mover/records/targets/table/move_from_fileobjs_source.py b/records_mover/records/targets/table/move_from_fileobjs_source.py
index 82c595686..516c5c05d 100644
--- a/records_mover/records/targets/table/move_from_fileobjs_source.py
+++ b/records_mover/records/targets/table/move_from_fileobjs_source.py
@@ -53,16 +53,17 @@ def reset_before_reload(self) -> None:
         self.fileobj.seek(0)
 
     def move(self) -> MoveResult:
-        with self.tbl.db_engine.begin() as db:
-            driver = self.tbl.db_driver(db)
-            schema_obj = self.fileobjs_source.records_schema
-            schema_sql = self.schema_sql_for_load(schema_obj, self.records_format, driver)
-            loader_from_fileobj = driver.loader_from_fileobj()
-            # This is only reached in move() when
-            # records_target.can_move_from_fileobjs_source() is true,
-            # which is only true when .load_from_fileobj() is not None.
-            assert loader_from_fileobj is not None
-            load_exception = loader_from_fileobj.load_failure_exception()
+        with self.tbl.db_engine.connect() as db_conn:
+            with db_conn.begin():
+                driver = self.tbl.db_driver(None, db_conn=db_conn)
+                schema_obj = self.fileobjs_source.records_schema
+                schema_sql = self.schema_sql_for_load(schema_obj, self.records_format, driver)
+                loader_from_fileobj = driver.loader_from_fileobj()
+                # This is only reached in move() when
+                # records_target.can_move_from_fileobjs_source() is true,
+                # which is only true when .load_from_fileobj() is not None.
+                assert loader_from_fileobj is not None
+                load_exception = loader_from_fileobj.load_failure_exception()
 
         return prep_and_load(self.tbl, self.prep, schema_sql, self.load,
                              load_exception,
diff --git a/records_mover/records/targets/table/move_from_records_directory.py b/records_mover/records/targets/table/move_from_records_directory.py
index f4e8034f7..2765d9235 100644
--- a/records_mover/records/targets/table/move_from_records_directory.py
+++ b/records_mover/records/targets/table/move_from_records_directory.py
@@ -61,14 +61,15 @@ def load(self, driver: DBDriver) -> Optional[int]:
     def move(self) -> MoveResult:
         logger.info("Connecting to database...")
 
-        with self.tbl.db_engine.begin() as db:
-            driver = self.tbl.db_driver(db)
-            loader = driver.loader()
-            # If we've gotten here, .can_move_from_format() has
-            # returned True in the move() method, and that can only happen
-            # if we have a valid loader.
-            assert loader is not None
-            load_exception_type = loader.load_failure_exception()
-            schema_sql = self.load_schema_sql(driver)
+        with self.tbl.db_engine.connect() as db_conn:
+            with db_conn.begin():
+                driver = self.tbl.db_driver(None, db_conn=db_conn)
+                loader = driver.loader()
+                # If we've gotten here, .can_move_from_format() has
+                # returned True in the move() method, and that can only happen
+                # if we have a valid loader.
+                assert loader is not None
+                load_exception_type = loader.load_failure_exception()
+                schema_sql = self.load_schema_sql(driver)
         return prep_and_load(self.tbl, self.prep, schema_sql, self.load,
                              load_exception_type)
diff --git a/records_mover/records/targets/table/move_from_temp_loc_after_filling_it.py b/records_mover/records/targets/table/move_from_temp_loc_after_filling_it.py
index fc725d96a..be7b51875 100644
--- a/records_mover/records/targets/table/move_from_temp_loc_after_filling_it.py
+++ b/records_mover/records/targets/table/move_from_temp_loc_after_filling_it.py
@@ -27,7 +27,7 @@ def __init__(self,
 
     @contextmanager
     def temporary_loadable_directory_loc(self) -> Iterator[BaseDirectoryUrl]:
-        driver = self.tbl.db_driver(self.tbl.db_engine)
+        driver = self.tbl.db_driver(db=None, db_engine=self.tbl.db_engine)
         loader = driver.loader()
         # This will only be reached in move() if
         # Source#has_compatible_format(records_target) returns true,
diff --git a/records_mover/records/targets/table/target.py b/records_mover/records/targets/table/target.py
index 87711c93d..da79c0f91 100644
--- a/records_mover/records/targets/table/target.py
+++ b/records_mover/records/targets/table/target.py
@@ -39,16 +39,20 @@ def __init__(self,
                  schema_name: str,
                  table_name: str,
                  db_engine: Engine,
-                 db_driver: Callable[[Union[Engine, Connection]], DBDriver],
+                 db_driver: Callable[[Optional[Union['Engine', 'Connection']],
+                                      Optional[Connection],
+                                      Optional[Engine]], DBDriver],
                  add_user_perms_for: Optional[Dict[str, List[str]]] = None,
                  add_group_perms_for: Optional[Dict[str, List[str]]] = None,
                  existing_table_handling: ExistingTableHandling =
                  ExistingTableHandling.DELETE_AND_OVERWRITE,
-                 drop_and_recreate_on_load_error: bool = False) -> None:
+                 drop_and_recreate_on_load_error: bool = False,
+                 db_conn: Optional[Connection] = None) -> None:
         self.schema_name = schema_name
         self.table_name = table_name
         self.db_driver = db_driver  # type: ignore
         self.db_engine = db_engine
+        self.db_conn = db_conn
         self.add_user_perms_for = add_user_perms_for
         self.add_group_perms_for = add_group_perms_for
         self.existing_table_handling = existing_table_handling
@@ -80,12 +84,16 @@ def move_from_fileobjs_source(self,
                                         processing_instructions).move()
 
     def can_move_from_fileobjs_source(self) -> bool:
-        driver = self.db_driver(self.db_engine)
+        driver = self.db_driver(None,
+                                db_engine=self.db_engine,
+                                db_conn=self.db_conn)
         loader = driver.loader_from_fileobj()
         return loader is not None
 
     def can_move_directly_from_scheme(self, scheme: str) -> bool:
-        driver = self.db_driver(self.db_engine)
+        driver = self.db_driver(None,
+                                db_engine=self.db_engine,
+                                db_conn=self.db_conn)
         loader = driver.loader()
         if loader is None:
             # can't bulk load at all, so can't load direct!
@@ -94,7 +102,9 @@ def can_move_directly_from_scheme(self, scheme: str) -> bool:
         return loader.best_scheme_to_load_from() == scheme
 
     def known_supported_records_formats(self) -> List[BaseRecordsFormat]:
-        driver = self.db_driver(self.db_engine)
+        driver = self.db_driver(None,
+                                db_engine=self.db_engine,
+                                db_conn=self.db_conn)
         loader = driver.loader()
         if loader is None:
             logger.warning(f"No loader configured for this database type ({self.db_engine.name})")
@@ -105,7 +115,9 @@ def can_move_from_format(self,
                              source_records_format: BaseRecordsFormat) -> bool:
         """Return true if writing the specified format satisfies our format
         needs"""
-        driver = self.db_driver(self.db_engine)
+        driver = self.db_driver(None,
+                                db_engine=self.db_engine,
+                                db_conn=self.db_conn)
         loader = driver.loader()
         if loader is None:
             logger.warning(f"No loader configured for this database type ({self.db_engine.name})")
@@ -113,7 +125,9 @@ def can_move_from_format(self,
         return loader.can_load_this_format(source_records_format)
 
     def can_move_from_temp_loc_after_filling_it(self) -> bool:
-        driver = self.db_driver(self.db_engine)
+        driver = self.db_driver(None,
+                                db_engine=self.db_engine,
+                                db_conn=self.db_conn)
         loader = driver.loader()
         if loader is None:
             logger.warning(f"No loader configured for this database type ({self.db_engine.name})")
@@ -130,7 +144,9 @@ def can_move_from_temp_loc_after_filling_it(self) -> bool:
         return has_scratch_location
 
     def temporary_loadable_directory_scheme(self) -> str:
-        driver = self.db_driver(self.db_engine)
+        driver = self.db_driver(None,
+                                db_engine=self.db_engine,
+                                db_conn=self.db_conn)
         loader = driver.loader()
         if loader is None:
             raise TypeError("Please check can_move_from_temp_loc_after_filling_it() "
diff --git a/records_mover/session.py b/records_mover/session.py
index 69ef66ac6..f54b8a303 100644
--- a/records_mover/session.py
+++ b/records_mover/session.py
@@ -261,7 +261,10 @@ def get_db_engine(self,
         db_facts = creds_provider.db_facts(db_creds_name)
         return engine_from_db_facts(db_facts)
 
-    def db_driver(self, db: Union['Engine', 'Connection']) -> 'DBDriver':
+    def db_driver(self,
+                  db: Optional[Union['Engine', 'Connection']],
+                  db_conn: Optional['Connection'] = None,
+                  db_engine: Optional['Engine'] = None) -> 'DBDriver':
         from .db.factory import db_driver
 
         kwargs = {}
@@ -282,6 +285,8 @@ def db_driver(self, db: Union['Engine', 'Connection']) -> 'DBDriver':
                 logger.debug('google.cloud.storage not installed', exc_info=True)
 
         return db_driver(db=db,
+                         db_conn=db_conn,
+                         db_engine=db_engine,
                          url_resolver=self.url_resolver,
                          **kwargs)
 
diff --git a/records_mover/utils/json_schema.py b/records_mover/utils/json_schema.py
index 48c0e0b91..3a2d1258a 100644
--- a/records_mover/utils/json_schema.py
+++ b/records_mover/utils/json_schema.py
@@ -51,8 +51,13 @@ def is_optional_type(tp: PythonType) -> bool:
 
 def is_impractical_type(python_type: PythonType) -> bool:
     # can't write code and pass it around in JSON!
+    # we would on the other hand like to support passing around connections
+    # hence allowing for typing.ForwardRef, which is used to allow ForwardRef('Connection')
+    # to be passed around
     return (is_callable_type(python_type) or
-            type(python_type) == enum.EnumMeta)
+            isinstance(python_type, enum.EnumMeta) or
+            isinstance(python_type, typing.ForwardRef)
+            )
 
 
 def parse_python_parameter_type(name: str,
diff --git a/setup.py b/setup.py
index 1bd5a8a80..aa7cb45a4 100755
--- a/setup.py
+++ b/setup.py
@@ -130,6 +130,7 @@ def initialize_options(self) -> None:
     'jsonschema',  # needed for directory_validator.py
     'pytz',
     'wheel',  # needed to support legacy 'setup.py install'
+    'parameterized',
 ] + (
     pytest_dependencies +
     # needed for records_database_fixture retrying drop/creates on
@@ -144,7 +145,7 @@ def initialize_options(self) -> None:
 ]
 
 db_dependencies = [
-    'sqlalchemy>=1.4,<2.0',
+    'sqlalchemy>=1.4',
 ]
 
 smart_open_dependencies = [
@@ -196,7 +197,7 @@ def initialize_options(self) -> None:
 
 redshift_dependencies_base = [
     # sqlalchemy-redshift 0.7.7 introduced support for Parquet in UNLOAD
-    'sqlalchemy-redshift>=0.7.7,<0.8.13',
+    'sqlalchemy-redshift>=0.7.7',
 ] + aws_dependencies + db_dependencies
 
 redshift_dependencies_binary = [
@@ -224,10 +225,7 @@ def initialize_options(self) -> None:
 ]
 
 vertica_dependencies = [
-    # sqlalchemy-vertica-python 0.5.5 introduced
-    # https://github.com/bluelabsio/sqlalchemy-vertica-python/pull/7
-    # which fixed a bug pulling schema information from Vertica
-    'sqlalchemy-vertica-python>=0.5.5,<0.6',
+    'sqlalchemy-vertica-python>=0.5.5',
 ] + db_dependencies
 
 literally_every_single_database_binary_dependencies = (
@@ -239,7 +237,8 @@ def initialize_options(self) -> None:
 )
 
 typecheck_dependencies = [
-    'mypy<1.5.0',  # https://github.com/python/mypy/issues/15891
+    "mypy@git+https://github.com/python/mypy#88ae1e4c1541e5b03d695cf63d1265b972e427d9"
+    # update to build from mypy without version qualifier soon
     'lxml',  # needed by mypy HTML coverage reporting
     'sqlalchemy-stubs>=0.3',
     'types-pytz',
diff --git a/types/stubs/sqlalchemy_redshift/dialect.pyi b/types/stubs/sqlalchemy_redshift/dialect.pyi
index 088591d68..7f627cc39 100644
--- a/types/stubs/sqlalchemy_redshift/dialect.pyi
+++ b/types/stubs/sqlalchemy_redshift/dialect.pyi
@@ -3,7 +3,6 @@ from .commands import AlterTableAppendCommand as AlterTableAppendCommand, Compre
 from alembic.ddl import postgresql
 from sqlalchemy.dialects.postgresql import DOUBLE_PRECISION as DOUBLE_PRECISION  # noqa
 from sqlalchemy.dialects.postgresql.base import (PGCompiler,
-                                                 PGDDLCompiler,
                                                  PGIdentifierPreparer,
                                                  PGTypeCompiler)
 from sqlalchemy.dialects.postgresql.psycopg2 import PGDialect_psycopg2
@@ -29,11 +28,6 @@ class RedshiftCompiler(PGCompiler):
     def visit_now_func(self, fn: Any, **kw: Any): ...
 
 
-class RedshiftDDLCompiler(PGDDLCompiler):
-    def post_create_table(self, table: Any): ...
-    def get_column_specification(self, column: Any, **kwargs: Any): ...
-
-
 class RedshiftTypeCompiler(PGTypeCompiler):
     def visit_TIMESTAMPTZ(self, type_: Any, **kw: Any): ...
 


Test patch:
diff --git a/tests/component/db/mysql/test_mysql_db_driver.py b/tests/component/db/mysql/test_mysql_db_driver.py
index 3863ff493..aaa5a7679 100644
--- a/tests/component/db/mysql/test_mysql_db_driver.py
+++ b/tests/component/db/mysql/test_mysql_db_driver.py
@@ -9,7 +9,7 @@ def setUp(self):
         self.mock_db_engine = MagicMock(name='db_engine')
         self.mock_url_resolver = Mock(name='url_resolver')
         self.mock_db_engine.engine = self.mock_db_engine
-        self.mysql_db_driver = MySQLDBDriver(db=self.mock_db_engine,
+        self.mysql_db_driver = MySQLDBDriver(db_engine=self.mock_db_engine,
                                              url_resolver=self.mock_url_resolver)
 
     def test_integer_limits(self):
diff --git a/tests/component/db/mysql/test_mysql_load_options_known.py b/tests/component/db/mysql/test_mysql_load_options_known.py
index 21a36d64e..3ab0264bd 100644
--- a/tests/component/db/mysql/test_mysql_load_options_known.py
+++ b/tests/component/db/mysql/test_mysql_load_options_known.py
@@ -9,7 +9,8 @@ class TestMySQLLoadOptionsKnown(unittest.TestCase):
     def test_load_known_formats(self):
         mock_db = Mock(name='db')
         mock_url_resolver = Mock(name='url_resolver')
-        loader = MySQLLoader(db=mock_db,
+        loader = MySQLLoader(None,
+                             db_conn=mock_db,
                              url_resolver=mock_url_resolver)
         known_load_formats = loader.known_supported_records_formats_for_load()
         for records_format in known_load_formats:
diff --git a/tests/component/db/postgres/test_postgres_copy_from_options_load_known.py b/tests/component/db/postgres/test_postgres_copy_from_options_load_known.py
index 80bd2d259..7b22bd153 100644
--- a/tests/component/db/postgres/test_postgres_copy_from_options_load_known.py
+++ b/tests/component/db/postgres/test_postgres_copy_from_options_load_known.py
@@ -13,7 +13,8 @@ def test_load_known_formats(self):
         mock_db = Mock(name='db')
         loader = PostgresLoader(url_resolver=mock_url_resolver,
                                 meta=mock_meta,
-                                db=mock_db)
+                                db=None,
+                                db_conn=mock_db)
         known_load_formats = loader.known_supported_records_formats_for_load()
         for records_format in known_load_formats:
             unhandled_hints = set(records_format.hints)
diff --git a/tests/component/db/postgres/test_postgres_copy_from_options_unload_known.py b/tests/component/db/postgres/test_postgres_copy_from_options_unload_known.py
index 44eace1e8..f4b451345 100644
--- a/tests/component/db/postgres/test_postgres_copy_from_options_unload_known.py
+++ b/tests/component/db/postgres/test_postgres_copy_from_options_unload_known.py
@@ -7,7 +7,7 @@
 class TestPostgresCopyOptionsUnloadKnown(unittest.TestCase):
     def test_unload_known_formats(self):
         mock_db = Mock(name='db')
-        loader = PostgresUnloader(db=mock_db)
+        loader = PostgresUnloader(None, db_conn=mock_db)
         known_unload_formats = loader.known_supported_records_formats_for_unload()
         for records_format in known_unload_formats:
             unhandled_hints = set(records_format.hints)
diff --git a/tests/component/db/postgres/test_postgres_db_driver.py b/tests/component/db/postgres/test_postgres_db_driver.py
index c63996406..78db2e28a 100644
--- a/tests/component/db/postgres/test_postgres_db_driver.py
+++ b/tests/component/db/postgres/test_postgres_db_driver.py
@@ -9,7 +9,8 @@ def setUp(self):
         self.mock_db_engine = MagicMock(name='db_engine')
         self.mock_url_resolver = Mock(name='url_resolver')
         self.mock_db_engine.engine = self.mock_db_engine
-        self.postgres_db_driver = PostgresDBDriver(db=self.mock_db_engine,
+        self.postgres_db_driver = PostgresDBDriver(None,
+                                                   db_engine=self.mock_db_engine,
                                                    url_resolver=self.mock_url_resolver)
 
     def test_integer_limits(self):
diff --git a/tests/integration/records/mover_test_case.py b/tests/integration/records/mover_test_case.py
index df1795013..1ae9beba9 100644
--- a/tests/integration/records/mover_test_case.py
+++ b/tests/integration/records/mover_test_case.py
@@ -9,7 +9,7 @@ def __init__(self,
                  source_db_engine: Optional[Engine] = None,
                  file_variant: Optional[DelimitedVariant] = None) -> None:
         """
-        :param db_engine: Target database of the records move as a SQLAlchemy Engine object.
+        :param target_db_engine: Target database of the records move as a SQLAlchemy Engine object.
 
         :param source_data_db_engine: Source database of the records
         move.  None if we are loading from a file or a dataframe
@@ -27,7 +27,7 @@ def database_has_no_usable_timestamptz_type(self, engine: Engine) -> bool:
         # timezone is stripped off without looking at it before the
         # timestamp itself is stored without any modifications to the
         # hour number.
-        return engine.name == 'mysql'
+        return engine.name == 'mysql' or engine.name is None
 
     def database_default_store_timezone_is_us_eastern(self) -> bool:
         """
diff --git a/tests/component/records/test_dataframe_schema_sql_creation.py b/tests/integration/records/multi_db/test_dataframe_schema_sql_creation.py
similarity index 76%
rename from tests/component/records/test_dataframe_schema_sql_creation.py
rename to tests/integration/records/multi_db/test_dataframe_schema_sql_creation.py
index 0ba5a2bc0..8ce7bd9a9 100644
--- a/tests/component/records/test_dataframe_schema_sql_creation.py
+++ b/tests/integration/records/multi_db/test_dataframe_schema_sql_creation.py
@@ -1,10 +1,10 @@
+import unittest
+from pandas import DataFrame
 from records_mover.records.sources.dataframes import DataframesRecordsSource
 from records_mover.db.redshift.redshift_db_driver import RedshiftDBDriver
 from records_mover.records.processing_instructions import ProcessingInstructions
-from sqlalchemy_redshift.dialect import RedshiftDialect
-import unittest
-from unittest.mock import Mock
-from pandas import DataFrame
+from records_mover import Session, set_stream_logging
+import logging
 
 
 class TestDataframeSchemaSqlCreation(unittest.TestCase):
@@ -15,6 +15,8 @@ def test_dataframe_to_int64_and_back_to_object_produces_int_columns(self) -> Non
         # limitation:
         #
         # https://github.com/bluelabsio/records-mover/pull/103
+        session = Session()
+        db_engine = session.get_db_engine('demo-itest')
         data = {'Population': [11190846, 1303171035, 207847528]}
         df = DataFrame(data, columns=['Population'])
 
@@ -24,10 +26,7 @@ def test_dataframe_to_int64_and_back_to_object_produces_int_columns(self) -> Non
         source = DataframesRecordsSource(dfs=[df])
         processing_instructions = ProcessingInstructions()
         schema = source.initial_records_schema(processing_instructions)
-        dialect = RedshiftDialect()
-        mock_engine = Mock(name='engine')
-        mock_engine.dialect = dialect
-        driver = RedshiftDBDriver(db=mock_engine)
+        driver = RedshiftDBDriver(None, db_engine=db_engine)
         schema_sql = schema.to_schema_sql(driver=driver,
                                           schema_name='my_schema_name',
                                           table_name='my_table_name')
@@ -38,3 +37,12 @@ def test_dataframe_to_int64_and_back_to_object_produces_int_columns(self) -> Non
 
 """
         self.assertEqual(schema_sql, expected_schema_sql)
+
+
+if __name__ == '__main__':
+    set_stream_logging(level=logging.DEBUG)
+    logging.getLogger('botocore').setLevel(logging.INFO)
+    logging.getLogger('boto3').setLevel(logging.INFO)
+    logging.getLogger('urllib3').setLevel(logging.INFO)
+
+    unittest.main()
diff --git a/tests/integration/records/multi_db/test_records_table2table.py b/tests/integration/records/multi_db/test_records_table2table.py
index 2c9a528a7..542946080 100644
--- a/tests/integration/records/multi_db/test_records_table2table.py
+++ b/tests/integration/records/multi_db/test_records_table2table.py
@@ -1,5 +1,7 @@
+# flake8: noqa
+
 from records_mover.db.quoting import quote_schema_and_table
-from records_mover import Session, set_stream_logging
+from records_mover import Session
 from records_mover.records import ExistingTableHandling
 import logging
 import time
@@ -8,6 +10,8 @@
 from ..records_database_fixture import RecordsDatabaseFixture
 from ..table_validator import RecordsTableValidator
 from ..purge_old_test_tables import purge_old_tables
+from parameterized import parameterized  # type: ignore[import-not-found]
+
 
 logger = logging.getLogger(__name__)
 
@@ -42,6 +46,14 @@ def schema_name(db_name):
         raise NotImplementedError('Teach me how to determine a schema name for ' + db_name)
 
 
+def name_func(testcase_func, param_num, param):
+    source, target = param[0]
+    return f"test_move_and_verify_from_{source}_to_{target}"
+
+
+SOURCE_TARGET_PAIRS = [(source, target) for source in DB_TYPES for target in DB_TYPES]
+
+
 class RecordsMoverTable2TableIntegrationTest(unittest.TestCase):
     #
     # actual test methods are defined dynamically after the class
@@ -63,60 +75,47 @@ def move_and_verify(self, source_dbname: str, target_dbname: str) -> None:
         sources = records.sources
         source_engine = session.get_db_engine(source_dbname)
         target_engine = session.get_db_engine(target_dbname)
-        source_schema_name = schema_name(source_dbname)
-        target_schema_name = schema_name(target_dbname)
-        source_table_name = f'itest_source_{BUILD_NUM}_{CURRENT_EPOCH}'
-        records_database_fixture = RecordsDatabaseFixture(source_engine,
-                                                          source_schema_name,
-                                                          source_table_name)
+        with source_engine.connect() as source_conn:
+            with target_engine.connect() as target_conn:
+                source_schema_name = schema_name(source_dbname)
+                target_schema_name = schema_name(target_dbname)
+                source_table_name = f'itest_source_{BUILD_NUM}_{CURRENT_EPOCH}'
+                records_database_fixture = RecordsDatabaseFixture(source_engine,
+                                                                  source_schema_name,
+                                                                  source_table_name)
+                records_database_fixture.tear_down()
+                records_database_fixture.bring_up()
+
+                existing = ExistingTableHandling.DROP_AND_RECREATE
+                source = sources.table(schema_name=source_schema_name,
+                                    table_name=source_table_name,
+                                    db_engine=source_engine,
+                                    db_conn=source_conn)
+                target = targets.table(schema_name=target_schema_name,
+                                    table_name=TARGET_TABLE_NAME,
+                                    db_engine=target_engine,
+                                    existing_table_handling=existing,
+                                    db_conn=target_conn)
+                out = records.move(source, target)
+                # redshift doesn't give reliable info on load results, so this
+                # will be None or 1
+                self.assertNotEqual(0, out.move_count)
+                validator = RecordsTableValidator(target_engine,
+                                                source_db_engine=source_engine)
+                validator.validate(schema_name=target_schema_name,
+                                table_name=TARGET_TABLE_NAME)
+
+                quoted_target = quote_schema_and_table(None, target_schema_name, TARGET_TABLE_NAME,
+                                                    db_engine=target_engine)
+                sql = f"DROP TABLE {quoted_target}"
+                with target_conn.begin():
+                    target_conn.exec_driver_sql(sql)
         records_database_fixture.tear_down()
-        records_database_fixture.bring_up()
-
-        existing = ExistingTableHandling.DROP_AND_RECREATE
-        source = sources.table(schema_name=source_schema_name,
-                               table_name=source_table_name,
-                               db_engine=source_engine)
-        target = targets.table(schema_name=target_schema_name,
-                               table_name=TARGET_TABLE_NAME,
-                               db_engine=target_engine,
-                               existing_table_handling=existing)
-        out = records.move(source, target)
-        # redshift doesn't give reliable info on load results, so this
-        # will be None or 1
-        self.assertNotEqual(0, out.move_count)
-        validator = RecordsTableValidator(target_engine,
-                                          source_db_engine=source_engine)
-        validator.validate(schema_name=target_schema_name,
-                           table_name=TARGET_TABLE_NAME)
-
-        quoted_target = quote_schema_and_table(target_engine, target_schema_name, TARGET_TABLE_NAME)
-        sql = f"DROP TABLE {quoted_target}"
-        target_engine.execute(sql)
 
-        records_database_fixture.tear_down()
 
-
-def create_test_func(source_name, target_name):
-    def source2target(self):
+    @parameterized.expand(SOURCE_TARGET_PAIRS, name_func=name_func)
+    def test_move_and_verify(self, source, target):
+        source_name = DB_NAMES[source]
+        target_name = DB_NAMES[target]
+        print(f"Moving from {source_name} to {target_name}")
         self.move_and_verify(source_name, target_name)
-    return source2target
-
-
-if __name__ == '__main__':
-    set_stream_logging(level=logging.DEBUG)
-    logging.getLogger('botocore').setLevel(logging.INFO)
-    logging.getLogger('boto3').setLevel(logging.INFO)
-    logging.getLogger('urllib3').setLevel(logging.INFO)
-
-    for source in DB_TYPES:
-        for target in DB_TYPES:
-            source_name = DB_NAMES[source]
-            target_name = DB_NAMES[target]
-            f = create_test_func(source_name, target_name)
-            func_name = f"test_{source}2{target}"
-
-            setattr(RecordsMoverTable2TableIntegrationTest,
-                    func_name,
-                    f)
-
-    unittest.main()
diff --git a/tests/integration/records/purge_old_test_tables.py b/tests/integration/records/purge_old_test_tables.py
index 76577e71a..cc7caa759 100755
--- a/tests/integration/records/purge_old_test_tables.py
+++ b/tests/integration/records/purge_old_test_tables.py
@@ -35,9 +35,13 @@ def purge_old_tables(schema_name: str, table_name_prefix: str,
     ]
     print(f"Tables to purge matching {schema_name}.{table_name_prefix}_: {purgable_table_names}")
     for table_name in purgable_table_names:
-        sql = f"DROP TABLE {quote_schema_and_table(db_engine, schema_name, table_name)}"
+        sql = (
+            "DROP TABLE "
+            f"{quote_schema_and_table(None, schema_name, table_name, db_engine=db_engine)}")
         print(sql)
-        db_engine.execute(sql)
+        with db_engine.connect() as connection:
+            with connection.begin():
+                connection.exec_driver_sql(sql)
 
 
 if __name__ == '__main__':
diff --git a/tests/integration/records/records_database_fixture.py b/tests/integration/records/records_database_fixture.py
index 6c10e1b56..685d3e0bd 100644
--- a/tests/integration/records/records_database_fixture.py
+++ b/tests/integration/records/records_database_fixture.py
@@ -1,5 +1,6 @@
 from records_mover.db.quoting import quote_schema_and_table
 from records_mover.utils.retry import bigquery_retry
+from sqlalchemy import text
 import logging
 
 logger = logging.getLogger(__name__)
@@ -7,7 +8,10 @@
 
 class RecordsDatabaseFixture:
     def quote_schema_and_table(self, schema, table):
-        return quote_schema_and_table(self.engine, schema, table)
+        return quote_schema_and_table(None,
+                                      schema=schema,
+                                      table=table,
+                                      db_engine=self.engine)
 
     def __init__(self, db_engine, schema_name, table_name):
         self.engine = db_engine
@@ -17,7 +21,9 @@ def __init__(self, db_engine, schema_name, table_name):
     @bigquery_retry()
     def drop_table_if_exists(self, schema, table):
         sql = f"DROP TABLE IF EXISTS {self.quote_schema_and_table(schema, table)}"
-        self.engine.execute(sql)
+        with self.engine.connect() as connection:
+            with connection.begin():
+                connection.execute(text(sql))
 
     def tear_down(self):
         self.drop_table_if_exists(self.schema_name, f"{self.table_name}_frozen")
@@ -102,4 +108,6 @@ def bring_up(self):
 """  # noqa
         else:
             raise NotImplementedError(f"Please teach me how to integration test {self.engine.name}")
-        self.engine.execute(create_tables)
+        with self.engine.connect() as connection:
+            with connection.begin():
+                connection.exec_driver_sql(create_tables)
diff --git a/tests/integration/records/records_datetime_fixture.py b/tests/integration/records/records_datetime_fixture.py
index 6774ca62e..668ffd421 100644
--- a/tests/integration/records/records_datetime_fixture.py
+++ b/tests/integration/records/records_datetime_fixture.py
@@ -1,28 +1,40 @@
+# flake8: noqa
+
 from records_mover.db.quoting import quote_schema_and_table
 from records_mover.utils.retry import bigquery_retry
 from .datetime_cases import (
     SAMPLE_YEAR, SAMPLE_MONTH, SAMPLE_DAY,
     SAMPLE_HOUR, SAMPLE_MINUTE, SAMPLE_SECOND, SAMPLE_OFFSET, SAMPLE_LONG_TZ
 )
-from sqlalchemy.engine import Engine
+from sqlalchemy import text
+from sqlalchemy.engine import Engine, Connection
+from typing import Optional
+
 import logging
 
 logger = logging.getLogger(__name__)
 
 
 class RecordsDatetimeFixture:
-    def __init__(self, engine: Engine, schema_name: str, table_name: str):
+    def __init__(self, engine: Engine, schema_name: str, table_name: str,
+                 connection: Optional[Connection] = None):
         self.engine = engine
         self.schema_name = schema_name
         self.table_name = table_name
+        self.connection = connection
 
     def quote_schema_and_table(self, schema, table):
-        return quote_schema_and_table(self.engine, schema, table)
+        return quote_schema_and_table(None, schema, table, db_engine=self.engine)
 
     @bigquery_retry()
     def drop_table_if_exists(self, schema, table):
         sql = f"DROP TABLE IF EXISTS {self.quote_schema_and_table(schema, table)}"
-        self.engine.execute(sql)
+        if not self.connection:
+            with self.engine.connect() as connection:
+                with connection.begin():
+                    connection.execute(text(sql))
+        else:
+            self.connection.execute(text(sql))
 
     def createDateTimeTzTable(self) -> None:
         if self.engine.name == 'redshift':
@@ -52,7 +64,9 @@ def createDateTimeTzTable(self) -> None:
 """  # noqa
         else:
             raise NotImplementedError(f"Please teach me how to integration test {self.engine.name}")
-        self.engine.execute(create_tables)
+        with self.engine.connect() as connection:
+            with connection.begin():
+                connection.exec_driver_sql(create_tables)
 
     def createDateTimeTable(self) -> None:
         if self.engine.name == 'redshift':
@@ -82,7 +96,13 @@ def createDateTimeTable(self) -> None:
 """  # noqa
         else:
             raise NotImplementedError(f"Please teach me how to integration test {self.engine.name}")
-        self.engine.execute(create_tables)
+        if not self.connection:
+            with self.engine.connect() as connection:
+                with connection.begin():
+                    connection.exec_driver_sql(create_tables)
+        else:
+            with self.connection.begin():
+                self.connection.exec_driver_sql(create_tables)  # type: ignore[attr-defined]
 
     @bigquery_retry()
     def createDateTable(self) -> None:
@@ -113,7 +133,13 @@ def createDateTable(self) -> None:
 """  # noqa
         else:
             raise NotImplementedError(f"Please teach me how to integration test {self.engine.name}")
-        self.engine.execute(create_tables)
+        if not self.connection:
+            with self.engine.connect() as connection:
+                with connection.begin():
+                    connection.exec_driver_sql(create_tables)
+        else:
+            with self.connection.begin():
+                self.connection.exec_driver_sql(create_tables)  # type: ignore[attr-defined]
 
     @bigquery_retry()
     def createTimeTable(self):
@@ -144,7 +170,13 @@ def createTimeTable(self):
 """  # noqa
         else:
             raise NotImplementedError(f"Please teach me how to integration test {self.engine.name}")
-        self.engine.execute(create_tables)
+        if not self.connection:
+            with self.engine.connect() as connection:
+                with connection.begin():
+                    connection.exec_driver_sql(create_tables)
+        else:
+            with self.connection.begin():
+                self.connection.exec_driver_sql(create_tables)
 
     def drop_tables(self):
         logger.info('Dropping tables...')
diff --git a/tests/integration/records/records_numeric_database_fixture.py b/tests/integration/records/records_numeric_database_fixture.py
index b8386700f..66c381ba9 100644
--- a/tests/integration/records/records_numeric_database_fixture.py
+++ b/tests/integration/records/records_numeric_database_fixture.py
@@ -1,4 +1,5 @@
 from records_mover.db.quoting import quote_schema_and_table
+from sqlalchemy import text
 
 
 class RecordsNumericDatabaseFixture:
@@ -126,14 +127,19 @@ def bring_up(self):
             raise NotImplementedError(f"Please teach me how to integration test {self.engine.name}")
         print(f"Creating: {create_tables}")
         for statement in create_tables:
-            self.engine.execute(statement)
+            with self.engine.connect() as connection:
+                with connection.begin():
+                    connection.exec_driver_sql(statement)
 
     def quote_schema_and_table(self, schema, table):
-        return quote_schema_and_table(self.engine, schema, table)
+        return quote_schema_and_table(None, schema, table,
+                                      db_engine=self.engine)
 
     def drop_table_if_exists(self, schema, table):
         sql = f"DROP TABLE IF EXISTS {self.quote_schema_and_table(schema, table)}"
-        self.engine.execute(sql)
+        with self.engine.connect() as connection:
+            with connection.begin():
+                connection.execute(text(sql))
 
     def tear_down(self):
         self.drop_table_if_exists(self.schema_name, self.table_name)
diff --git a/tests/integration/records/single_db/base_records_test.py b/tests/integration/records/single_db/base_records_test.py
index 284e1e988..06284a1cd 100644
--- a/tests/integration/records/single_db/base_records_test.py
+++ b/tests/integration/records/single_db/base_records_test.py
@@ -46,7 +46,8 @@ def setUp(self):
                                default_db_creds_name=None,
                                default_aws_creds_name=None)
         self.engine = self.session.get_default_db_engine()
-        self.driver = self.session.db_driver(self.engine)
+        self.connection = self.engine.connect()
+        self.driver = self.session.db_driver(None, db_conn=self.connection, db_engine=self.engine)
         if self.engine.name == 'bigquery':
             self.schema_name = 'bq_itest'
             # avoid per-table rate limits
@@ -70,11 +71,12 @@ def setUp(self):
         self.records = self.session.records
 
     def tearDown(self):
+        self.connection.close()
         self.session = None
         self.fixture.tear_down()
 
     def table(self, schema, table):
-        return Table(table, self.meta, schema=schema, autoload=True, autoload_with=self.engine)
+        return Table(table, self.meta, schema=schema, autoload_with=self.engine)
 
     def variant_has_header(self, variant):
         return variant in ['csv', 'bigquery']
@@ -108,7 +110,8 @@ def unload_column_to_string(self,
         with tempfile.TemporaryDirectory() as directory_name:
             source = sources.table(schema_name=self.schema_name,
                                    table_name=self.table_name,
-                                   db_engine=self.engine)
+                                   db_engine=self.engine,
+                                   db_conn=self.connection)
             directory_url = pathlib.Path(directory_name).as_uri() + '/'
             target = targets.directory_from_url(output_url=directory_url,
                                                 records_format=records_format)
diff --git a/tests/integration/records/single_db/numeric_expectations.py b/tests/integration/records/single_db/numeric_expectations.py
index 1bf77afb3..92d23f595 100644
--- a/tests/integration/records/single_db/numeric_expectations.py
+++ b/tests/integration/records/single_db/numeric_expectations.py
@@ -180,8 +180,8 @@
         'uint64': 'NUMERIC(20, 0)',
         'float16': 'REAL',
         'float32': 'REAL',
-        'float64': 'DOUBLE_PRECISION',
-        'float128': 'DOUBLE_PRECISION',  # Postgres doesn't support >float64
+        'float64': 'DOUBLE PRECISION',
+        'float128': 'DOUBLE PRECISION',  # Postgres doesn't support >float64
         'fixed_6_2': 'NUMERIC(6, 2)',
         'fixed_38_9': 'NUMERIC(38, 9)',
         'fixed_100_4': 'NUMERIC(100, 4)',
diff --git a/tests/integration/records/single_db/test_records_load_datetime.py b/tests/integration/records/single_db/test_records_load_datetime.py
index 951a7a3ea..b00b0e1ac 100644
--- a/tests/integration/records/single_db/test_records_load_datetime.py
+++ b/tests/integration/records/single_db/test_records_load_datetime.py
@@ -8,6 +8,7 @@
 )
 from records_mover.records import RecordsSchema, RecordsFormat, PartialRecordsHints
 from records_mover.records.schema.field.field_types import FieldType
+from sqlalchemy import text
 
 logger = logging.getLogger(__name__)
 
@@ -58,12 +59,16 @@ def load(self,
 
     def pull_result(self,
                     column_name: str) -> datetime.datetime:
-        out = self.engine.execute(f'SELECT {column_name} '
-                                  f'from {self.schema_name}.{self.table_name}')
-        ret_all = out.fetchall()
+        with self.engine.connect() as connection:
+            with connection.begin():
+                out = connection.execute(text(
+                        f'SELECT {column_name} '
+                        f'from {self.schema_name}.{self.table_name}'))
+                ret_all = out.fetchall()
         assert 1 == len(ret_all)
         ret = ret_all[0]
-        return ret[column_name]
+        ret_mapping = ret._mapping
+        return ret_mapping[column_name]
 
     def database_provides_times_as_timedeltas(self) -> bool:
         return self.engine.name == 'mysql'
diff --git a/tests/integration/records/single_db/test_records_numeric.py b/tests/integration/records/single_db/test_records_numeric.py
index 63b9656e2..731c2cfb8 100644
--- a/tests/integration/records/single_db/test_records_numeric.py
+++ b/tests/integration/records/single_db/test_records_numeric.py
@@ -60,7 +60,8 @@ def test_numeric_schema_fields_created(self) -> None:
             processing_instructions = ProcessingInstructions()
             source = self.records.sources.table(schema_name=self.schema_name,
                                                 table_name=self.table_name,
-                                                db_engine=self.engine)
+                                                db_engine=self.engine,
+                                                db_conn=self.connection)
             target = self.records.targets.directory_from_url(output_url,
                                                              records_format=records_format)
             out = self.records.move(source, target, processing_instructions)
diff --git a/tests/integration/records/single_db/test_records_unload.py b/tests/integration/records/single_db/test_records_unload.py
index cf833037b..9230fb7aa 100644
--- a/tests/integration/records/single_db/test_records_unload.py
+++ b/tests/integration/records/single_db/test_records_unload.py
@@ -74,7 +74,8 @@ def unload(self, variant, directory, hints={}) -> None:
         sources = self.records.sources
         source = sources.table(schema_name=self.schema_name,
                                table_name=self.table_name,
-                               db_engine=self.engine)
+                               db_engine=self.engine,
+                               db_conn=self.connection)
         target = targets.directory_from_url(output_url=directory_url,
                                             records_format=records_format)
         out = self.records.move(source, target)
diff --git a/tests/integration/records/single_db/test_records_unload_datetime.py b/tests/integration/records/single_db/test_records_unload_datetime.py
index 2242ef90a..0c59aee4e 100644
--- a/tests/integration/records/single_db/test_records_unload_datetime.py
+++ b/tests/integration/records/single_db/test_records_unload_datetime.py
@@ -24,11 +24,12 @@ def setUp(self) -> None:
         super().setUp()
         self.datetime_fixture = RecordsDatetimeFixture(engine=self.engine,
                                                        table_name=self.table_name,
-                                                       schema_name=self.schema_name)
+                                                       schema_name=self.schema_name,
+                                                       connection=self.connection)
 
     def tearDown(self):
-        super().tearDown()
         self.datetime_fixture.drop_tables()
+        super().tearDown()
 
     def test_unload_date(self) -> None:
         self.datetime_fixture.createDateTable()
@@ -73,7 +74,7 @@ def test_unload_date(self) -> None:
                                                'dateformat': dateformat,
                                                'compression': None,
                                                'header-row': False,
-                                               **addl_hints,  # type: ignore
+                                               **addl_hints,
             })
             expect_pandas_failure = (not self.has_pandas()) and uses_pandas
             try:
@@ -141,7 +142,7 @@ def test_unload_datetime(self) -> None:
                                                'datetimeformat': datetimeformat,
                                                'compression': None,
                                                'header-row': False,
-                                               **addl_hints,  # type: ignore
+                                               **addl_hints,
             })
             expect_pandas_failure = (not self.has_pandas()) and uses_pandas
             try:
@@ -215,7 +216,7 @@ def test_unload_datetimetz(self) -> None:
                                                'datetimeformattz': datetimeformattz,
                                                'compression': None,
                                                'header-row': False,
-                                               **addl_hints,  # type: ignore
+                                               **addl_hints,
             })
             expect_pandas_failure = (not self.has_pandas()) and uses_pandas
             try:
@@ -284,7 +285,7 @@ def test_unload_timeonly(self) -> None:
                                                'timeonlyformat': timeonlyformat,
                                                'compression': None,
                                                'header-row': False,
-                                               **addl_hints,  # type: ignore
+                                               **addl_hints,
             })
             expect_pandas_failure = (not self.has_pandas()) and uses_pandas
             try:
diff --git a/tests/integration/records/table_timezone_validator.py b/tests/integration/records/table_timezone_validator.py
index d4fdb8dbb..d7b7063e6 100644
--- a/tests/integration/records/table_timezone_validator.py
+++ b/tests/integration/records/table_timezone_validator.py
@@ -41,8 +41,8 @@ def validate(self,
             # doesn't store timezones, the database in question just
             # strips off the timezone and stores the '12'
             utc_hour = 12
-        elif(self.tc.loaded_from_dataframe() and
-             self.tc.database_has_no_usable_timestamptz_type(self.target_db_engine)):
+        elif (self.tc.loaded_from_dataframe() and
+              self.tc.database_has_no_usable_timestamptz_type(self.target_db_engine)):
             #
             # In this case, we correctly tell Pandas that we have are
             # at noon:34 US/Eastern, and tell Pandas to format the
@@ -144,11 +144,17 @@ def validate(self,
             assert timestamptzstr in [
                 f'2000-01-02 {utc_hour}:34:{seconds}.{micros} ',
                 f'2000-01-02 {utc_hour}:34:{seconds}.{micros} UTC',
+                f'2000-01-02 {utc_hour}:34:{seconds} UTC',
                 f'2000-01-02 {utc_hour}:34:{seconds}.{micros}+00'
             ],\
                 (f"translated timestamptzstr was {timestamptzstr} and "
                  f"class is {type(timestamptzstr)} - expected "
-                 f"hour to be {utc_hour}")
+                 f"hour to be {utc_hour}"
+                 " EXPECTED OUTPUT NO USABLE TIMESTAMP TYPE"
+                 f" source_db_engine {self.source_db_engine}:"
+                 f' 2000-01-02 {utc_hour}:34:{seconds}.{micros} '
+                 f' OR 2000-01-02 {utc_hour}:34:{seconds}.{micros} UTC'
+                 f' OR 2000-01-02 {utc_hour}:34:{seconds}.{micros}+00')
         else:
             if micros == '000000':
                 assert timestamptzstr in [
@@ -158,7 +164,12 @@ def validate(self,
                 ],\
                     (f"translated timestamptzstr was {timestamptzstr} and "
                      f"class is {type(timestamptzstr)} - expected "
-                     f"hour to be {utc_hour}")
+                     f"hour to be {utc_hour}"
+                     " EXPECTED OUTPUT MICROS 000000"
+                     f" source_db_engine {self.source_db_engine}:"
+                     f' 2000-01-02 {utc_hour}:34:{seconds} UTC'
+                     f' OR 2000-01-02 {utc_hour}:34:{seconds}.{micros} UTC'
+                     f' OR 2000-01-02 {utc_hour}:34:{seconds}.{micros}+00')
             else:
                 assert timestamptzstr in [
                     f'2000-01-02 {utc_hour}:34:{seconds}.{micros} UTC',
@@ -166,7 +177,11 @@ def validate(self,
                 ],\
                     (f"translated timestamptzstr was {timestamptzstr} and "
                      f"class is {type(timestamptzstr)} - expected "
-                     f"hour to be {utc_hour}")
+                     f"hour to be {utc_hour}"
+                     " EXPECTED OUTPUT USUABLE TIMESTAMP TYPE"
+                     f" source_db_engine {self.source_db_engine}:"
+                     f' OR 2000-01-02 {utc_hour}:34:{seconds}.{micros} UTC'
+                     f' OR 2000-01-02 {utc_hour}:34:{seconds}.{micros}+00')
 
         utc = pytz.timezone('UTC')
         if ((load_variant is not None and
diff --git a/tests/integration/records/table_validator.py b/tests/integration/records/table_validator.py
index a6a6f52d5..2b730466c 100644
--- a/tests/integration/records/table_validator.py
+++ b/tests/integration/records/table_validator.py
@@ -177,79 +177,80 @@ def validate_data_values(self,
                     "formatstr": "%E4Y-%m-%d %H:%M:%E*S",
                 }
             elif self.target_db_engine.name == 'mysql':
-                select_sql = f"""
+                select_sql = text(f"""
                 SELECT num, numstr, comma, doublequote, quotecommaquote, date, `time`,
                        `timestamp`,
-                       DATE_FORMAT(`timestamp`, '%%Y-%%m-%%d %%H:%%i:%%s.%%f') as timestampstr,
+                       DATE_FORMAT(`timestamp`, '%Y-%m-%d %H:%i:%s.%f') as timestampstr,
                        timestamptz,
-                       DATE_FORMAT(timestamptz, '%%Y-%%m-%%d %%H:%%i:%%s.%%f+00') as timestamptzstr
+                       DATE_FORMAT(timestamptz, '%Y-%m-%d %H:%i:%s.%f+00') as timestamptzstr
                 FROM {schema_name}.{table_name}
-                """
+                """)
             elif self.tc.raw_avro_types_written():
                 # no real date/time column types used, so can't cast types
-                select_sql = f"""
+                select_sql = text(f"""
                 SELECT num, numstr, comma, doublequote, quotecommaquote, date, "time",
                        "timestamp",
                        "timestamp" as timestampstr,
                        timestamptz,
                        timestamptz as timestamptzstr
                 FROM {schema_name}.{table_name}
-                """
+                """)
             else:
-                select_sql = f"""
+                select_sql = text(f"""
                 SELECT num, numstr, comma, doublequote, quotecommaquote, date, "time",
                        "timestamp",
                        to_char("timestamp", 'YYYY-MM-DD HH24:MI:SS.US') as timestampstr,
                        timestamptz, to_char(timestamptz,
                                             'YYYY-MM-DD HH24:MI:SS.US TZ') as timestamptzstr
                 FROM {schema_name}.{table_name}
-                """
-            out = connection.execute(select_sql, **params)
+                """)
+            out = connection.execute(select_sql, params)
             ret_all = out.fetchall()
         assert 1 == len(ret_all)
         ret = ret_all[0]
 
-        assert ret['num'] == 123
-        assert ret['numstr'] == '123', ret['numstr']
-        assert ret['comma'] == ','
-        assert ret['doublequote'] == '"'
-        assert ret['quotecommaquote'] == '","'
+        assert ret.num == 123
+        assert ret.numstr == '123', ret.numstr
+        assert ret.comma == ','
+        assert ret.doublequote == '"'
+        assert ret.quotecommaquote == '","'
         if self.tc.raw_avro_types_written():
-            assert ret['date'] == 10957, ret['date']
+            assert ret.date == 10957, ret.date
         else:
-            assert ret['date'] == datetime.date(2000, 1, 1),\
-                f"Expected datetime.date(2000, 1, 1), got {ret['date']}"
+            assert ret.date == datetime.date(2000, 1, 1),\
+                f"Expected datetime.date(2000, 1, 1), got {ret.date}"
 
         if self.tc.raw_avro_types_written():
-            assert ret['time'] == 0, ret['time']
+            assert ret.time == 0, ret.time
         elif self.tc.supports_time_without_date():
             if self.tc.selects_time_types_as_timedelta():
-                assert ret['time'] == datetime.timedelta(0, 0),\
-                    f"Incorrect time: {ret['time']} (of type {type(ret['time'])})"
+                assert ret.time == datetime.timedelta(0, 0),\
+                    f"Incorrect time: {ret.time} (of type {type(ret.time)})"
             else:
-                assert ret['time'] == datetime.time(0, 0),\
-                    f"Incorrect time: {ret['time']} (of type {type(ret['time'])})"
+                assert ret.time == datetime.time(0, 0),\
+                    f"Incorrect time: {ret.time} (of type {type(ret.time)})"
         else:
             # fall back to storing as string
             if load_variant is not None and self.tc.variant_uses_am_pm(load_variant):
-                assert ret['time'] == '12:00 AM', f"time was {ret['time']}"
+                assert ret.time == '12:00 AM', f"time was {ret.time}"
             else:
-                assert ret['time'] == '00:00:00', f"time was {ret['time']}"
+                assert ret.time == '00:00:00', f"time was {ret.time}"
 
         if self.tc.raw_avro_types_written():
-            assert ret['timestamp'] == '2000-01-02T12:34:56.789012'
+            assert ret.timestamp == '2000-01-02T12:34:56.789012'
         elif (((load_variant is not None) and
                self.tc.variant_doesnt_support_seconds(load_variant)) or
               ((self.file_variant is not None) and
               self.tc.variant_doesnt_support_seconds(self.file_variant))):
-            assert ret['timestamp'] ==\
+            assert ret.timestamp ==\
                 datetime.datetime(2000, 1, 2, 12, 34),\
-                f"Found timestamp {ret['timestamp']}"
+                f"Found timestamp {ret.timestamp}"
 
         else:
-            assert (ret['timestamp'] == datetime.datetime(2000, 1, 2, 12, 34, 56, 789012)),\
-                f"ret['timestamp'] was {ret['timestamp']} of type {type(ret['timestamp'])}"
-
-        self.tz_validator.validate(timestampstr=ret['timestampstr'],
-                                   timestamptzstr=ret['timestamptzstr'],
-                                   timestamptz=ret['timestamptz'])
+            assert (ret.timestamp == datetime.datetime(2000, 1, 2, 12, 34, 56, 789012)),\
+                f"ret.timestamp was {ret.timestamp} of type {type(ret.timestamp)}"
+        print("ROW OUTPUT VALUES:", ret)
+        print("ROW OUTPUT AS DICTIONARY:", ret._asdict())
+        self.tz_validator.validate(timestampstr=ret.timestampstr,
+                                   timestamptzstr=ret.timestamptzstr,
+                                   timestamptz=ret.timestamptz)
diff --git a/tests/integration/records/timezone.py b/tests/integration/records/timezone.py
index 55338475c..b3600ba7f 100644
--- a/tests/integration/records/timezone.py
+++ b/tests/integration/records/timezone.py
@@ -1,4 +1,6 @@
 import logging
+from sqlalchemy import text
+
 
 logger = logging.getLogger(__name__)
 
@@ -13,5 +15,5 @@ def set_session_tz(conn):
     # being of the session timezone.
     if conn.engine.name == 'vertica':
         tz = DB_SESSION_TIMEZONE
-        conn.execute('SET TIME ZONE TO :tz', tz=tz)
+        conn.execute(text('SET TIME ZONE TO :tz'), {'tz': tz})
         logger.info(f"Set database session timezone to {tz}")
diff --git a/tests/unit/db/bigquery/test_bigquery_db_driver.py b/tests/unit/db/bigquery/test_bigquery_db_driver.py
index 86acaf653..617830b5b 100644
--- a/tests/unit/db/bigquery/test_bigquery_db_driver.py
+++ b/tests/unit/db/bigquery/test_bigquery_db_driver.py
@@ -14,7 +14,8 @@ def setUp(self, mock_BigQueryLoader):
         self.mock_db_engine = MagicMock(name='db_engine')
         self.mock_url_resolver = Mock(name='url_resolver')
         self.mock_BigQueryLoader = mock_BigQueryLoader
-        self.bigquery_db_driver = BigQueryDBDriver(db=self.mock_db_engine,
+        self.bigquery_db_driver = BigQueryDBDriver(db=None,
+                                                   db_engine=self.mock_db_engine,
                                                    url_resolver=self.mock_url_resolver)
 
     def test_load_implemented(self):
diff --git a/tests/unit/db/bigquery/test_bigquery_loader.py b/tests/unit/db/bigquery/test_bigquery_loader.py
index c38a765bc..e0de2dea3 100644
--- a/tests/unit/db/bigquery/test_bigquery_loader.py
+++ b/tests/unit/db/bigquery/test_bigquery_loader.py
@@ -15,8 +15,8 @@ class TestBigQueryLoader(unittest.TestCase):
     def test_load_with_bad_schema_name(self, mock_load_job_config):
         mock_db = Mock(name='mock_db')
         mock_url_resolver = MagicMock(name='mock_url_resolver')
-        big_query_loader = BigQueryLoader(db=mock_db, url_resolver=mock_url_resolver,
-                                          gcs_temp_base_loc=None)
+        big_query_loader = BigQueryLoader(db=None, url_resolver=mock_url_resolver,
+                                          gcs_temp_base_loc=None, db_conn=mock_db)
         mock_schema = 'my_project.my_dataset.something_invalid'
         mock_table = Mock(name='mock_table')
         mock_load_plan = Mock(name='mock_load_plan')
@@ -38,8 +38,8 @@ def test_load_with_bad_schema_name(self, mock_load_job_config):
     def test_load_with_default_project(self, mock_load_job_config):
         mock_db = Mock(name='mock_db')
         mock_url_resolver = MagicMock(name='mock_url_resolver')
-        big_query_loader = BigQueryLoader(db=mock_db, url_resolver=mock_url_resolver,
-                                          gcs_temp_base_loc=None)
+        big_query_loader = BigQueryLoader(db=None, url_resolver=mock_url_resolver,
+                                          gcs_temp_base_loc=None, db_conn=mock_db)
         mock_schema = 'my_dataset'
         mock_table = 'my_table'
         mock_load_plan = Mock(name='mock_load_plan')
@@ -75,8 +75,8 @@ def test_load(self, mock_load_job_config):
         mock_db = Mock(name='mock_db')
         mock_url_resolver = MagicMock(name='mock_url_resolver')
         mock_gcs_temp_base_loc = None
-        big_query_loader = BigQueryLoader(db=mock_db, url_resolver=mock_url_resolver,
-                                          gcs_temp_base_loc=mock_gcs_temp_base_loc)
+        big_query_loader = BigQueryLoader(db=None, url_resolver=mock_url_resolver,
+                                          gcs_temp_base_loc=mock_gcs_temp_base_loc, db_conn=mock_db)
         mock_schema = 'my_project.my_dataset'
         mock_table = 'mytable'
         mock_load_plan = Mock(name='mock_load_plan')
@@ -111,8 +111,8 @@ def test_load_with_job_failure(self, mock_load_job_config):
         mock_db = Mock(name='mock_db')
         mock_url_resolver = MagicMock(name='mock_url_resolver')
         mock_gcs_temp_base_loc = None
-        big_query_loader = BigQueryLoader(db=mock_db, url_resolver=mock_url_resolver,
-                                          gcs_temp_base_loc=mock_gcs_temp_base_loc)
+        big_query_loader = BigQueryLoader(db=None, url_resolver=mock_url_resolver,
+                                          gcs_temp_base_loc=mock_gcs_temp_base_loc, db_conn=mock_db)
         mock_schema = 'my_project.my_dataset'
         mock_table = 'mytable'
         mock_load_plan = Mock(name='mock_load_plan')
@@ -142,8 +142,8 @@ def test_load_no_table(self, mock_load_job_config):
         mock_db = Mock(name='mock_db')
         mock_url_resolver = MagicMock(name='mock_url_resolver')
         mock_gcs_temp_base_loc = None
-        big_query_loader = BigQueryLoader(db=mock_db, url_resolver=mock_url_resolver,
-                                          gcs_temp_base_loc=mock_gcs_temp_base_loc)
+        big_query_loader = BigQueryLoader(db=None, url_resolver=mock_url_resolver,
+                                          gcs_temp_base_loc=mock_gcs_temp_base_loc, db_conn=mock_db)
         mock_schema = 'my_project.my_dataset'
         mock_table = 'mytable'
         mock_load_plan = Mock(name='mock_load_plan')
@@ -169,8 +169,8 @@ def test_load_from_fileobj_true(self, mock_load_job_config):
         mock_db = Mock(name='mock_db')
         mock_url_resolver = MagicMock(name='mock_url_resolver')
         mock_gcs_temp_base_loc = None
-        big_query_loader = BigQueryLoader(db=mock_db, url_resolver=mock_url_resolver,
-                                          gcs_temp_base_loc=mock_gcs_temp_base_loc)
+        big_query_loader = BigQueryLoader(db=None, url_resolver=mock_url_resolver,
+                                          gcs_temp_base_loc=mock_gcs_temp_base_loc, db_conn=mock_db)
         mock_schema = 'my_project.my_dataset'
         mock_table = 'mytable'
         mock_load_plan = Mock(name='mock_load_plan')
@@ -205,8 +205,8 @@ def test_load_from_fileobj_error(self, mock_load_job_config):
         mock_db = Mock(name='mock_db')
         mock_url_resolver = MagicMock(name='mock_url_resolver')
         mock_gcs_temp_base_loc = None
-        big_query_loader = BigQueryLoader(db=mock_db, url_resolver=mock_url_resolver,
-                                          gcs_temp_base_loc=mock_gcs_temp_base_loc)
+        big_query_loader = BigQueryLoader(db=None, url_resolver=mock_url_resolver,
+                                          gcs_temp_base_loc=mock_gcs_temp_base_loc, db_conn=mock_db)
         mock_schema = 'my_project.my_dataset'
         mock_table = 'mytable'
         mock_load_plan = Mock(name='mock_load_plan')
@@ -241,8 +241,8 @@ def test_load_with_fileobj_fallback(self, mock_load_job_config):
         mock_db = Mock(name='mock_db')
         mock_url_resolver = MagicMock(name='mock_url_resolver')
         mock_gcs_temp_base_loc = None
-        big_query_loader = BigQueryLoader(db=mock_db, url_resolver=mock_url_resolver,
-                                          gcs_temp_base_loc=mock_gcs_temp_base_loc)
+        big_query_loader = BigQueryLoader(db=None, url_resolver=mock_url_resolver,
+                                          gcs_temp_base_loc=mock_gcs_temp_base_loc, db_conn=mock_db)
         mock_schema = 'my_project.my_dataset'
         mock_table = 'mytable'
         mock_load_plan = Mock(name='mock_load_plan')
@@ -282,8 +282,8 @@ def test_load_with_fileobj_fallback(self, mock_load_job_config):
     def test_known_supported_records_formats_for_load(self):
         mock_db = Mock(name='db')
         mock_url_resolver = Mock(name='url_resolver')
-        bigquery_loader = BigQueryLoader(db=mock_db, url_resolver=mock_url_resolver,
-                                         gcs_temp_base_loc=None)
+        bigquery_loader = BigQueryLoader(db=None, url_resolver=mock_url_resolver,
+                                         gcs_temp_base_loc=None, db_conn=mock_db)
         out = bigquery_loader.known_supported_records_formats_for_load()
         self.assertEqual(3, len(out))
         delimited_records_format = out[0]
@@ -297,8 +297,8 @@ def test_known_supported_records_formats_for_load(self):
     def test_temporary_gcs_directory_loc_none(self):
         mock_db = Mock(name='db')
         mock_url_resolver = Mock(name='url_resolver')
-        bigquery_loader = BigQueryLoader(db=mock_db, url_resolver=mock_url_resolver,
-                                         gcs_temp_base_loc=None)
+        bigquery_loader = BigQueryLoader(db=None, url_resolver=mock_url_resolver,
+                                         gcs_temp_base_loc=None, db_conn=mock_db)
         with self.assertRaises(NoTemporaryBucketConfiguration):
             with bigquery_loader.temporary_gcs_directory_loc():
                 pass
@@ -307,8 +307,8 @@ def test_temporary_loadable_directory_loc(self):
         mock_db = Mock(name='db')
         mock_url_resolver = Mock(name='url_resolver')
         mock_gcs_temp_base_loc = MagicMock(name='gcs_temp_base_loc')
-        bigquery_loader = BigQueryLoader(db=mock_db, url_resolver=mock_url_resolver,
-                                         gcs_temp_base_loc=mock_gcs_temp_base_loc)
+        bigquery_loader = BigQueryLoader(db=None, url_resolver=mock_url_resolver,
+                                         gcs_temp_base_loc=mock_gcs_temp_base_loc, db_conn=mock_db)
         with bigquery_loader.temporary_loadable_directory_loc() as loc:
             self.assertEqual(loc,
                              mock_gcs_temp_base_loc.temporary_directory.return_value.__enter__.
@@ -318,8 +318,8 @@ def test_temporary_gcs_directory_loc(self):
         mock_db = Mock(name='db')
         mock_url_resolver = Mock(name='url_resolver')
         mock_gcs_temp_base_loc = MagicMock(name='gcs_temp_base_loc')
-        bigquery_loader = BigQueryLoader(db=mock_db, url_resolver=mock_url_resolver,
-                                         gcs_temp_base_loc=mock_gcs_temp_base_loc)
+        bigquery_loader = BigQueryLoader(db=None, url_resolver=mock_url_resolver,
+                                         gcs_temp_base_loc=mock_gcs_temp_base_loc, db_conn=mock_db)
         with bigquery_loader.temporary_gcs_directory_loc() as loc:
             self.assertEqual(loc,
                              mock_gcs_temp_base_loc.temporary_directory.return_value.__enter__.
@@ -329,22 +329,22 @@ def test_has_temporary_loadable_directory_loc_true(self):
         mock_db = Mock(name='db')
         mock_url_resolver = Mock(name='url_resolver')
         mock_gcs_temp_base_loc = MagicMock(name='gcs_temp_base_loc')
-        bigquery_loader = BigQueryLoader(db=mock_db, url_resolver=mock_url_resolver,
-                                         gcs_temp_base_loc=mock_gcs_temp_base_loc)
+        bigquery_loader = BigQueryLoader(db=None, url_resolver=mock_url_resolver,
+                                         gcs_temp_base_loc=mock_gcs_temp_base_loc, db_conn=mock_db)
         self.assertTrue(bigquery_loader.has_temporary_loadable_directory_loc())
 
     def test_temporary_loadable_directory_scheme(self):
         mock_db = Mock(name='db')
         mock_url_resolver = Mock(name='url_resolver')
         mock_gcs_temp_base_loc = MagicMock(name='gcs_temp_base_loc')
-        bigquery_loader = BigQueryLoader(db=mock_db, url_resolver=mock_url_resolver,
-                                         gcs_temp_base_loc=mock_gcs_temp_base_loc)
+        bigquery_loader = BigQueryLoader(db=None, url_resolver=mock_url_resolver,
+                                         gcs_temp_base_loc=mock_gcs_temp_base_loc, db_conn=mock_db)
         self.assertEqual('gs', bigquery_loader.temporary_loadable_directory_scheme())
 
     def test_best_scheme_to_load_from(self):
         mock_db = Mock(name='db')
         mock_url_resolver = Mock(name='url_resolver')
         mock_gcs_temp_base_loc = MagicMock(name='gcs_temp_base_loc')
-        bigquery_loader = BigQueryLoader(db=mock_db, url_resolver=mock_url_resolver,
-                                         gcs_temp_base_loc=mock_gcs_temp_base_loc)
+        bigquery_loader = BigQueryLoader(db=None, url_resolver=mock_url_resolver,
+                                         gcs_temp_base_loc=mock_gcs_temp_base_loc, db_conn=mock_db)
         self.assertEqual('gs', bigquery_loader.best_scheme_to_load_from())
diff --git a/tests/unit/db/bigquery/test_bigquery_loader_can_load_this_format.py b/tests/unit/db/bigquery/test_bigquery_loader_can_load_this_format.py
index ced3eda19..a2b6ca5cc 100644
--- a/tests/unit/db/bigquery/test_bigquery_loader_can_load_this_format.py
+++ b/tests/unit/db/bigquery/test_bigquery_loader_can_load_this_format.py
@@ -29,8 +29,8 @@ def test_can_load_this_format_true(self,
         mock_load_plan.records_format = mock_source_records_format
         mock_url_resolver = Mock(name='url_resolver')
         mock_source_records_format.hints = {}
-        bigquery_loader = BigQueryLoader(db=mock_db, url_resolver=mock_url_resolver,
-                                         gcs_temp_base_loc=None)
+        bigquery_loader = BigQueryLoader(db=None, url_resolver=mock_url_resolver,
+                                         gcs_temp_base_loc=None, db_conn=mock_db)
         out = bigquery_loader.can_load_this_format(mock_source_records_format)
         mock_ProcessingInstructions.assert_called_with()
         mock_RecordsLoadPlan.\
@@ -55,8 +55,8 @@ def test_can_load_this_format_delimited_false(self,
         mock_url_resolver = Mock(name='url_resolver')
         mock_load_job_config.side_effect = NotImplementedError
         mock_source_records_format.hints = {}
-        bigquery_loader = BigQueryLoader(db=mock_db, url_resolver=mock_url_resolver,
-                                         gcs_temp_base_loc=None)
+        bigquery_loader = BigQueryLoader(db=None, url_resolver=mock_url_resolver,
+                                         gcs_temp_base_loc=None, db_conn=mock_db)
         out = bigquery_loader.can_load_this_format(mock_source_records_format)
         mock_ProcessingInstructions.assert_called_with()
         mock_RecordsLoadPlan.\
@@ -80,8 +80,8 @@ def test_can_load_this_format_true_avro(self,
         mock_load_plan.records_format = mock_source_records_format
         mock_url_resolver = Mock(name='url_resolver')
         mock_source_records_format.hints = {}
-        bigquery_loader = BigQueryLoader(db=mock_db, url_resolver=mock_url_resolver,
-                                         gcs_temp_base_loc=None)
+        bigquery_loader = BigQueryLoader(db=None, url_resolver=mock_url_resolver,
+                                         gcs_temp_base_loc=None, db_conn=mock_db)
         out = bigquery_loader.can_load_this_format(mock_source_records_format)
         mock_ProcessingInstructions.assert_called_with()
         mock_RecordsLoadPlan.\
@@ -104,8 +104,8 @@ def test_can_load_this_format_false_newformat(self,
         mock_load_plan.records_format = mock_source_records_format
         mock_url_resolver = Mock(name='url_resolver')
         mock_source_records_format.hints = {}
-        bigquery_loader = BigQueryLoader(db=mock_db, url_resolver=mock_url_resolver,
-                                         gcs_temp_base_loc=None)
+        bigquery_loader = BigQueryLoader(db=None, url_resolver=mock_url_resolver,
+                                         gcs_temp_base_loc=None, db_conn=mock_db)
         out = bigquery_loader.can_load_this_format(mock_source_records_format)
         mock_ProcessingInstructions.assert_called_with()
         mock_RecordsLoadPlan.\
@@ -128,8 +128,8 @@ def test_can_load_this_format_true_parquet(self,
         mock_load_plan.records_format = mock_source_records_format
         mock_url_resolver = Mock(name='url_resolver')
         mock_source_records_format.hints = {}
-        bigquery_loader = BigQueryLoader(db=mock_db, url_resolver=mock_url_resolver,
-                                         gcs_temp_base_loc=None)
+        bigquery_loader = BigQueryLoader(db=None, url_resolver=mock_url_resolver,
+                                         gcs_temp_base_loc=None, db_conn=mock_db)
         out = bigquery_loader.can_load_this_format(mock_source_records_format)
         mock_ProcessingInstructions.assert_called_with()
         mock_RecordsLoadPlan.\
diff --git a/tests/unit/db/bigquery/test_bigquery_unloader.py b/tests/unit/db/bigquery/test_bigquery_unloader.py
index 5e262727e..6e4ac28a3 100644
--- a/tests/unit/db/bigquery/test_bigquery_unloader.py
+++ b/tests/unit/db/bigquery/test_bigquery_unloader.py
@@ -14,8 +14,11 @@ def test_can_unload_format_avro_true(self):
         mock_db = Mock(name='mock_db')
         mock_url_resolver = MagicMock(name='mock_url_resolver')
         mock_gcs_temp_base_loc = MagicMock(name='gcs_temp_base_loc')
-        big_query_unloader = BigQueryUnloader(db=mock_db, url_resolver=mock_url_resolver,
-                                              gcs_temp_base_loc=mock_gcs_temp_base_loc)
+        big_query_unloader = BigQueryUnloader(
+            db=None,
+            url_resolver=mock_url_resolver,
+            gcs_temp_base_loc=mock_gcs_temp_base_loc,
+            db_conn=mock_db)
         avro_format = AvroRecordsFormat()
         self.assertTrue(big_query_unloader.can_unload_format(avro_format))
 
@@ -23,8 +26,11 @@ def test_can_unload_format_delimited_false(self):
         mock_db = Mock(name='mock_db')
         mock_url_resolver = MagicMock(name='mock_url_resolver')
         mock_gcs_temp_base_loc = MagicMock(name='gcs_temp_base_loc')
-        big_query_unloader = BigQueryUnloader(db=mock_db, url_resolver=mock_url_resolver,
-                                              gcs_temp_base_loc=mock_gcs_temp_base_loc)
+        big_query_unloader = BigQueryUnloader(
+            db=None,
+            url_resolver=mock_url_resolver,
+            gcs_temp_base_loc=mock_gcs_temp_base_loc,
+            db_conn=mock_db)
         delimited_format = DelimitedRecordsFormat()
         self.assertFalse(big_query_unloader.can_unload_format(delimited_format))
 
@@ -32,32 +38,44 @@ def test_can_unload_to_scheme_gs_true(self):
         mock_db = Mock(name='mock_db')
         mock_url_resolver = MagicMock(name='mock_url_resolver')
         mock_gcs_temp_base_loc = MagicMock(name='gcs_temp_base_loc')
-        big_query_unloader = BigQueryUnloader(db=mock_db, url_resolver=mock_url_resolver,
-                                              gcs_temp_base_loc=mock_gcs_temp_base_loc)
+        big_query_unloader = BigQueryUnloader(
+            db=None,
+            url_resolver=mock_url_resolver,
+            gcs_temp_base_loc=mock_gcs_temp_base_loc,
+            db_conn=mock_db)
         self.assertTrue(big_query_unloader.can_unload_to_scheme('gs'))
 
     def test_can_unload_to_scheme_other_with_temp_bucket_true(self):
         mock_db = Mock(name='mock_db')
         mock_url_resolver = MagicMock(name='mock_url_resolver')
         mock_gcs_temp_base_loc = MagicMock(name='gcs_temp_base_loc')
-        big_query_unloader = BigQueryUnloader(db=mock_db, url_resolver=mock_url_resolver,
-                                              gcs_temp_base_loc=mock_gcs_temp_base_loc)
+        big_query_unloader = BigQueryUnloader(
+            db=None,
+            url_resolver=mock_url_resolver,
+            gcs_temp_base_loc=mock_gcs_temp_base_loc,
+            db_conn=mock_db)
         self.assertTrue(big_query_unloader.can_unload_to_scheme('blah'))
 
     def test_can_unload_to_scheme_other_with_no_temp_bucket_true(self):
         mock_db = Mock(name='mock_db')
         mock_url_resolver = MagicMock(name='mock_url_resolver')
         mock_gcs_temp_base_loc = None
-        big_query_unloader = BigQueryUnloader(db=mock_db, url_resolver=mock_url_resolver,
-                                              gcs_temp_base_loc=mock_gcs_temp_base_loc)
+        big_query_unloader = BigQueryUnloader(
+            db=None,
+            url_resolver=mock_url_resolver,
+            gcs_temp_base_loc=mock_gcs_temp_base_loc,
+            db_conn=mock_db)
         self.assertFalse(big_query_unloader.can_unload_to_scheme('blah'))
 
     def test_known_supported_records_formats_for_unload(self):
         mock_db = Mock(name='mock_db')
         mock_url_resolver = MagicMock(name='mock_url_resolver')
         mock_gcs_temp_base_loc = MagicMock(name='gcs_temp_base_loc')
-        big_query_unloader = BigQueryUnloader(db=mock_db, url_resolver=mock_url_resolver,
-                                              gcs_temp_base_loc=mock_gcs_temp_base_loc)
+        big_query_unloader = BigQueryUnloader(
+            db=None,
+            url_resolver=mock_url_resolver,
+            gcs_temp_base_loc=mock_gcs_temp_base_loc,
+            db_conn=mock_db)
         self.assertEqual([type(format)
                           for format in
                           big_query_unloader.known_supported_records_formats_for_unload()],
@@ -67,8 +85,11 @@ def test_temporary_unloadable_directory_loc_raises(self):
         mock_db = Mock(name='mock_db')
         mock_url_resolver = MagicMock(name='mock_url_resolver')
         mock_gcs_temp_base_loc = None
-        big_query_unloader = BigQueryUnloader(db=mock_db, url_resolver=mock_url_resolver,
-                                              gcs_temp_base_loc=mock_gcs_temp_base_loc)
+        big_query_unloader = BigQueryUnloader(
+            db=None,
+            url_resolver=mock_url_resolver,
+            gcs_temp_base_loc=mock_gcs_temp_base_loc,
+            db_conn=mock_db)
         with self.assertRaises(NoTemporaryBucketConfiguration):
             with big_query_unloader.temporary_unloadable_directory_loc():
                 pass
@@ -77,8 +98,11 @@ def test_temporary_unloadable_directory_loc(self):
         mock_db = Mock(name='mock_db')
         mock_url_resolver = MagicMock(name='mock_url_resolver')
         mock_gcs_temp_base_loc = MagicMock(name='gcs_temp_base_loc')
-        big_query_unloader = BigQueryUnloader(db=mock_db, url_resolver=mock_url_resolver,
-                                              gcs_temp_base_loc=mock_gcs_temp_base_loc)
+        big_query_unloader = BigQueryUnloader(
+            db=None,
+            url_resolver=mock_url_resolver,
+            gcs_temp_base_loc=mock_gcs_temp_base_loc,
+            db_conn=mock_db)
         with big_query_unloader.temporary_unloadable_directory_loc() as temp_loc:
             self.assertEqual(temp_loc,
                              mock_gcs_temp_base_loc.temporary_directory.return_value.__enter__.
@@ -88,8 +112,11 @@ def test_unload(self):
         mock_db = Mock(name='mock_db')
         mock_url_resolver = MagicMock(name='mock_url_resolver')
         mock_gcs_temp_base_loc = MagicMock(name='gcs_temp_base_loc')
-        big_query_unloader = BigQueryUnloader(db=mock_db, url_resolver=mock_url_resolver,
-                                              gcs_temp_base_loc=mock_gcs_temp_base_loc)
+        big_query_unloader = BigQueryUnloader(
+            db=None,
+            url_resolver=mock_url_resolver,
+            gcs_temp_base_loc=mock_gcs_temp_base_loc,
+            db_conn=mock_db)
         mock_schema = 'myproject.mydataset'
         mock_table = 'mytable'
         mock_unload_plan = Mock(name='unload_plan')
diff --git a/tests/unit/db/mysql/test_loader.py b/tests/unit/db/mysql/test_loader.py
index 883e5c7ef..77cb6817a 100644
--- a/tests/unit/db/mysql/test_loader.py
+++ b/tests/unit/db/mysql/test_loader.py
@@ -10,8 +10,9 @@ def setUp(self):
         self.mock_url_resolver = Mock(name='url_resolver')
         self.mock_db_engine = MagicMock(name='db_engine')
         self.mock_db_engine.engine = self.mock_db_engine
-        self.loader = MySQLLoader(db=self.mock_db_engine,
-                                  url_resolver=self.mock_url_resolver)
+        self.loader = MySQLLoader(db=None,
+                                  url_resolver=self.mock_url_resolver,
+                                  db_engine=self.mock_db_engine)
 
     @patch('records_mover.db.mysql.loader.complain_on_unhandled_hints')
     @patch('records_mover.db.mysql.loader.mysql_load_options')
@@ -67,5 +68,5 @@ def test_load_happy_path(self,
                                mock_table,
                                mock_load_plan,
                                mock_directory)
-        self.mock_db_engine.execute.assert_called_with(mock_sql)
+        self.mock_db_engine.connect.return_value.execute.assert_called_with(mock_sql)
         self.assertEqual(out, None)
diff --git a/tests/unit/db/postgres/test_loader.py b/tests/unit/db/postgres/test_loader.py
index 44ec4d19f..172321a96 100644
--- a/tests/unit/db/postgres/test_loader.py
+++ b/tests/unit/db/postgres/test_loader.py
@@ -12,7 +12,8 @@ def setUp(self):
         self.mock_db = MagicMock(name='db')
         self.loader = PostgresLoader(self.mock_url_resolver,
                                      self.mock_meta,
-                                     self.mock_db)
+                                     None,
+                                     db_conn=self.mock_db)
 
     @patch('records_mover.db.postgres.loader.quote_value')
     @patch('records_mover.db.postgres.loader.copy_from')
@@ -57,11 +58,12 @@ def test_load_from_fileobj(self,
         mock_Table.assert_called_with(mock_table,
                                       self.mock_meta,
                                       schema=mock_schema,
-                                      autoload=True,
-                                      autoload_with=self.mock_db)
-        mock_conn = self.mock_db.engine.begin.return_value.__enter__.return_value
-        mock_quote_value.assert_called_with(mock_conn, 'ISO, DATE_ORDER_STYLE')
-        mock_conn.execute.assert_called_with('SET LOCAL DateStyle = ABC')
+                                      autoload_with=self.mock_db.engine)
+        mock_conn = self.mock_db
+        mock_quote_value.assert_called_with(None, 'ISO, DATE_ORDER_STYLE',
+                                            db_engine=mock_conn.engine)
+        str_arg = str(mock_conn.execute.call_args.args[0])
+        self.assertEqual(str_arg, 'SET LOCAL DateStyle = ABC')
         mock_copy_from.assert_called_with(mock_fileobj,
                                           mock_table_obj,
                                           mock_conn,
@@ -110,11 +112,11 @@ def test_load_from_fileobj_default_date_order_style(self,
         mock_Table.assert_called_with(mock_table,
                                       self.mock_meta,
                                       schema=mock_schema,
-                                      autoload=True,
-                                      autoload_with=self.mock_db)
-        mock_conn = self.mock_db.engine.begin.return_value.__enter__.return_value
-        mock_quote_value.assert_called_with(mock_conn, 'ISO, MDY')
-        mock_conn.execute.assert_called_with('SET LOCAL DateStyle = ABC')
+                                      autoload_with=self.mock_db.engine)
+        mock_conn = self.mock_db
+        mock_quote_value.assert_called_with(None, 'ISO, MDY', db_engine=mock_conn.engine)
+        str_arg = str(mock_conn.execute.call_args.args[0])
+        self.assertEqual(str_arg, 'SET LOCAL DateStyle = ABC')
         mock_copy_from.assert_called_with(mock_fileobj,
                                           mock_table_obj,
                                           mock_conn,
@@ -172,11 +174,12 @@ def test_load(self,
         mock_Table.assert_called_with(mock_table,
                                       self.mock_meta,
                                       schema=mock_schema,
-                                      autoload=True,
-                                      autoload_with=self.mock_db)
-        mock_conn = self.mock_db.engine.begin.return_value.__enter__.return_value
-        mock_quote_value.assert_called_with(mock_conn, 'ISO, DATE_ORDER_STYLE')
-        mock_conn.execute.assert_called_with('SET LOCAL DateStyle = ABC')
+                                      autoload_with=self.mock_db.engine)
+        mock_conn = self.mock_db
+        mock_quote_value.assert_called_with(None, 'ISO, DATE_ORDER_STYLE',
+                                            db_engine=mock_conn.engine)
+        str_arg = str(mock_conn.execute.call_args.args[0])
+        self.assertEqual(str_arg, 'SET LOCAL DateStyle = ABC')
         mock_copy_from.assert_called_with(mock_loc.open.return_value.__enter__.return_value,
                                           mock_table_obj,
                                           mock_conn,
diff --git a/tests/unit/db/postgres/test_unloader.py b/tests/unit/db/postgres/test_unloader.py
index e98153661..279d950f3 100644
--- a/tests/unit/db/postgres/test_unloader.py
+++ b/tests/unit/db/postgres/test_unloader.py
@@ -8,7 +8,7 @@ class TestPostgresUnloader(unittest.TestCase):
     def setUp(self):
         self.mock_url_resolver = Mock(name='url_resolver')
         self.mock_db = MagicMock(name='db')
-        self.unloader = PostgresUnloader(self.mock_db)
+        self.unloader = PostgresUnloader(None, db_conn=self.mock_db)
 
     @patch('records_mover.db.postgres.unloader.quote_value')
     @patch('records_mover.db.postgres.unloader.copy_to')
@@ -55,11 +55,12 @@ def test_unload(self,
         mock_Table.assert_called_with(mock_table,
                                       ANY,
                                       schema=mock_schema,
-                                      autoload=True,
-                                      autoload_with=self.mock_db)
-        mock_conn = self.mock_db.engine.begin.return_value.__enter__.return_value
-        mock_quote_value.assert_called_with(mock_conn, 'DATE_OUTPUT_STYLE, DATE_ORDER_STYLE')
-        mock_conn.execute.assert_called_with('SET LOCAL DateStyle = ABC')
+                                      autoload_with=self.mock_db.engine)
+        mock_conn = self.mock_db
+        mock_quote_value.assert_called_with(None, 'DATE_OUTPUT_STYLE, DATE_ORDER_STYLE',
+                                            db_engine=mock_conn.engine)
+        str_arg = str(mock_conn.execute.call_args.args[0])
+        self.assertEqual(str_arg, 'SET LOCAL DateStyle = ABC')
         mock_fileobj = mock_directory.loc.file_in_this_directory.return_value.open.\
             return_value.__enter__.return_value
         mock_copy_to.assert_called_with(mock_table_obj.select.return_value,
@@ -112,11 +113,12 @@ def test_unload_default_date_order_style(self,
         mock_Table.assert_called_with(mock_table,
                                       ANY,
                                       schema=mock_schema,
-                                      autoload=True,
-                                      autoload_with=self.mock_db)
-        mock_conn = self.mock_db.engine.begin.return_value.__enter__.return_value
-        mock_quote_value.assert_called_with(mock_conn, 'DATE_OUTPUT_STYLE, MDY')
-        mock_conn.execute.assert_called_with('SET LOCAL DateStyle = ABC')
+                                      autoload_with=self.mock_db.engine)
+        mock_conn = self.mock_db
+        mock_quote_value.assert_called_with(None, 'DATE_OUTPUT_STYLE, MDY',
+                                            db_engine=mock_conn.engine)
+        str_arg = str(mock_conn.execute.call_args.args[0])
+        self.assertEqual(str_arg, 'SET LOCAL DateStyle = ABC')
         mock_fileobj = mock_directory.loc.file_in_this_directory.return_value.open.\
             return_value.__enter__.return_value
         mock_copy_to.assert_called_with(mock_table_obj.select.return_value,
diff --git a/tests/unit/db/redshift/base_test_redshift_db_driver.py b/tests/unit/db/redshift/base_test_redshift_db_driver.py
index 156e72395..df67fc8a6 100644
--- a/tests/unit/db/redshift/base_test_redshift_db_driver.py
+++ b/tests/unit/db/redshift/base_test_redshift_db_driver.py
@@ -22,9 +22,11 @@ def setUp(self):
             quote.return_value.\
             __add__.return_value.\
             __add__.return_value = 'myschema.mytable'
-        self.redshift_db_driver = RedshiftDBDriver(db=self.mock_db_engine,
-                                                   s3_temp_base_loc=self.mock_s3_temp_base_loc)
-
+        self.redshift_db_driver = RedshiftDBDriver(db=None,
+                                                   s3_temp_base_loc=self.mock_s3_temp_base_loc,
+                                                   db_engine=self.mock_db_engine)
+        self.mock_db_engine.dialect.ddl_compiler.return_value = (
+            '\nCREATE TABLE myschema.mytable (\n)\n\n')
         mock_records_unload_plan = create_autospec(RecordsUnloadPlan)
         mock_records_unload_plan.records_format = create_autospec(DelimitedRecordsFormat)
         mock_records_unload_plan.records_format.format_type = 'delimited'
diff --git a/tests/unit/db/redshift/test_loader.py b/tests/unit/db/redshift/test_loader.py
index 0c0bde19e..19393ea76 100644
--- a/tests/unit/db/redshift/test_loader.py
+++ b/tests/unit/db/redshift/test_loader.py
@@ -12,9 +12,10 @@ def setUp(self):
         self.s3_temp_base_loc = MagicMock(name='s3_temp_base_loc')
 
         self.redshift_loader =\
-            RedshiftLoader(db=self.mock_db,
+            RedshiftLoader(db=None,
                            meta=self.mock_meta,
-                           s3_temp_base_loc=self.s3_temp_base_loc)
+                           s3_temp_base_loc=self.s3_temp_base_loc,
+                           db_conn=self.mock_db)
 
     @patch('records_mover.db.redshift.loader.redshift_copy_options')
     @patch('records_mover.db.redshift.loader.ProcessingInstructions')
@@ -134,7 +135,7 @@ def test_load_load_error(self,
         mock_s3_directory.loc.url = 's3://foo/bar/baz/'
 
         def db_execute(command):
-            if command == 'SELECT pg_backend_pid();':
+            if str(command) == 'SELECT pg_backend_pid();':
                 mock_result_proxy = Mock(name='result_proxy')
                 mock_result_proxy.scalar.return_value = 123
                 return mock_result_proxy
diff --git a/tests/unit/db/redshift/test_loader_temporary_bucket.py b/tests/unit/db/redshift/test_loader_temporary_bucket.py
index 966ac6607..201cd4825 100644
--- a/tests/unit/db/redshift/test_loader_temporary_bucket.py
+++ b/tests/unit/db/redshift/test_loader_temporary_bucket.py
@@ -10,9 +10,10 @@ def test_temporary_s3_directory_loc_no_bucket(self):
         mock_meta = Mock(name='meta')
 
         redshift_loader =\
-            RedshiftLoader(db=mock_db,
+            RedshiftLoader(db=None,
                            meta=mock_meta,
-                           s3_temp_base_loc=None)
+                           s3_temp_base_loc=None,
+                           db_conn=mock_db)
 
         with self.assertRaises(NoTemporaryBucketConfiguration):
             with redshift_loader.temporary_s3_directory_loc():
diff --git a/tests/unit/db/redshift/test_redshift_db_driver.py b/tests/unit/db/redshift/test_redshift_db_driver.py
index 2a23f907a..4f94e9221 100644
--- a/tests/unit/db/redshift/test_redshift_db_driver.py
+++ b/tests/unit/db/redshift/test_redshift_db_driver.py
@@ -7,10 +7,8 @@ class TestRedshiftDBDriver(BaseTestRedshiftDBDriver):
     maxDiff = None
 
     def test_schema_sql(self):
-        self.mock_db_engine.execute.return_value.fetchall.return_value =\
-            [{}, {'ddl': 'real line 1'}, {'ddl': 'real line 2'}]
         sql = self.redshift_db_driver.schema_sql('myschema', 'mytable')
-        self.assertEqual(sql, "real line 1\nreal line 2")
+        self.assertEqual(sql, '\nCREATE TABLE myschema.mytable (\n)\n\n')
 
     def test_schema_sql_no_admin_views(self):
         self.mock_db_engine.execute.return_value.fetchall.return_value = []
@@ -28,8 +26,11 @@ def test_set_grant_permissions_for_group(self, mock_quote_schema_and_table,
         mock_quote_group_name.return_value = '"a_group"'
         groups = {'all': ['a_group']}
         mock_conn = self.mock_db_engine.engine.connect.return_value.__enter__.return_value
-        self.redshift_db_driver.set_grant_permissions_for_groups(mock_schema, mock_table,
-                                                                 groups, mock_conn)
+        self.redshift_db_driver.set_grant_permissions_for_groups(mock_schema,
+                                                                 mock_table,
+                                                                 groups,
+                                                                 None,
+                                                                 db_conn=mock_conn)
         mock_conn.execute.assert_called_with(
             f'GRANT all ON TABLE {mock_schema}.{mock_table} TO GROUP "a_group"')
 
diff --git a/tests/unit/db/redshift/test_redshift_db_driver_format_negotiation.py b/tests/unit/db/redshift/test_redshift_db_driver_format_negotiation.py
index 98833fa80..821d228fe 100644
--- a/tests/unit/db/redshift/test_redshift_db_driver_format_negotiation.py
+++ b/tests/unit/db/redshift/test_redshift_db_driver_format_negotiation.py
@@ -23,8 +23,9 @@ def setUp(self, mock_RedshiftUnloader, mock_RedshiftLoader):
             quote.return_value.\
             __add__.return_value.\
             __add__.return_value = 'myschema.mytable'
-        self.redshift_db_driver = RedshiftDBDriver(db=mock_db_engine,
-                                                   s3_temp_base_loc=mock_s3_temp_base_loc)
+        self.redshift_db_driver = RedshiftDBDriver(db=None,
+                                                   s3_temp_base_loc=mock_s3_temp_base_loc,
+                                                   db_engine=mock_db_engine,)
 
     def test_can_load_this_format(self):
         mock_source_records_format = Mock(name='source_records_format', spec=DelimitedRecordsFormat)
diff --git a/tests/unit/db/redshift/test_redshift_db_driver_unload.py b/tests/unit/db/redshift/test_redshift_db_driver_unload.py
index 0d8d0bb79..a68adf374 100644
--- a/tests/unit/db/redshift/test_redshift_db_driver_unload.py
+++ b/tests/unit/db/redshift/test_redshift_db_driver_unload.py
@@ -23,7 +23,9 @@ def test_unload_to_non_s3(self,
             DelimitedRecordsFormat(variant='bluelabs',
                                    hints=bluelabs_format_hints)
         self.mock_directory.scheme = 'mumble'
-        self.mock_db_engine.execute.return_value.scalar.return_value = 456
+        self.mock_db_engine.connect.return_value \
+            .execute.return_value \
+            .scalar.return_value = 456
         rows = self.redshift_db_driver.unloader().\
             unload(schema='myschema',
                    table='mytable',
@@ -61,7 +63,9 @@ def test_unload(self,
             DelimitedRecordsFormat(variant='bluelabs',
                                    hints=bluelabs_format_hints)
         self.mock_directory.scheme = 's3'
-        self.mock_db_engine.execute.return_value.scalar.return_value = 456
+        self.mock_db_engine.connect.return_value \
+            .execute.return_value \
+            .scalar.return_value = 456
         rows = self.redshift_db_driver.unloader().\
             unload(schema='myschema',
                    table='mytable',
diff --git a/tests/unit/db/redshift/test_sql.py b/tests/unit/db/redshift/test_sql.py
index 398c27c7e..825ca3a46 100644
--- a/tests/unit/db/redshift/test_sql.py
+++ b/tests/unit/db/redshift/test_sql.py
@@ -1,18 +1,23 @@
 import unittest
 import sqlalchemy
 from records_mover.db.redshift.sql import schema_sql_from_admin_views
-from mock import patch, Mock
+from mock import patch, Mock, MagicMock
 
 
 class TestSQL(unittest.TestCase):
     @patch('records_mover.db.redshift.sql.logger')
     def test_schema_sql_from_admin_views_not_installed(self,
                                                        mock_logger):
-        mock_db = Mock(name='db')
+        mock_db = MagicMock(name='db')
         mock_table = Mock(name='table')
         mock_schema = Mock(name='schema')
-        mock_db.execute.side_effect = sqlalchemy.exc.ProgrammingError('statement', {}, {})
-        out = schema_sql_from_admin_views(mock_schema, mock_table, mock_db)
+        mock_connection = MagicMock(name='connection')
+        mock_db.connect.return_value \
+               .__enter__.return_value = mock_connection
+        mock_connection.begin.return_value \
+                       .__enter__.return_value = None
+        mock_connection.execute.side_effect = sqlalchemy.exc.ProgrammingError('statement', {}, {})
+        out = schema_sql_from_admin_views(mock_schema, mock_table, None, db_conn=mock_connection)
         self.assertIsNone(out)
         mock_logger.debug.assert_called_with('Error while generating SQL', exc_info=True)
         mock_logger.warning.\
diff --git a/tests/unit/db/redshift/test_unloader.py b/tests/unit/db/redshift/test_unloader.py
index 8b81f6b97..9437e6746 100644
--- a/tests/unit/db/redshift/test_unloader.py
+++ b/tests/unit/db/redshift/test_unloader.py
@@ -23,9 +23,10 @@ def test_can_unload_format_true(self,
         mock_target_records_format.hints = {}
 
         redshift_unloader =\
-            RedshiftUnloader(db=mock_db,
+            RedshiftUnloader(db=None,
                              table=mock_table,
-                             s3_temp_base_loc=mock_s3_temp_base_loc)
+                             s3_temp_base_loc=mock_s3_temp_base_loc,
+                             db_conn=mock_db)
         out = redshift_unloader.can_unload_format(mock_target_records_format)
         mock_RecordsUnloadPlan.\
             assert_called_with(records_format=mock_target_records_format)
@@ -53,9 +54,10 @@ def test_can_unload_format_delimite_false(self,
         mock_redshift_unload_options.side_effect = NotImplementedError
 
         redshift_unloader =\
-            RedshiftUnloader(db=mock_db,
+            RedshiftUnloader(db=None,
                              table=mock_table,
-                             s3_temp_base_loc=mock_s3_temp_base_loc)
+                             s3_temp_base_loc=mock_s3_temp_base_loc,
+                             db_conn=mock_db)
         out = redshift_unloader.can_unload_format(mock_target_records_format)
         mock_RecordsUnloadPlan.\
             assert_called_with(records_format=mock_target_records_format)
@@ -70,9 +72,10 @@ def test_can_unload_to_scheme_s3_true(self):
         mock_table = Mock(name='table')
 
         redshift_unloader =\
-            RedshiftUnloader(db=mock_db,
+            RedshiftUnloader(db=None,
                              table=mock_table,
-                             s3_temp_base_loc=None)
+                             s3_temp_base_loc=None,
+                             db_conn=mock_db)
         self.assertTrue(redshift_unloader.can_unload_to_scheme('s3'))
 
     def test_can_unload_to_scheme_file_without_temp_bucket_true(self):
@@ -80,9 +83,10 @@ def test_can_unload_to_scheme_file_without_temp_bucket_true(self):
         mock_table = Mock(name='table')
 
         redshift_unloader =\
-            RedshiftUnloader(db=mock_db,
+            RedshiftUnloader(db=None,
                              table=mock_table,
-                             s3_temp_base_loc=None)
+                             s3_temp_base_loc=None,
+                             db_conn=mock_db)
         self.assertFalse(redshift_unloader.can_unload_to_scheme('file'))
 
     def test_can_unload_to_scheme_file_with_temp_bucket_true(self):
@@ -91,9 +95,10 @@ def test_can_unload_to_scheme_file_with_temp_bucket_true(self):
         mock_s3_temp_base_loc = Mock(name='s3_temp_base_loc')
 
         redshift_unloader =\
-            RedshiftUnloader(db=mock_db,
+            RedshiftUnloader(db=None,
                              table=mock_table,
-                             s3_temp_base_loc=mock_s3_temp_base_loc)
+                             s3_temp_base_loc=mock_s3_temp_base_loc,
+                             db_conn=mock_db)
         self.assertTrue(redshift_unloader.can_unload_to_scheme('file'))
 
     def test_known_supported_records_formats_for_unload(self):
@@ -102,9 +107,10 @@ def test_known_supported_records_formats_for_unload(self):
         mock_s3_temp_base_loc = Mock(name='s3_temp_base_loc')
 
         redshift_unloader =\
-            RedshiftUnloader(db=mock_db,
+            RedshiftUnloader(db=None,
                              table=mock_table,
-                             s3_temp_base_loc=mock_s3_temp_base_loc)
+                             s3_temp_base_loc=mock_s3_temp_base_loc,
+                             db_conn=mock_db)
         formats = redshift_unloader.known_supported_records_formats_for_unload()
 
         self.assertEqual([f.__class__ for f in formats],
@@ -116,9 +122,10 @@ def test_temporary_unloadable_directory_loc(self):
         mock_s3_temp_base_loc = MagicMock(name='s3_temp_base_loc')
 
         redshift_unloader =\
-            RedshiftUnloader(db=mock_db,
+            RedshiftUnloader(db=None,
                              table=mock_table,
-                             s3_temp_base_loc=mock_s3_temp_base_loc)
+                             s3_temp_base_loc=mock_s3_temp_base_loc,
+                             db_conn=mock_db)
         with redshift_unloader.temporary_unloadable_directory_loc() as loc:
             self.assertEqual(loc,
                              mock_s3_temp_base_loc.temporary_directory.return_value.__enter__.
@@ -130,9 +137,10 @@ def test_temporary_unloadable_directory_loc_unset(self):
         mock_s3_temp_base_loc = None
 
         redshift_unloader =\
-            RedshiftUnloader(db=mock_db,
+            RedshiftUnloader(db=None,
                              table=mock_table,
-                             s3_temp_base_loc=mock_s3_temp_base_loc)
+                             s3_temp_base_loc=mock_s3_temp_base_loc,
+                             db_conn=mock_db)
         with self.assertRaises(NoTemporaryBucketConfiguration):
             with redshift_unloader.temporary_unloadable_directory_loc():
                 pass
diff --git a/tests/unit/db/test_db_conn_mixin.py b/tests/unit/db/test_db_conn_mixin.py
new file mode 100644
index 000000000..57709278e
--- /dev/null
+++ b/tests/unit/db/test_db_conn_mixin.py
@@ -0,0 +1,46 @@
+import pytest
+from unittest.mock import MagicMock
+from sqlalchemy.engine import Connection, Engine
+from records_mover.db.db_conn_mixin import DBConnMixin
+
+
+class TestDBConnMixin:
+    @pytest.fixture
+    def mixin(self):
+        class TestMixin(DBConnMixin):
+            def __init__(self):
+                self._db_conn = None
+                self.db_engine = MagicMock(spec=Engine)
+                self.conn_opened_here = False
+        return TestMixin()
+
+    def test_get_db_conn(self, mixin):
+        conn = MagicMock(spec=Connection)
+        mixin.db_engine.connect.return_value = conn
+        assert mixin.get_db_conn() == conn
+        assert mixin.conn_opened_here is True
+
+    def test_get_db_conn_with_existing_conn(self, mixin):
+        conn = MagicMock(spec=Connection)
+        mixin._db_conn = conn
+        assert mixin.get_db_conn() == conn
+        assert mixin.conn_opened_here is False
+
+    def test_set_db_conn(self, mixin):
+        conn = MagicMock(spec=Connection)
+        mixin.set_db_conn(conn)
+        assert mixin._db_conn == conn
+
+    def test_del_db_conn(self, mixin):
+        mixin.conn_opened_here = True
+        conn = MagicMock(spec=Connection)
+        mixin._db_conn = conn
+        mixin.del_db_conn()
+        assert mixin._db_conn is None
+        assert mixin.conn_opened_here is False
+
+    def test_del_db_conn_with_no_conn(self, mixin):
+        mixin.conn_opened_here = True
+        mixin.del_db_conn()
+        assert mixin._db_conn is None
+        assert mixin.conn_opened_here is False
diff --git a/tests/unit/db/test_db_driver.py b/tests/unit/db/test_db_driver.py
index 69075441b..c2060105b 100644
--- a/tests/unit/db/test_db_driver.py
+++ b/tests/unit/db/test_db_driver.py
@@ -11,7 +11,8 @@ def setUp(self):
         self.mock_url_resolver = Mock(name='url_resolver')
         self.mock_s3_temp_base_loc = MagicMock(name='s3_temp_base_loc')
         self.mock_s3_temp_base_loc.url = 's3://fakebucket/fakedir/fakesubdir/'
-        self.db_driver = GenericDBDriver(db=self.mock_db_engine,
+        self.db_driver = GenericDBDriver(db=None,
+                                         db_engine=self.mock_db_engine,
                                          s3_temp_base_loc=self.mock_s3_temp_base_loc,
                                          url_resolver=self.mock_url_resolver,
                                          text=fake_text)
@@ -69,10 +70,9 @@ def test_has_table(self):
         mock_schema = Mock(name='schema')
         mock_table = Mock(name='table')
         out = self.db_driver.has_table(mock_schema, mock_table)
-        self.assertEqual(out, self.mock_db_engine.dialect.has_table.return_value)
-        self.mock_db_engine.dialect.has_table.assert_called_with(self.mock_db_engine,
-                                                                 schema=mock_schema,
-                                                                 table_name=mock_table)
+        self.assertEqual(out, self.mock_db_engine._sa_instance_state.has_table.return_value)
+        self.mock_db_engine._sa_instance_state.has_table.assert_called_with(mock_table,
+                                                                            schema=mock_schema)
 
     @patch('records_mover.db.driver.quote_group_name')
     @patch('records_mover.db.driver.quote_schema_and_table')
@@ -90,10 +90,12 @@ def test_set_grant_permissions_for_groups(self,
         self.db_driver.set_grant_permissions_for_groups(mock_schema_name,
                                                         mock_table,
                                                         groups,
-                                                        mock_db)
-        mock_quote_schema_and_table.assert_called_with(self.mock_db_engine.engine,
+                                                        None,
+                                                        db_conn=mock_db)
+        mock_quote_schema_and_table.assert_called_with(None,
                                                        mock_schema_name,
-                                                       mock_table)
+                                                       mock_table,
+                                                       db_engine=self.mock_db_engine)
         mock_db.execute.assert_has_calls([
             call(f"GRANT write ON TABLE {mock_schema_and_table} TO {mock_group_name}"),
             call(f"GRANT write ON TABLE {mock_schema_and_table} TO {mock_group_name}"),
@@ -115,10 +117,12 @@ def test_set_grant_permissions_for_users(self,
         self.db_driver.set_grant_permissions_for_users(mock_schema_name,
                                                        mock_table,
                                                        users,
-                                                       mock_db)
-        mock_quote_schema_and_table.assert_called_with(self.mock_db_engine.engine,
+                                                       None,
+                                                       db_conn=mock_db)
+        mock_quote_schema_and_table.assert_called_with(None,
                                                        mock_schema_name,
-                                                       mock_table)
+                                                       mock_table,
+                                                       db_engine=self.mock_db_engine)
         mock_db.execute.assert_has_calls([
             call(f"GRANT write ON TABLE {mock_schema_and_table} TO {mock_user_name}"),
             call(f"GRANT write ON TABLE {mock_schema_and_table} TO {mock_user_name}"),
@@ -139,7 +143,8 @@ def test_set_grant_permissions_for_users_bobby_tables(self,
             self.db_driver.set_grant_permissions_for_users(mock_schema_name,
                                                            mock_table,
                                                            users,
-                                                           mock_db)
+                                                           None,
+                                                           db_conn=mock_db)
 
     @patch('records_mover.db.driver.quote_user_name')
     @patch('records_mover.db.driver.quote_schema_and_table')
@@ -156,7 +161,8 @@ def test_set_grant_permissions_for_groups_bobby_tables(self,
             self.db_driver.set_grant_permissions_for_groups(mock_schema_name,
                                                             mock_table,
                                                             groups,
-                                                            mock_db)
+                                                            None,
+                                                            db_conn=mock_db)
 
     def test_tweak_records_schema_for_load_no_tweak(self):
         mock_records_schema = Mock(name='records_schema')
diff --git a/tests/unit/db/test_factory.py b/tests/unit/db/test_factory.py
index 368ee6967..3a40751bc 100644
--- a/tests/unit/db/test_factory.py
+++ b/tests/unit/db/test_factory.py
@@ -10,7 +10,7 @@ def test_db_driver_vertica(self,
         mock_db = Mock(name='db')
         mock_engine = mock_db.engine
         mock_engine.name = 'vertica'
-        out = db_driver(mock_db)
+        out = db_driver(None, db_conn=mock_db)
         self.assertEqual(out, mock_VerticaDBDriver.return_value)
 
     @patch('records_mover.db.redshift.redshift_db_driver.RedshiftDBDriver')
@@ -19,7 +19,7 @@ def test_db_driver_redshift(self,
         mock_db = Mock(name='db')
         mock_engine = mock_db.engine
         mock_engine.name = 'redshift'
-        out = db_driver(mock_db)
+        out = db_driver(None, db_conn=mock_db)
         self.assertEqual(out, mock_RedshiftDBDriver.return_value)
 
     @patch('records_mover.db.bigquery.bigquery_db_driver.BigQueryDBDriver')
@@ -28,7 +28,7 @@ def test_db_driver_bigquery(self,
         mock_db = Mock(name='db')
         mock_engine = mock_db.engine
         mock_engine.name = 'bigquery'
-        out = db_driver(mock_db)
+        out = db_driver(None, db_conn=mock_db)
         self.assertEqual(out, mock_BigQueryDBDriver.return_value)
 
     @patch('records_mover.db.factory.GenericDBDriver')
@@ -37,5 +37,5 @@ def test_db_driver_other(self,
         mock_db = Mock(name='db')
         mock_engine = mock_db.engine
         mock_engine.name = 'somaskdfaksjf'
-        out = db_driver(mock_db)
+        out = db_driver(None, db_conn=mock_db)
         self.assertEqual(out, mock_GenericDBDriver.return_value)
diff --git a/tests/unit/db/test_quoting.py b/tests/unit/db/test_quoting.py
index 7697c4005..4eee5ab34 100644
--- a/tests/unit/db/test_quoting.py
+++ b/tests/unit/db/test_quoting.py
@@ -8,7 +8,7 @@ def test_quote_table_only(self):
         mock_engine = Mock()
         quotable_value = Mock()
         mock_preparer = mock_engine.dialect.preparer.return_value
-        quoted = quoting.quote_table_only(mock_engine, quotable_value)
+        quoted = quoting.quote_table_only(None, quotable_value, db_engine=mock_engine)
         self.assertEqual(mock_preparer.quote.return_value, quoted)
         mock_preparer.quote.assert_called_with(quotable_value)
 
@@ -16,7 +16,7 @@ def test_quote_column_name(self):
         mock_engine = Mock()
         quotable_value = Mock()
         mock_preparer = mock_engine.dialect.preparer.return_value
-        quoted = quoting.quote_column_name(mock_engine, quotable_value)
+        quoted = quoting.quote_column_name(None, quotable_value, db_engine=mock_engine)
         self.assertEqual(mock_preparer.quote.return_value, quoted)
         mock_preparer.quote.assert_called_with(quotable_value)
 
@@ -26,9 +26,10 @@ def test_quote_schema_and_table(self):
         quotable_schema = Mock()
         mock_preparer = mock_engine.dialect.preparer.return_value
         mock_preparer.quote.return_value = '"foo"'
-        quoted = quoting.quote_schema_and_table(mock_engine,
+        quoted = quoting.quote_schema_and_table(None,
                                                 quotable_schema,
-                                                quotable_table)
+                                                quotable_table,
+                                                db_engine=mock_engine)
         self.assertEqual(mock_preparer.quote.return_value + "." +
                          mock_preparer.quote.return_value,
                          quoted)
@@ -39,7 +40,7 @@ def test_quote_value(self):
         mock_engine = Mock()
         quotable_value = Mock()
         mock_preparer = mock_engine.dialect.preparer.return_value
-        quoted = quoting.quote_value(mock_engine, quotable_value)
+        quoted = quoting.quote_value(None, quotable_value, db_engine=mock_engine)
         self.assertEqual(mock_preparer.quote.return_value, quoted)
         mock_preparer.quote.assert_called_with(quotable_value)
         mock_engine.dialect.preparer.assert_called_with(mock_engine.dialect,
diff --git a/tests/unit/db/vertica/base_test_vertica_db_driver.py b/tests/unit/db/vertica/base_test_vertica_db_driver.py
index 9661fb18f..11baa015c 100644
--- a/tests/unit/db/vertica/base_test_vertica_db_driver.py
+++ b/tests/unit/db/vertica/base_test_vertica_db_driver.py
@@ -20,9 +20,10 @@ def setUp(self):
         self.mock_s3_temp_base_loc.url = 's3://fakebucket/fakedir/fakesubdir/'
         with patch('records_mover.db.vertica.vertica_db_driver.VerticaLoader') \
                 as mock_VerticaLoader:
-            self.vertica_db_driver = VerticaDBDriver(db=self.mock_db_engine,
+            self.vertica_db_driver = VerticaDBDriver(db=None,
                                                      s3_temp_base_loc=self.mock_s3_temp_base_loc,
-                                                     url_resolver=self.mock_url_resolver)
+                                                     url_resolver=self.mock_url_resolver,
+                                                     db_conn=self.mock_db_engine)
             self.mock_VerticaLoader = mock_VerticaLoader
             self.mock_vertica_loader = mock_VerticaLoader.return_value
 
diff --git a/tests/unit/db/vertica/test_import_sql.py b/tests/unit/db/vertica/test_import_sql.py
index 9b6dcf25c..3a96ef63e 100644
--- a/tests/unit/db/vertica/test_import_sql.py
+++ b/tests/unit/db/vertica/test_import_sql.py
@@ -16,12 +16,12 @@ class TestImportSQL(unittest.TestCase):
     def test_vertica_import_sql(self,
                                 mock_quote_value,
                                 mock_quote_schema_and_table):
-        def null_schema_table_quoter(db_engine, schema, table):
+        def null_schema_table_quoter(db, schema, table, db_engine=None):
             return f"{schema}.{table}"
 
         mock_quote_schema_and_table.side_effect = null_schema_table_quoter
 
-        def simple_value_quoter(db_engine, s):
+        def simple_value_quoter(db, s, db_engine=None):
             return f"'{s}'"
 
         mock_quote_value.side_effect = simple_value_quoter
diff --git a/tests/unit/db/vertica/test_loader.py b/tests/unit/db/vertica/test_loader.py
index 00aafef2f..eb81c8adf 100644
--- a/tests/unit/db/vertica/test_loader.py
+++ b/tests/unit/db/vertica/test_loader.py
@@ -11,7 +11,8 @@ class TestVerticaLoader(unittest.TestCase):
     def setUp(self):
         mock_url_resolver = Mock(name='url_resolver')
         mock_db = Mock(name='db')
-        self.vertica_loader = VerticaLoader(url_resolver=mock_url_resolver, db=mock_db)
+        self.vertica_loader = VerticaLoader(url_resolver=mock_url_resolver, db=None,
+                                            db_conn=mock_db)
 
     @patch('records_mover.db.vertica.loader.ProcessingInstructions')
     @patch('records_mover.db.vertica.loader.RecordsLoadPlan')
diff --git a/tests/unit/db/vertica/test_unloader.py b/tests/unit/db/vertica/test_unloader.py
index 4a83eada9..06ac3ec97 100644
--- a/tests/unit/db/vertica/test_unloader.py
+++ b/tests/unit/db/vertica/test_unloader.py
@@ -12,7 +12,8 @@ def test_can_unload_format_true(self, mock_vertica_export_options):
         mock_db = Mock(name='db')
         mock_source_records_format = Mock(name='source_records_format', spec=DelimitedRecordsFormat)
         mock_s3_temp_base_loc = Mock(name='s3_temp_base_loc')
-        vertica_unloader = VerticaUnloader(db=mock_db, s3_temp_base_loc=mock_s3_temp_base_loc)
+        vertica_unloader = VerticaUnloader(db=None, s3_temp_base_loc=mock_s3_temp_base_loc,
+                                           db_conn=mock_db)
         mock_source_records_format.hints = {}
         out = vertica_unloader.can_unload_format(mock_source_records_format)
         mock_vertica_export_options.assert_called_with(set(), ANY)
@@ -23,7 +24,8 @@ def test_can_unload_format_false(self, mock_vertica_export_options):
         mock_db = Mock(name='db')
         mock_source_records_format = Mock(name='source_records_format', spec=DelimitedRecordsFormat)
         mock_s3_temp_base_loc = Mock(name='s3_temp_base_loc')
-        vertica_unloader = VerticaUnloader(db=mock_db, s3_temp_base_loc=mock_s3_temp_base_loc)
+        vertica_unloader = VerticaUnloader(db=None, s3_temp_base_loc=mock_s3_temp_base_loc,
+                                           db_conn=mock_db)
         mock_source_records_format.hints = {}
         mock_vertica_export_options.side_effect = NotImplementedError
         out = vertica_unloader.can_unload_format(mock_source_records_format)
@@ -34,7 +36,8 @@ def test_known_supported_records_formats_for_unload(self):
         mock_db = Mock(name='db')
         mock_source_records_format = Mock(name='source_records_format', spec=DelimitedRecordsFormat)
         mock_s3_temp_base_loc = Mock(name='s3_temp_base_loc')
-        vertica_unloader = VerticaUnloader(db=mock_db, s3_temp_base_loc=mock_s3_temp_base_loc)
+        vertica_unloader = VerticaUnloader(db=None, s3_temp_base_loc=mock_s3_temp_base_loc,
+                                           db_conn=mock_db)
         mock_source_records_format.hints = {}
         out = vertica_unloader.known_supported_records_formats_for_unload()
         self.assertEqual(out, [DelimitedRecordsFormat(variant='vertica')])
diff --git a/tests/unit/db/vertica/test_unloader_no_aws_creds.py b/tests/unit/db/vertica/test_unloader_no_aws_creds.py
index fab9cc512..856732c0b 100644
--- a/tests/unit/db/vertica/test_unloader_no_aws_creds.py
+++ b/tests/unit/db/vertica/test_unloader_no_aws_creds.py
@@ -11,7 +11,8 @@ def test_unload_to_s3_directory_with_token(self):
         mock_s3_temp_base_loc = Mock(name='s3_temp_base_loc')
         mock_out = mock_db.execute.return_value
         mock_out.fetchall.return_value = []
-        unloader = VerticaUnloader(db=mock_db, s3_temp_base_loc=mock_s3_temp_base_loc)
+        unloader = VerticaUnloader(db=None, s3_temp_base_loc=mock_s3_temp_base_loc,
+                                   db_conn=mock_db)
         mock_table = Mock(name='table')
         mock_unload_plan = Mock(name='unload_plan')
         mock_schema = Mock(name='schema')
@@ -29,13 +30,17 @@ def test_unload_to_s3_directory_with_token(self):
     @patch("builtins.open", new_callable=mock_open)
     def test_unload_with_no_aws_creds(self,
                                       mock_open):
-        mock_db = Mock(name='db')
+        mock_db = MagicMock(name='db')
         mock_s3_temp_base_loc = MagicMock(name='s3_temp_base_loc')
         mock_target_records_format = Mock(name='target_records_format', spec=DelimitedRecordsFormat)
         mock_target_records_format.hints = {}
-        mock_out = mock_db.execute.return_value
+        mock_connection = MagicMock(name='connection')
+        mock_db.connect.return_value \
+               .__enter__.return_value = mock_connection
+        mock_out = mock_connection.execute.return_value
         mock_out.fetchall.return_value = ['awslib']
-        unloader = VerticaUnloader(db=mock_db, s3_temp_base_loc=mock_s3_temp_base_loc)
+        unloader = VerticaUnloader(db=None, s3_temp_base_loc=mock_s3_temp_base_loc,
+                                   db_conn=mock_db)
         mock_table = Mock(name='table')
         mock_unload_plan = Mock(name='unload_plan')
         mock_unload_plan.records_format = Mock(spec=DelimitedRecordsFormat)
@@ -56,11 +61,15 @@ def test_unload_with_no_aws_creds(self,
                             directory=mock_directory)
 
     def test_s3_export_available_false_no_awslib(self):
-        mock_db = Mock(name='db')
+        mock_db = MagicMock(name='db')
         mock_s3_temp_base_loc = Mock(name='s3_temp_base_loc')
         mock_target_records_format = Mock(name='target_records_format', spec=DelimitedRecordsFormat)
         mock_target_records_format.hints = {}
-        mock_out = mock_db.execute.return_value
+        mock_connection = MagicMock(name='connection')
+        mock_db.connect.return_value \
+               .__enter__.return_value = mock_connection
+        mock_out = mock_connection.execute.return_value
         mock_out.fetchall.return_value = []
-        unloader = VerticaUnloader(db=mock_db, s3_temp_base_loc=mock_s3_temp_base_loc)
+        unloader = VerticaUnloader(db=None, s3_temp_base_loc=mock_s3_temp_base_loc,
+                                   db_conn=mock_db)
         self.assertEqual(False, unloader.s3_export_available())
diff --git a/tests/unit/db/vertica/test_unloader_no_s3.py b/tests/unit/db/vertica/test_unloader_no_s3.py
index d8a9641d6..3e69a91dc 100644
--- a/tests/unit/db/vertica/test_unloader_no_s3.py
+++ b/tests/unit/db/vertica/test_unloader_no_s3.py
@@ -10,7 +10,8 @@ class TestVerticaUnloaderNoS3(unittest.TestCase):
     def test_temporary_unloadable_directory_load_with_no_s3_temp_bucket_configured(self):
         mock_db = Mock(name='db')
         mock_s3_temp_base_loc = None
-        vertica_unloader = VerticaUnloader(db=mock_db, s3_temp_base_loc=mock_s3_temp_base_loc)
+        vertica_unloader = VerticaUnloader(db=None, s3_temp_base_loc=mock_s3_temp_base_loc,
+                                           db_conn=mock_db)
         with self.assertRaises(NoTemporaryBucketConfiguration):
             with vertica_unloader.temporary_unloadable_directory_loc():
                 pass
@@ -22,12 +23,10 @@ def test_can_unload_to_scheme_s3_but_no_s3_export_false(self):
         mock_resultset.fetchall.return_value = []
 
         mock_s3_temp_base_loc = None
-        vertica_unloader = VerticaUnloader(db=mock_db, s3_temp_base_loc=mock_s3_temp_base_loc)
+        vertica_unloader = VerticaUnloader(db=None, s3_temp_base_loc=mock_s3_temp_base_loc,
+                                           db_conn=mock_db)
         self.assertFalse(vertica_unloader.can_unload_to_scheme('s3'))
 
-        mock_db.execute.\
-            assert_called_with("SELECT lib_name from user_libraries where lib_name = 'awslib'")
-
     def test_can_unload_to_scheme_s3_but_with_s3_export_true(self):
         mock_db = Mock(name='db')
         mock_resultset = Mock(name='resultset')
@@ -35,13 +34,14 @@ def test_can_unload_to_scheme_s3_but_with_s3_export_true(self):
         mock_resultset.fetchall.return_value = ['awslib']
 
         mock_s3_temp_base_loc = None
-        vertica_unloader = VerticaUnloader(db=mock_db, s3_temp_base_loc=mock_s3_temp_base_loc)
+        vertica_unloader = VerticaUnloader(db=None, s3_temp_base_loc=mock_s3_temp_base_loc,
+                                           db_conn=mock_db)
         self.assertTrue(vertica_unloader.can_unload_to_scheme('s3'))
 
-        mock_db.execute.\
-            assert_called_with("SELECT lib_name from user_libraries where lib_name = 'awslib'")
+        str_arg = str(mock_db.execute.call_args.args[0])
+        self.assertEqual(str_arg, "SELECT lib_name from user_libraries where lib_name = 'awslib'")
 
     def test_s3_temp_bucket_available_false(self):
         mock_db = Mock(name='db')
-        vertica_unloader = VerticaUnloader(db=mock_db, s3_temp_base_loc=None)
+        vertica_unloader = VerticaUnloader(db=None, s3_temp_base_loc=None, db_conn=mock_db)
         self.assertFalse(vertica_unloader.s3_temp_bucket_available())
diff --git a/tests/unit/db/vertica/test_vertica_db_driver_load.py b/tests/unit/db/vertica/test_vertica_db_driver_load.py
index 1ee13a73d..af7a10c32 100644
--- a/tests/unit/db/vertica/test_vertica_db_driver_load.py
+++ b/tests/unit/db/vertica/test_vertica_db_driver_load.py
@@ -18,9 +18,10 @@ def setUp(self):
         self.mock_s3_temp_base_loc = MagicMock(name='s3_temp_base_loc')
         self.mock_url_resolver = Mock(name='url_resolver')
         self.mock_s3_temp_base_loc.url = 's3://fakebucket/fakedir/fakesubdir/'
-        self.vertica_db_driver = VerticaDBDriver(db=self.mock_db_engine,
+        self.vertica_db_driver = VerticaDBDriver(db=None,
                                                  s3_temp_base_loc=self.mock_s3_temp_base_loc,
-                                                 url_resolver=self.mock_url_resolver)
+                                                 url_resolver=self.mock_url_resolver,
+                                                 db_conn=self.mock_db_engine)
 
         mock_records_unload_plan = create_autospec(RecordsUnloadPlan)
         mock_records_unload_plan.records_format = create_autospec(DelimitedRecordsFormat)
diff --git a/tests/unit/records/schema/field/sqlalchemy/base_test_field_from_sqlalchemy_column.py b/tests/unit/records/schema/field/sqlalchemy/base_test_field_from_sqlalchemy_column.py
index 273a71983..3b5bc9a45 100644
--- a/tests/unit/records/schema/field/sqlalchemy/base_test_field_from_sqlalchemy_column.py
+++ b/tests/unit/records/schema/field/sqlalchemy/base_test_field_from_sqlalchemy_column.py
@@ -32,7 +32,7 @@ def verify(self,
         out = field_from_sqlalchemy_column(self.mock_column, self.mock_driver, self.mock_rep_type)
         mock_RecordsSchemaFieldRepresentation.from_sqlalchemy_column.\
             assert_called_with(self.mock_column,
-                               self.mock_driver.db.dialect,
+                               self.mock_driver.db_engine.dialect,
                                self.mock_rep_type)
         expected = mock_RecordsSchemaField.return_value
         self.mock_field_type = field_type
diff --git a/tests/unit/records/sources/test_table.py b/tests/unit/records/sources/test_table.py
index 027c6c8d8..0c0ac34df 100644
--- a/tests/unit/records/sources/test_table.py
+++ b/tests/unit/records/sources/test_table.py
@@ -30,24 +30,29 @@ def test_to_dataframes_source(self,
                                   mock_DataframesRecordsSource):
         mock_processing_instructions = Mock(name='processing_instructions')
         mock_records_schema = mock_RecordsSchema.from_db_table.return_value
-        mock_db = self.mock_driver.db
+        mock_db_engine = self.mock_driver.db_engine
+        mock_db_conn = self.mock_driver.db_conn
+        mock_connection = MagicMock(name='connection')
+        mock_db_engine.connect.return_value.__enter__.return_value = mock_connection
         mock_column = Mock(name='column')
         mock_columns = [mock_column]
-        mock_db.dialect.get_columns.return_value = mock_columns
+        mock_db_engine.dialect.get_columns.return_value = mock_columns
         mock_quoted_table = mock_quote_schema_and_table.return_value
         mock_chunks = mock_read_sql.return_value
         with self.table_records_source.to_dataframes_source(mock_processing_instructions) as\
                 df_source:
             self.assertEqual(df_source, mock_DataframesRecordsSource.return_value)
-            mock_db.dialect.get_columns.assert_called_with(mock_db,
-                                                           self.mock_table_name,
-                                                           schema=self.mock_schema_name)
+            mock_db_engine.dialect.get_columns.assert_called_with(mock_db_conn,
+                                                                  self.mock_table_name,
+                                                                  schema=self.mock_schema_name)
             mock_RecordsSchema.from_db_table.assert_called_with(self.mock_schema_name,
                                                                 self.mock_table_name,
                                                                 driver=self.mock_driver)
-            mock_read_sql.assert_called_with(f"SELECT * FROM {mock_quoted_table}",
-                                             chunksize=2000000,
-                                             con=mock_db)
+            str_arg = str(mock_read_sql.call_args.args[0])
+            self.assertEqual(str_arg, f"SELECT * FROM {mock_quoted_table}")
+            kwargs = mock_read_sql.call_args.kwargs
+            self.assertEqual(kwargs['con'], mock_db_conn)
+            self.assertEqual(kwargs['chunksize'], 2000000)
             mock_DataframesRecordsSource.\
                 assert_called_with(dfs=ANY,
                                    processing_instructions=mock_processing_instructions,
diff --git a/tests/unit/records/targets/table/test_move_from_dataframes_source.py b/tests/unit/records/targets/table/test_move_from_dataframes_source.py
index 31746874d..98d74ed6f 100644
--- a/tests/unit/records/targets/table/test_move_from_dataframes_source.py
+++ b/tests/unit/records/targets/table/test_move_from_dataframes_source.py
@@ -62,7 +62,7 @@ def test_move_via_insert(self, mock_prep_and_load):
         mock_records_schema = self.mock_dfs_source.initial_records_schema.return_value
         mock_schema_sql = mock_records_schema.to_schema_sql.return_value
         out = self.algo.move()
-        self.mock_tbl.db_driver.assert_called_with(self.mock_tbl.db_engine)
+        self.mock_tbl.db_driver.assert_called_with(db=None, db_engine=self.mock_tbl.db_engine)
         self.mock_dfs_source.initial_records_schema.\
             assert_called_with(self.mock_processing_instructions)
         mock_records_schema.to_schema_sql.assert_called_with(mock_driver,
@@ -79,7 +79,7 @@ def test_load(self, mock_purge_unnamed_unused_columns):
         mock_df = Mock(name='df')
         mock_records_schema = self.mock_dfs_source.initial_records_schema.return_value
         self.mock_dfs_source.dfs = [mock_df]
-        mock_db = self.mock_tbl.db_engine.begin.return_value.__enter__.return_value
+        mock_db = self.mock_tbl.db_engine.connect.return_value.__enter__.return_value
         mock_df_1 = mock_purge_unnamed_unused_columns.return_value
         mock_df_2 = mock_records_schema.assign_dataframe_names.return_value
         mock_df_2.index = [1, 2, 3]
diff --git a/tests/unit/records/targets/table/test_move_from_fileobjs_source.py b/tests/unit/records/targets/table/test_move_from_fileobjs_source.py
index ab4289f0f..1c4587252 100644
--- a/tests/unit/records/targets/table/test_move_from_fileobjs_source.py
+++ b/tests/unit/records/targets/table/test_move_from_fileobjs_source.py
@@ -34,14 +34,14 @@ def setUp(self, mock_RecordsLoadPlan):
                                      processing_instructions=self.mock_processing_instructions)
 
     def test_move(self):
-        mock_db = self.mock_tbl.db_engine.begin.return_value.__enter__.return_value
+        mock_db = self.mock_tbl.db_engine.connect.return_value.__enter__.return_value
         mock_driver = self.mock_tbl.db_driver.return_value
         mock_tweaked_records_schema = mock_driver.tweak_records_schema_for_load.return_value
         mock_schema_sql = mock_tweaked_records_schema.to_schema_sql.return_value
         mock_loader_from_fileobj = mock_driver.loader_from_fileobj.return_value
         mock_import_count = mock_loader_from_fileobj.load_from_fileobj.return_value
         out = self.algo.move()
-        self.mock_tbl.db_driver.assert_called_with(mock_db)
+        self.mock_tbl.db_driver.assert_called_with(db=None, db_conn=mock_db)
         self.mock_RecordsLoadPlan.\
             assert_called_with(records_format=self.mock_fileobjs_source.records_format,
                                processing_instructions=self.mock_processing_instructions)
@@ -62,7 +62,7 @@ def test_move_with_load_failure_dont_recreate_reraise(self):
         class MyException(Exception):
             pass
 
-        mock_db = self.mock_tbl.db_engine.begin.return_value.__enter__.return_value
+        mock_db = self.mock_tbl.db_engine.connect.return_value.__enter__.return_value
         mock_driver = self.mock_tbl.db_driver.return_value
         mock_tweaked_records_schema = mock_driver.tweak_records_schema_for_load.return_value
         mock_schema_sql = mock_tweaked_records_schema.to_schema_sql.return_value
@@ -75,7 +75,7 @@ class MyException(Exception):
 
         mock_loader_from_fileobj.load_failure_exception.assert_called_with()
 
-        self.mock_tbl.db_driver.assert_called_with(mock_db)
+        self.mock_tbl.db_driver.assert_called_with(db=None, db_conn=mock_db)
         mock_tweaked_records_schema.to_schema_sql.\
             assert_called_with(mock_driver,
                                self.mock_tbl.schema_name,
@@ -95,7 +95,7 @@ def test_move_with_load_failure_recreate(self):
         class MyException(Exception):
             pass
 
-        mock_db = self.mock_tbl.db_engine.begin.return_value.__enter__.return_value
+        mock_db = self.mock_tbl.db_engine.connect.return_value.__enter__.return_value
         mock_driver = self.mock_tbl.db_driver.return_value
         mock_tweaked_records_schema = mock_driver.tweak_records_schema_for_load.return_value
         mock_schema_sql = mock_tweaked_records_schema.to_schema_sql.return_value
@@ -112,7 +112,7 @@ class MyException(Exception):
         out = self.algo.move()
         mock_loader_from_fileobj.load_failure_exception.assert_called_with()
 
-        self.mock_tbl.db_driver.assert_called_with(mock_db)
+        self.mock_tbl.db_driver.assert_called_with(db=None, db_conn=mock_db)
         mock_tweaked_records_schema.to_schema_sql.\
             assert_called_with(mock_driver,
                                self.mock_tbl.schema_name,
diff --git a/tests/unit/records/targets/table/test_move_from_records_directory.py b/tests/unit/records/targets/table/test_move_from_records_directory.py
index 37d03ac4c..4c214db63 100644
--- a/tests/unit/records/targets/table/test_move_from_records_directory.py
+++ b/tests/unit/records/targets/table/test_move_from_records_directory.py
@@ -25,7 +25,7 @@ def setUp(self):
     def test_move_happy_path(self,
                              mock_RecordsLoadPlan):
 
-        mock_db = self.mock_tbl.db_engine.begin.return_value.__enter__.return_value
+        mock_db = self.mock_tbl.db_engine.connect.return_value.__enter__.return_value
         mock_driver = self.mock_tbl.db_driver.return_value
         mock_loader = mock_driver.loader.return_value
         mock_records_format = self.mock_override_records_format
@@ -36,7 +36,7 @@ def test_move_happy_path(self,
         mock_plan = mock_RecordsLoadPlan.return_value
         out = self.algo.move()
         self.mock_prep.prep.assert_called_with(schema_sql=mock_schema_sql, driver=mock_driver)
-        self.mock_tbl.db_driver.assert_called_with(mock_db)
+        self.mock_tbl.db_driver.assert_called_with(db=None, db_conn=mock_db)
         mock_driver.tweak_records_schema_for_load.\
             assert_called_with(mock_schema_obj, mock_plan.records_format)
         mock_tweaked_records_schema.to_schema_sql.assert_called_with(mock_driver,
@@ -54,7 +54,7 @@ def test_move_happy_path(self,
     @patch('records_mover.records.targets.table.move_from_records_directory.RecordsLoadPlan')
     def test_move_legacy_schema_sql(self,
                                     mock_RecordsLoadPlan):
-        mock_db = self.mock_tbl.db_engine.begin.return_value.__enter__.return_value
+        mock_db = self.mock_tbl.db_engine.connect.return_value.__enter__.return_value
         mock_driver = self.mock_tbl.db_driver.return_value
         mock_loader = mock_driver.loader.return_value
         mock_records_format = self.mock_override_records_format
@@ -65,7 +65,7 @@ def test_move_legacy_schema_sql(self,
         out = self.algo.move()
 
         self.mock_prep.prep.assert_called_with(schema_sql=mock_schema_sql, driver=mock_driver)
-        self.mock_tbl.db_driver.assert_called_with(mock_db)
+        self.mock_tbl.db_driver.assert_called_with(db=None, db_conn=mock_db)
         self.mock_directory.load_schema_sql_from_sql_file.assert_called_with()
         mock_RecordsLoadPlan.\
             assert_called_with(records_format=mock_records_format,
@@ -87,7 +87,7 @@ def test_move_mutant_records_directory(self,
     @patch('records_mover.records.targets.table.move_from_records_directory.RecordsLoadPlan')
     def test_move_no_override(self,
                               mock_RecordsLoadPlan):
-        mock_db = self.mock_tbl.db_engine.begin.return_value.__enter__.return_value
+        mock_db = self.mock_tbl.db_engine.connect.return_value.__enter__.return_value
         mock_driver = self.mock_tbl.db_driver.return_value
         mock_loader = mock_driver.loader.return_value
         self.mock_directory.load_schema_json_obj.return_value = None
@@ -101,7 +101,7 @@ def test_move_no_override(self,
         self.mock_directory.load_format.\
             assert_called_with(self.mock_processing_instructions.fail_if_dont_understand)
         self.mock_prep.prep.assert_called_with(schema_sql=mock_schema_sql, driver=mock_driver)
-        self.mock_tbl.db_driver.assert_called_with(mock_db)
+        self.mock_tbl.db_driver.assert_called_with(db=None, db_conn=mock_db)
         self.mock_directory.load_schema_sql_from_sql_file.assert_called_with()
 
         mock_RecordsLoadPlan.\
diff --git a/tests/unit/records/targets/table/test_move_from_temp_loc_after_filling_it.py b/tests/unit/records/targets/table/test_move_from_temp_loc_after_filling_it.py
index 1a8b7cb57..66cc02545 100644
--- a/tests/unit/records/targets/table/test_move_from_temp_loc_after_filling_it.py
+++ b/tests/unit/records/targets/table/test_move_from_temp_loc_after_filling_it.py
@@ -31,7 +31,7 @@ def test_init(self,
         mock_directory = mock_RecordsDirectory.return_value
         out = algo.move()
         mock_records_source.compatible_format.assert_called_with(mock_table_target)
-        mock_tbl.db_driver.assert_called_with(mock_tbl.db_engine)
+        mock_tbl.db_driver.assert_called_with(db=None, db_engine=mock_tbl.db_engine)
         mock_RecordsDirectory.assert_called_with(records_loc=mock_temp_loc)
         mock_records_source.move_to_records_directory.\
             assert_called_with(records_directory=mock_directory,
diff --git a/tests/unit/records/targets/table/test_target.py b/tests/unit/records/targets/table/test_target.py
index 20067ccfe..df145a13c 100644
--- a/tests/unit/records/targets/table/test_target.py
+++ b/tests/unit/records/targets/table/test_target.py
@@ -37,21 +37,21 @@ def setUp(self):
     def test_can_move_from_fileobjs_source_yes(self):
         self.assertTrue(self.target.can_move_from_fileobjs_source())
 
-        self.mock_db_driver.assert_called_with(self.mock_db_engine)
+        self.mock_db_driver.assert_called_with(None, db_engine=self.mock_db_engine, db_conn=None)
 
     def test_can_move_directly_from_scheme_no_loader(self):
         mock_driver = self.mock_db_driver.return_value
         mock_driver.loader.return_value = None
         self.assertFalse(self.target.can_move_directly_from_scheme('whatever'))
 
-        self.mock_db_driver.assert_called_with(self.mock_db_engine)
+        self.mock_db_driver.assert_called_with(None, db_engine=self.mock_db_engine, db_conn=None)
 
     def test_known_supported_records_formats_no_loader(self):
         mock_driver = self.mock_db_driver.return_value
         mock_driver.loader.return_value = None
         self.assertEqual([], self.target.known_supported_records_formats())
 
-        self.mock_db_driver.assert_called_with(self.mock_db_engine)
+        self.mock_db_driver.assert_called_with(None, db_engine=self.mock_db_engine, db_conn=None)
 
     def test_can_move_from_format_no_loader(self):
         mock_driver = self.mock_db_driver.return_value
@@ -59,7 +59,7 @@ def test_can_move_from_format_no_loader(self):
         mock_driver.loader.return_value = None
         self.assertFalse(self.target.can_move_from_format(mock_source_records_format))
 
-        self.mock_db_driver.assert_called_with(self.mock_db_engine)
+        self.mock_db_driver.assert_called_with(None, db_engine=self.mock_db_engine, db_conn=None)
 
     def test_can_move_from_format_with_loader_true(self):
         mock_driver = self.mock_db_driver.return_value
@@ -68,7 +68,7 @@ def test_can_move_from_format_with_loader_true(self):
         mock_loader.has_temporary_loadable_directory_loc.return_value = True
         self.assertTrue(self.target.can_move_from_temp_loc_after_filling_it())
 
-        self.mock_db_driver.assert_called_with(self.mock_db_engine)
+        self.mock_db_driver.assert_called_with(None, db_engine=self.mock_db_engine, db_conn=None)
         mock_loader.has_temporary_loadable_directory_loc.assert_called_with()
 
     def test_temporary_loadable_directory_schemer(self):
diff --git a/tests/unit/records/targets/test_factory.py b/tests/unit/records/targets/test_factory.py
index 243582d8b..5bd4366ba 100644
--- a/tests/unit/records/targets/test_factory.py
+++ b/tests/unit/records/targets/test_factory.py
@@ -45,7 +45,8 @@ def test_table(self, mock_TableRecordsTarget):
                                drop_and_recreate_on_load_error=False,
                                existing_table_handling=existing_table_handling,
                                add_group_perms_for=None,
-                               add_user_perms_for=None)
+                               add_user_perms_for=None,
+                               db_conn=None)
         self.assertEqual(table, mock_TableRecordsTarget.return_value)
 
     @patch('records_mover.records.targets.google_sheets.GoogleSheetsRecordsTarget')
diff --git a/tests/unit/records/targets/test_spectrum.py b/tests/unit/records/targets/test_spectrum.py
index 24f3fad2f..bddbdbc84 100644
--- a/tests/unit/records/targets/test_spectrum.py
+++ b/tests/unit/records/targets/test_spectrum.py
@@ -36,7 +36,7 @@ def setUp(self,
 
     def test_init(self):
         self.assertEqual(self.target.records_format, self.records_format)
-        self.assertEqual(self.target.db, self.mock_db)
+        self.assertEqual(self.target.db_engine, self.mock_db)
 
     @patch('records_mover.records.targets.spectrum.quote_schema_and_table')
     def test_pre_load_hook_preps_bucket_with_default_prep(self, mock_quote_schema_and_table):
@@ -44,11 +44,14 @@ def test_pre_load_hook_preps_bucket_with_default_prep(self, mock_quote_schema_an
         mock_cursor = self.target.driver.db_engine.connect.return_value.__enter__.return_value
 
         self.target.pre_load_hook()
-        mock_quote_schema_and_table.assert_called_with(self.target.db,
+        mock_quote_schema_and_table.assert_called_with(None,
                                                        self.target.schema_name,
-                                                       self.target.table_name)
+                                                       self.target.table_name,
+                                                       db_engine=self.target.db_engine)
         mock_cursor.execution_options.assert_called_with(isolation_level='AUTOCOMMIT')
-        mock_cursor.execute.assert_called_with(f"DROP TABLE IF EXISTS {mock_schema_and_table}")
+        arg = mock_cursor.execute.call_args.args[0]
+        arg_str = str(arg)
+        self.assertEqual(arg_str, f"DROP TABLE IF EXISTS {mock_schema_and_table}")
         self.mock_output_loc.purge_directory.assert_called_with()
 
     @patch('records_mover.records.targets.spectrum.RecordsDirectory')
diff --git a/tests/unit/records/targets/test_table_file_objects.py b/tests/unit/records/targets/test_table_file_objects.py
index 1ae9a8ad5..ada5f2c73 100644
--- a/tests/unit/records/targets/test_table_file_objects.py
+++ b/tests/unit/records/targets/test_table_file_objects.py
@@ -37,4 +37,4 @@ def test_can_move_directly_from_scheme(self):
         mock_loader = mock_driver.loader.return_value
         mock_loader.best_scheme_to_load_from.return_value = mock_scheme
         self.assertEqual(True, self.table.can_move_directly_from_scheme(mock_scheme))
-        self.mock_db_driver.assert_called_with(self.mock_db_engine)
+        self.mock_db_driver.assert_called_with(None, db_engine=self.mock_db_engine, db_conn=None)
diff --git a/tests/unit/records/test_prep.py b/tests/unit/records/test_prep.py
index 3e40bd525..16b403d0b 100644
--- a/tests/unit/records/test_prep.py
+++ b/tests/unit/records/test_prep.py
@@ -1,5 +1,5 @@
 import unittest
-from mock import Mock, MagicMock, patch, call
+from mock import Mock, MagicMock, patch
 from records_mover.records.prep import TablePrep
 from records_mover.records.existing_table_handling import ExistingTableHandling
 from records_mover.db import DBDriver
@@ -12,9 +12,10 @@ def setUp(self):
 
     @patch('records_mover.records.prep.quote_schema_and_table')
     def test_prep_table_exists_append_implicit(self, mock_quote_schema_and_table):
-        mock_schema_sql = Mock(name='schema_sql')
+        mock_schema_sql = 'mock_schema_sql'
         mock_driver = Mock(name='driver', spec=DBDriver)
-        mock_driver.db = Mock(name='db')
+        mock_driver.db_conn = MagicMock(name='db')
+        mock_driver.db_engine = MagicMock(name='db_engine')
 
         mock_driver.has_table.return_value = True
         how_to_prep = ExistingTableHandling.APPEND
@@ -24,9 +25,10 @@ def test_prep_table_exists_append_implicit(self, mock_quote_schema_and_table):
 
     @patch('records_mover.records.prep.quote_schema_and_table')
     def test_prep_table_exists_truncate_implicit(self, mock_quote_schema_and_table):
-        mock_schema_sql = Mock(name='schema_sql')
+        mock_schema_sql = 'mock_schema_sql'
         mock_driver = Mock(name='driver', spec=DBDriver)
-        mock_driver.db = Mock(name='db')
+        mock_driver.db_conn = MagicMock(name='db')
+        mock_driver.db_engine = MagicMock(name='db_engine')
 
         mock_quote_schema_and_table
         mock_driver.has_table.return_value = True
@@ -36,16 +38,19 @@ def test_prep_table_exists_truncate_implicit(self, mock_quote_schema_and_table):
 
         self.prep.prep(mock_schema_sql, mock_driver)
 
-        mock_quote_schema_and_table.assert_called_with(mock_driver.db,
+        mock_quote_schema_and_table.assert_called_with(None,
                                                        self.mock_tbl.schema_name,
-                                                       self.mock_tbl.table_name)
-        mock_driver.db.execute.assert_called_with(f"TRUNCATE TABLE {mock_schema_and_table}")
+                                                       self.mock_tbl.table_name,
+                                                       db_engine=mock_driver.db_engine)
+        str_arg = str(mock_driver.db_conn.execute.call_args.args[0])
+        self.assertEqual(str_arg, f"TRUNCATE TABLE {mock_schema_and_table}")
 
     @patch('records_mover.records.prep.quote_schema_and_table')
     def test_prep_table_exists_delete_implicit(self, mock_quote_schema_and_table):
-        mock_schema_sql = Mock(name='schema_sql')
+        mock_schema_sql = 'mock_schema_sql'
         mock_driver = Mock(name='driver', spec=DBDriver)
-        mock_driver.db = Mock(name='db')
+        mock_driver.db_conn = MagicMock(name='db')
+        mock_driver.db_engine = MagicMock(name='db_engine')
 
         mock_quote_schema_and_table
         mock_driver.has_table.return_value = True
@@ -55,17 +60,21 @@ def test_prep_table_exists_delete_implicit(self, mock_quote_schema_and_table):
 
         self.prep.prep(mock_schema_sql, mock_driver)
 
-        mock_quote_schema_and_table.assert_called_with(mock_driver.db,
+        mock_quote_schema_and_table.assert_called_with(None,
                                                        self.mock_tbl.schema_name,
-                                                       self.mock_tbl.table_name)
-        mock_driver.db.execute.assert_called_with(f"DELETE FROM {mock_schema_and_table} WHERE true")
+                                                       self.mock_tbl.table_name,
+                                                       db_engine=mock_driver.db_engine)
+        str_arg = str(mock_driver.db_conn.execute.call_args.args[0])
+        self.assertEqual(str_arg, f"DELETE FROM {mock_schema_and_table} WHERE true")
 
     @patch('records_mover.records.prep.quote_schema_and_table')
     def test_prep_table_exists_drop_implicit(self, mock_quote_schema_and_table):
-        mock_schema_sql = Mock(name='schema_sql')
+        mock_schema_sql = 'mock_schema_sql'
         mock_driver = Mock(name='driver', spec=DBDriver)
+        mock_driver.db_engine = mock_driver
         mock_db = MagicMock(name='db')
-        mock_driver.db = mock_db
+        mock_driver.db_conn = mock_db
+        mock_driver.db_engine = MagicMock(name='db_engine')
 
         mock_quote_schema_and_table
         mock_driver.has_table.return_value = True
@@ -74,32 +83,37 @@ def test_prep_table_exists_drop_implicit(self, mock_quote_schema_and_table):
         self.mock_tbl.existing_table_handling = how_to_prep
 
         self.prep.prep(mock_schema_sql, mock_driver)
-        mock_conn = mock_db.engine.connect.return_value.__enter__.return_value
+        mock_conn = mock_driver.db_engine.connect.return_value.__enter__.return_value
 
-        mock_quote_schema_and_table.assert_called_with(mock_driver.db,
+        mock_quote_schema_and_table.assert_called_with(None,
                                                        self.mock_tbl.schema_name,
-                                                       self.mock_tbl.table_name)
-        mock_conn.execute.assert_has_calls([
-            call(f"DROP TABLE {mock_schema_and_table}"),
-            call(mock_schema_sql),
-        ])
+                                                       self.mock_tbl.table_name,
+                                                       db_engine=mock_driver.db_engine)
+        str_args = [str(call_arg.args[0]) for call_arg in mock_conn.execute.call_args_list]
+        drop_table_str_arg, mock_schema_sql_str_arg = str_args[0], str_args[1]
+        self.assertEqual(drop_table_str_arg, f"DROP TABLE {mock_schema_and_table}")
+        self.assertEqual(mock_schema_sql_str_arg, mock_schema_sql)
         mock_driver.set_grant_permissions_for_groups.\
             assert_called_with(self.mock_tbl.schema_name,
                                self.mock_tbl.table_name,
                                self.mock_tbl.add_group_perms_for,
-                               mock_conn)
+                               None,
+                               db_conn=mock_conn)
         mock_driver.set_grant_permissions_for_users.\
             assert_called_with(self.mock_tbl.schema_name,
                                self.mock_tbl.table_name,
                                self.mock_tbl.add_user_perms_for,
-                               mock_conn)
+                               None,
+                               db_conn=mock_conn)
 
     @patch('records_mover.records.prep.quote_schema_and_table')
     def test_prep_table_not_exists(self, mock_quote_schema_and_table):
-        mock_schema_sql = Mock(name='schema_sql')
+        mock_schema_sql = 'mock_schema_sql'
         mock_driver = Mock(name='driver', spec=DBDriver)
+        mock_driver.db_engine = mock_driver
         mock_db = MagicMock(name='db')
-        mock_driver.db = mock_db
+        mock_driver.db_conn = mock_db
+        mock_driver.db_engine = MagicMock(name='db_engine')
 
         mock_quote_schema_and_table
         mock_driver.has_table.return_value = False
@@ -107,28 +121,32 @@ def test_prep_table_not_exists(self, mock_quote_schema_and_table):
         self.mock_tbl.existing_table_handling = how_to_prep
 
         self.prep.prep(mock_schema_sql, mock_driver)
-        mock_conn = mock_db.engine.connect.return_value.__enter__.return_value
-
-        mock_conn.execute.assert_has_calls([
-            call(mock_schema_sql),
-        ])
+        mock_conn = mock_driver.db_engine.connect.return_value.__enter__.return_value
+        print(mock_conn.execute)
+        str_args = [str(call_arg.args[0]) for call_arg in mock_conn.execute.call_args_list]
+        mock_schema_sql_str_arg = str_args[0]
+        self.assertEqual(mock_schema_sql_str_arg, mock_schema_sql)
         mock_driver.set_grant_permissions_for_groups.\
             assert_called_with(self.mock_tbl.schema_name,
                                self.mock_tbl.table_name,
                                self.mock_tbl.add_group_perms_for,
-                               mock_conn)
+                               None,
+                               db_conn=mock_conn)
         mock_driver.set_grant_permissions_for_users.\
             assert_called_with(self.mock_tbl.schema_name,
                                self.mock_tbl.table_name,
                                self.mock_tbl.add_user_perms_for,
-                               mock_conn)
+                               None,
+                               db_conn=mock_conn)
 
     @patch('records_mover.records.prep.quote_schema_and_table')
     def test_prep_table_exists_drop_explicit(self, mock_quote_schema_and_table):
-        mock_schema_sql = Mock(name='schema_sql')
+        mock_schema_sql = 'mock_schema_sql'
         mock_driver = Mock(name='driver', spec=DBDriver)
         mock_db = MagicMock(name='db')
-        mock_driver.db = mock_db
+        mock_driver.db_conn = mock_db
+        mock_db_engine = MagicMock(name='db_engine')
+        mock_driver.db_engine = mock_db_engine
 
         mock_quote_schema_and_table
         mock_driver.has_table.return_value = True
@@ -138,22 +156,24 @@ def test_prep_table_exists_drop_explicit(self, mock_quote_schema_and_table):
 
         self.prep.prep(mock_schema_sql, mock_driver,
                        existing_table_handling=ExistingTableHandling.DROP_AND_RECREATE)
-        mock_conn = mock_db.engine.connect.return_value.__enter__.return_value
-
-        mock_quote_schema_and_table.assert_called_with(mock_driver.db,
+        mock_conn = mock_db_engine.connect.return_value.__enter__.return_value
+        mock_quote_schema_and_table.assert_called_with(None,
                                                        self.mock_tbl.schema_name,
-                                                       self.mock_tbl.table_name)
-        mock_conn.execute.assert_has_calls([
-            call(f"DROP TABLE {mock_schema_and_table}"),
-            call(mock_schema_sql),
-        ])
+                                                       self.mock_tbl.table_name,
+                                                       db_engine=mock_driver.db_engine)
+        str_args = [str(call_arg.args[0]) for call_arg in mock_conn.execute.call_args_list]
+        drop_table_str_arg, mock_schema_sql_str_arg = str_args[0], str_args[1]
+        self.assertEqual(drop_table_str_arg, f"DROP TABLE {mock_schema_and_table}")
+        self.assertEqual(mock_schema_sql_str_arg, mock_schema_sql)
         mock_driver.set_grant_permissions_for_groups.\
             assert_called_with(self.mock_tbl.schema_name,
                                self.mock_tbl.table_name,
                                self.mock_tbl.add_group_perms_for,
-                               mock_conn)
+                               None,
+                               db_conn=mock_conn)
         mock_driver.set_grant_permissions_for_users.\
             assert_called_with(self.mock_tbl.schema_name,
                                self.mock_tbl.table_name,
                                self.mock_tbl.add_user_perms_for,
-                               mock_conn)
+                               None,
+                               db_conn=mock_conn)
diff --git a/tests/unit/test_check_db_conn_engine.py b/tests/unit/test_check_db_conn_engine.py
new file mode 100644
index 000000000..76bc42eb8
--- /dev/null
+++ b/tests/unit/test_check_db_conn_engine.py
@@ -0,0 +1,84 @@
+import unittest
+from unittest.mock import MagicMock
+from records_mover.check_db_conn_engine import check_db_conn_engine
+import sqlalchemy
+
+
+class TestCheckDbConnEngine(unittest.TestCase):
+    def test_provide_db_only_db_is_engine(self):
+        mock_db = MagicMock(name='db', spec=sqlalchemy.engine.Engine)
+        with self.assertWarns(DeprecationWarning):
+            db, db_conn, db_engine = check_db_conn_engine(db=mock_db)
+        self.assertEqual(mock_db, db)
+        self.assertEqual(mock_db, db_engine)
+        assert not db_conn
+
+    def test_provide_db_only_db_is_conn(self):
+        mock_db = MagicMock(name='db', spec=sqlalchemy.engine.Connection)
+        mock_db_engine = MagicMock(name='db_engine', spec=sqlalchemy.engine.Engine)
+        mock_db.engine = mock_db_engine
+        with self.assertWarns(DeprecationWarning):
+            db, db_conn, db_engine = check_db_conn_engine(db=mock_db)
+        self.assertEqual(mock_db, db)
+        self.assertEqual(mock_db, db_conn)
+        self.assertEqual(mock_db_engine, db_engine)
+
+    def test_provide_db_conn_only(self):
+        mock_db_conn = MagicMock(name='db_conn', spec=sqlalchemy.engine.Connection)
+        mock_db_engine = MagicMock(name='db_engine', spec=sqlalchemy.engine.Engine)
+        mock_db_conn.engine = mock_db_engine
+        db, db_conn, db_engine = check_db_conn_engine(db_conn=mock_db_conn)
+        self.assertEqual(mock_db_conn, db_conn)
+        self.assertEqual(mock_db_engine, db_engine)
+        assert not db
+
+    def test_provide_db_engine_only(self):
+        mock_db_engine = MagicMock(name='db_engine', spec=sqlalchemy.engine.Engine)
+        db, db_conn, db_engine = check_db_conn_engine(db_engine=mock_db_engine)
+        self.assertEqual(mock_db_engine, db_engine)
+        assert not db
+        assert not db_conn
+
+    def test_provide_nothing(self):
+        with self.assertRaises(ValueError):
+            check_db_conn_engine(db=None, db_conn=None, db_engine=None)
+
+    def test_provide_db_and_db_conn(self):
+        mock_db = MagicMock(name='db', spec=sqlalchemy.engine.Engine)
+        mock_db_conn = MagicMock(name='db_conn', spec=sqlalchemy.engine.Connection)
+        with self.assertWarns(DeprecationWarning):
+            db, db_conn, db_engine = check_db_conn_engine(db=mock_db, db_conn=mock_db_conn)
+        self.assertEqual(mock_db, db)
+        self.assertEqual(mock_db_conn, db_conn)
+        self.assertEqual(mock_db, db_engine)
+
+    def test_provide_db_and_db_engine(self):
+        mock_db = MagicMock(name='db', spec=sqlalchemy.engine.Engine)
+        mock_db_engine = MagicMock(name='db_engine', spec=sqlalchemy.engine.Engine)
+        with self.assertWarns(DeprecationWarning):
+            db, db_conn, db_engine = check_db_conn_engine(db=mock_db, db_engine=mock_db_engine)
+        self.assertEqual(mock_db, db)
+        self.assertEqual(mock_db_engine, db_engine)
+        assert not db_conn
+
+    def test_provide_db_conn_and_db_engine(self):
+        mock_db_conn = MagicMock(name='db_conn', spec=sqlalchemy.engine.Connection)
+        mock_db_engine = MagicMock(name='db_engine', spec=sqlalchemy.engine.Engine)
+        mock_db_conn.engine = mock_db_engine
+        db, db_conn, db_engine = check_db_conn_engine(db_conn=mock_db_conn,
+                                                      db_engine=mock_db_engine)
+        self.assertEqual(mock_db_conn, db_conn)
+        self.assertEqual(mock_db_engine, db_engine)
+        assert not db
+
+    def test_provide_db_and_db_conn_and_db_engine(self):
+        mock_db = MagicMock(name='db', spec=sqlalchemy.engine.Engine)
+        mock_db_conn = MagicMock(name='db_conn', spec=sqlalchemy.engine.Connection)
+        mock_db_engine = MagicMock(name='db_engine', spec=sqlalchemy.engine.Engine)
+        with self.assertWarns(DeprecationWarning):
+            db, db_conn, db_engine = check_db_conn_engine(db=mock_db,
+                                                          db_conn=mock_db_conn,
+                                                          db_engine=mock_db_engine)
+        self.assertEqual(mock_db, db)
+        self.assertEqual(mock_db_conn, db_conn)
+        self.assertEqual(mock_db_engine, db_engine)
diff --git a/tests/unit/test_records_hook.py b/tests/unit/test_records_hook.py
index 5339067e1..28b5a7e49 100644
--- a/tests/unit/test_records_hook.py
+++ b/tests/unit/test_records_hook.py
@@ -24,5 +24,5 @@ def test_get_conn(self,
         self.assertEqual(out, mock_Records.return_value)
         args, kwargs = mock_Records.call_args
         mock_db = Mock(name='db')
-        self.assertEqual(mock_db_driver.return_value, kwargs['db_driver'](mock_db))
+        self.assertEqual(mock_db_driver.return_value, kwargs['db_driver'](db_conn=mock_db))
         self.assertEqual(mock_UrlResolver.return_value, kwargs['url_resolver'])
diff --git a/tests/unit/test_session.py b/tests/unit/test_session.py
index ba8204993..9e8ceb39a 100644
--- a/tests/unit/test_session.py
+++ b/tests/unit/test_session.py
@@ -61,13 +61,15 @@ def test_db_driver(self,
         mock_s3_temp_base_loc = mock_url_resolver.directory_url.return_value
         mock_gcs_temp_base_loc = mock_url_resolver.directory_url.return_value
         session = Session(creds=mock_creds)
-        out = session.db_driver(mock_db)
+        out = session.db_driver(None, db_conn=mock_db)
         self.assertEqual(out, mock_db_driver.return_value)
         mock_url_resolver.directory_url.assert_has_calls(
             [call(mock_scratch_s3_url),
              call(mock_scratch_gcs_url)]
         )
-        mock_db_driver.assert_called_with(db=mock_db,
+        mock_db_driver.assert_called_with(db=None,
+                                          db_conn=mock_db,
+                                          db_engine=None,
                                           url_resolver=mock_url_resolver,
                                           s3_temp_base_loc=mock_s3_temp_base_loc,
                                           gcs_temp_base_loc=mock_gcs_temp_base_loc)