Karlo
11/10/2022, 10:40 PMmerlin
11/11/2022, 6:10 AM# VPN_connection is another task that tests a DB service connection
@flow(name="bulk-extracts")
def bulk_extracts(directory_name, scan_minutes):
# TODO: use map to enable concurrent flows
logger = get_run_logger()
extract_folder_items=Path(directory_name).iterdir()
<http://logger.info|logger.info>("Conditional flow dependent on VPN connection begins.")
if VPN_connected(time_period_to_scan_minutes=scan_minutes):
for item in extract_folder_items:
<http://logger.info|logger.info>(f"daily extract file: {item}")
extract_config = ExtractConfig(filepath=item)
with tags(extract_config.dataset_name, "extract"):
extract_write(extract_config)
else:
return Cancelled(message="VPN connection failed, cancelling flow")
Chris Reuter
11/11/2022, 12:31 PMKarlo
11/11/2022, 11:09 PMKalise Richmond
11/12/2022, 1:00 AMKarlo
11/12/2022, 4:48 PM