Add drop_table flow

forklift/forklift/pipeline/flows/drop_table.py

@@ -0,0 +1,15 @@
+from pathlib import Path
+
+from prefect import Flow, Parameter, case
+
+from forklift.pipeline.shared_tasks.control_flow import check_flow_not_running
+from forklift.pipeline.shared_tasks.generic import drop_table_if_exists
+
+with Flow("Drop table") as flow:
+    flow_not_running = check_flow_not_running()
+    with case(flow_not_running, True):
+        database = Parameter("database")
+        table = Parameter("table")
+        drop_table = drop_table_if_exists(database, table)
+
+flow.file_name = Path(__file__).name

forklift/forklift/pipeline/flows_config.py

@@ -17,6 +17,7 @@
 from forklift.pipeline.flows import (
     catches,
     clean_flow_runs,
+    drop_table,
     reset_proxy_pg_database,
     sync_table_from_db_connection,
     sync_table_with_pandas,
@@ -33,6 +34,7 @@ def make_cron_clock_from_run_param_series(s: pd.Series) -> clocks.CronClock:
 def get_flows_to_register():
     catches_flow = deepcopy(catches.flow)
     clean_flow_runs_flow = deepcopy(clean_flow_runs.flow)
+    drop_table_flow = deepcopy(drop_table.flow)
     reset_proxy_pg_database_flow = deepcopy(reset_proxy_pg_database.flow)
     sync_table_from_db_connection_flow = deepcopy(sync_table_from_db_connection.flow)
     sync_table_with_pandas_flow = deepcopy(sync_table_with_pandas.flow)
@@ -64,6 +66,7 @@ def get_flows_to_register():
     flows_to_register = [
         catches_flow,
         clean_flow_runs_flow,
+        drop_table_flow,
         reset_proxy_pg_database_flow,
         sync_table_from_db_connection_flow,
         sync_table_with_pandas_flow,