pulled writing of blob out of list paralleliztion function

cal-itp · Nov 15, 2023 · 40a6e9a · 40a6e9a
1 parent b3b93dc
commit 40a6e9a
Showing 1 changed file with 11 additions and 11 deletions.
diff --git a/runbooks/data/change_blob_storage_class.ipynb b/runbooks/data/change_blob_storage_class.ipynb
@@ -86,20 +86,13 @@
    "outputs": [],
    "source": [
     "# list contents of read bucket in gcs file object\n",
-    "def list_bucket_contents(file_list, blob):\n",
+    "def list_bucket_contents(file_list):\n",
     "    \n",
     "    all_target_files = []\n",
     "    \n",
     "    for item in file_list:\n",
     "        target_file_name = item.name\n",
-    "        all_target_files.append(target_file_name)\n",
-    "        \n",
-    "    # open up new blob in write mode\n",
-    "    with blob.open('w') as f:\n",
-    "\n",
-    "        for result in tqdm(all_target_files):\n",
-    "            f.write(result + '\\n')\n",
-    "        f.close()"
+    "        all_target_files.append(target_file_name)"
    ]
   },
   {
@@ -115,7 +108,7 @@
     "    with concurrent.futures.ThreadPoolExecutor(max_workers=12) as executor:\n",
     "        \n",
     "        # Start the load operations and mark each future with its URL                    \n",
-    "        future_to_list_bucket_contents = {executor.submit(list_bucket_contents, date_files_list, blob)}\n",
+    "        future_to_list_bucket_contents = {executor.submit(list_bucket_contents, date_files_list)}\n",
     "        ## future_to_list_bucket_contents = {executor.submit(list_bucket_contents, list_item): list_item for list_item in date_files_list}\n",
     "        \n",
     "        for future in concurrent.futures.as_completed(future_to_list_bucket_contents):\n",
@@ -236,7 +229,14 @@
     "                blob = bucket.blob(f'{shortened_subpath}/{unique_subpath}.txt')\n",
     "                \n",
     "                # parallelization\n",
-    "                parallelize_list_bucket_contents(all_date_files, bucket, blob)\n",
+    "                parallelize_list_bucket_contents(all_date_files)\n",
+    "                \n",
+    "                # open up new blob in write mode\n",
+    "                with blob.open('w') as f:\n",
+    "\n",
+    "                    for result in tqdm(all_target_files):\n",
+    "                        f.write(result + '\\n')\n",
+    "                    f.close()\n",
     "                                                 \n",
     "            print(f'finished with {subpath}')\n",
     "        \n",