Load via INSERT on Redshift when scratch bucket not available

metrics/mypy_high_water_mark

@@ -1 +1 @@
-92.3400
+92.3600

records_mover/db/loader.py

@@ -47,6 +47,9 @@ def temporary_loadable_directory_loc(self) -> Iterator[BaseDirectoryUrl]:
         with TemporaryDirectory(prefix='temporary_loadable_directory_loc') as dirname:
             yield FilesystemDirectoryUrl(dirname)
 
+    def has_temporary_loadable_directory_loc(self) -> bool:
+        return True
+
     @abstractmethod
     def known_supported_records_formats_for_load(self) -> List[BaseRecordsFormat]:
         """Candidates to look through when negotiating a common records format

records_mover/db/redshift/loader.py

@@ -23,10 +23,13 @@ class RedshiftLoader(LoaderFromRecordsDirectory):
     def __init__(self,
                  db: Union[sqlalchemy.engine.Engine, sqlalchemy.engine.Connection],
                  meta: sqlalchemy.MetaData,
+                 s3_temp_base_loc: Optional[BaseDirectoryUrl],
+                 # TODO: Is this silly or?
                  temporary_s3_directory_loc: Callable[[], ContextManager[BaseDirectoryUrl]])\
             -> None:
         self.db = db
         self.meta = meta
+        self.s3_temp_base_loc = s3_temp_base_loc
         self.temporary_s3_directory_loc = temporary_s3_directory_loc
 
     def load(self,
@@ -158,3 +161,6 @@ def best_scheme_to_load_from(self) -> str:
     def temporary_loadable_directory_loc(self) -> Iterator[BaseDirectoryUrl]:
         with self.temporary_s3_directory_loc() as temp_loc:
             yield temp_loc
+
+    def has_temporary_loadable_directory_loc(self) -> bool:
+        return self.s3_temp_base_loc is not None

records_mover/db/redshift/redshift_db_driver.py

@@ -33,6 +33,7 @@ def __init__(self,
         self._redshift_loader =\
             RedshiftLoader(db=db,
                            meta=self.meta,
+                           s3_temp_base_loc=s3_temp_base_loc,
                            temporary_s3_directory_loc=self.temporary_s3_directory_loc)
         self._redshift_unloader =\
             RedshiftUnloader(db=db,

records_mover/db/vertica/unloader.py

@@ -35,6 +35,9 @@ def temporary_loadable_directory_loc(self) -> Iterator[BaseDirectoryUrl]:
             with self.s3_temp_base_loc.temporary_directory() as temp_loc:
                 yield temp_loc
 
+    def has_temporary_loadable_directory_loc(self) -> bool:
+        raise NotImplementedError  # TODO
+
     def aws_creds_sql(self, aws_id: str, aws_secret: str) -> str:
         return """
             ALTER SESSION SET UDPARAMETER FOR awslib aws_id={aws_id};

records_mover/records/mover.py

@@ -2,7 +2,7 @@
                       SupportsToFileobjsSource,
                       FileobjsSource, SupportsToDataframesSource)
 from .targets.base import (RecordsTarget, SupportsMoveFromRecordsDirectory,
-                           SupportsMoveFromTempLocAfterFillingIt,
+                           MightSupportMoveFromTempLocAfterFillingIt,
                            MightSupportMoveFromFileobjsSource,
                            SupportsMoveFromDataframes)
 from .sources import base as sources_base
@@ -123,8 +123,9 @@ def move(records_source: RecordsSource,
                 as fileobjs_source:
             return move(fileobjs_source, records_target, processing_instructions)
     elif (isinstance(records_source, SupportsMoveToRecordsDirectory) and
-          isinstance(records_target, SupportsMoveFromTempLocAfterFillingIt) and
-          records_source.has_compatible_format(records_target)):
+          isinstance(records_target, MightSupportMoveFromTempLocAfterFillingIt) and
+          records_source.has_compatible_format(records_target) and
+          records_target.can_move_from_temp_loc_after_filling_it()):
         logger.info(f"Mover: copying from {records_source} to {records_target} "
                     f"by filling in a temporary location...")
         return records_target.move_from_temp_loc_after_filling_it(records_source,

records_mover/records/targets/base.py

@@ -123,7 +123,16 @@ def can_move_from_fileobjs_source(self) -> bool:
         pass
 
 
-class SupportsMoveFromTempLocAfterFillingIt(NegotiatesRecordsFormat, metaclass=ABCMeta):
+class MightSupportMoveFromTempLocAfterFillingIt(NegotiatesRecordsFormat, metaclass=ABCMeta):
+    # TODO: Make sure Redshift and Vertica target is configured as
+    # well on this.  Do we really not do any of this today?
+    @abstractmethod
+    def can_move_from_temp_loc_after_filling_it(self) -> bool:
+        """Returns True if target as currently configured can be handed a
+        temporary location and fill it.
+        """
+        pass
+
     @abstractmethod
     def move_from_temp_loc_after_filling_it(self,
                                             records_source:

records_mover/records/targets/data_url.py

@@ -1,7 +1,7 @@
 from ..results import MoveResult
 from ..records_directory import RecordsDirectory
 from .base import (SupportsMoveFromDataframes,
-                   SupportsMoveFromTempLocAfterFillingIt,
+                   MightSupportMoveFromTempLocAfterFillingIt,
                    SupportsMoveToRecordsDirectory,
                    SupportsMoveFromRecordsDirectory)
 from ..processing_instructions import ProcessingInstructions
@@ -16,7 +16,7 @@
 
 
 class DataUrlTarget(SupportsMoveFromDataframes,
-                    SupportsMoveFromTempLocAfterFillingIt,
+                    MightSupportMoveFromTempLocAfterFillingIt,
                     SupportsMoveFromRecordsDirectory):
     def __init__(self,
                  output_loc: BaseFileUrl,
@@ -63,6 +63,13 @@ def move_from_records_directory(self,
                               records_format.generate_filename('data'): self.output_loc.url
                           })
 
+    def can_move_from_temp_loc_after_filling_it(self) -> bool:
+        # we can always create a temporary directory
+        return True
+
+    # TODO: Why even does this implement this protocol?  Can I go
+    # ahead and yank it and still have the tests pass?  Try a PR and
+    # see.
     def move_from_temp_loc_after_filling_it(self,
                                             records_source:
                                             SupportsMoveToRecordsDirectory,

records_mover/records/targets/table/move_from_dataframes_source.py

@@ -30,25 +30,29 @@ def __init__(self,
         super().__init__(prep, target_table_details, processing_instructions)
 
     def move(self) -> MoveResult:
-        if len(self.table_target.known_supported_records_formats()) != 0:
-            if self.table_target.can_move_from_fileobjs_source():
-                return self.move_from_dataframes_source_via_fileobjs()
-            else:
-                # Some databases, like Redshift, can't load from a
-                # stream, but can load from files on an object store
-                # they're pointed to.
-                return self.move_from_dataframes_source_via_records_directory()
+        # TODO: Clean up redundant clause
+        if (len(self.table_target.known_supported_records_formats()) != 0 and
+           self.table_target.can_move_from_fileobjs_source()):
+            return self.move_from_dataframes_source_via_fileobjs()
+        elif (len(self.table_target.known_supported_records_formats()) != 0 and
+              self.table_target.can_move_from_fileobjs_source() and
+              self.table_target.can_move_from_temp_loc_after_filling_it()):
+            # Some databases, like Redshift, can't load from a
+            # stream, but can load from files on an object store
+            # they're pointed to.
+            return self.move_from_dataframes_source_via_temporary_records_directory()
         else:
             logger.info("Known formats for target database: "
                         f"{self.table_target.known_supported_records_formats()}")
             logger.info("Table target can move from fileobjs source? "
                         f"{self.table_target.can_move_from_fileobjs_source()}")
             logger.warning("Loading via INSERT statement as this DB "
-                           "driver does not yet support more direct LOAD methods.  "
+                           "driver does not yet support or is not configured for "
+                           "more direct load methods.  "
                            "This may be very slow.")
             return self.move_from_dataframes_source_via_insert()
 
-    def move_from_dataframes_source_via_records_directory(self) -> MoveResult:
+    def move_from_dataframes_source_via_temporary_records_directory(self) -> MoveResult:
         records_format = next(iter(self.table_target.known_supported_records_formats()), None)
         with self.dfs_source.to_fileobjs_source(self.processing_instructions,
                                                 records_format) as fileobjs_source:

records_mover/records/targets/table/target.py

@@ -1,6 +1,6 @@
 from records_mover.records.targets.base import (
     SupportsMoveFromRecordsDirectory,
-    SupportsMoveFromTempLocAfterFillingIt,
+    MightSupportMoveFromTempLocAfterFillingIt,
     MightSupportMoveFromFileobjsSource,
     SupportsMoveFromDataframes,
 )
@@ -31,7 +31,7 @@
 
 
 class TableRecordsTarget(SupportsMoveFromRecordsDirectory,
-                         SupportsMoveFromTempLocAfterFillingIt,
+                         MightSupportMoveFromTempLocAfterFillingIt,
                          MightSupportMoveFromFileobjsSource,
                          SupportsMoveFromDataframes,
                          TargetTableDetails):
@@ -109,6 +109,13 @@ def can_move_from_this_format(self,
             return False
         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)
+        loader = driver.loader()
+        if loader is None:
+            return False
+        return loader.has_temporary_loadable_directory_loc()
+
     def move_from_temp_loc_after_filling_it(self,
                                             records_source:
                                             SupportsMoveToRecordsDirectory,