GoogleCloudStorage - Handle zip / gzip files flexibly (#937)

* Update release (#894) * Zoom Polls (#886) * Merge main into major-release (#814) * Use black formatting in addition to flake8 (#796) * Run black formatter on entire repository * Update requirements.txt and CONTRIBUTING.md to reflect black format * Use black linting in circleci test job * Use longer variable name to resolve flake8 E741 * Move noqa comments back to proper lines after black reformat * Standardize S3 Prefix Conventions (#803) This PR catches exception errors when a user does not exhaustive access to keys in an S3 bucket * Add Default Parameter Flexibility (#807) Skips over new `/` logic checks if prefix is `None` (which is true by default) * MoveOn Shopify / AK changes (#801) * Add access_token authentication option for Shopify * Remove unnecessary check The access token will either be None or explicitly set; don't worry about an empty string. * Add get_orders function and test * Add get_transactions function and test * Add function and test to get order * style fixes * style fixes --------- Co-authored-by: sjwmoveon <[email protected]> Co-authored-by: Alex French <[email protected]> Co-authored-by: Kathy Nguyen <[email protected]> * Catch File Extensions in S3 Prefix (#809) * add exception handling * Shortened logs for flake8 * add logic for default case * added file logic + note to user * restructured prefix logic This change moves the prefix -> prefix/ logic into a try/except block ... this will be more robust to most use cases, while adding flexibility that we desire for split-permission buckets * drop nested try/catch + add verbose error log * Add error message verbosity Co-authored-by: willyraedy <[email protected]> --------- Co-authored-by: willyraedy <[email protected]> --------- Co-authored-by: Austin Weisgrau <[email protected]> Co-authored-by: Ian <[email protected]> Co-authored-by: Cody Gordon <[email protected]> Co-authored-by: sjwmoveon <[email protected]> Co-authored-by: Alex French <[email protected]> Co-authored-by: Kathy Nguyen <[email protected]> Co-authored-by: willyraedy <[email protected]> * black format * black format * jwt -> s2s oauth * scaffold new functions * add docs * return * DatabaseConnector Interface to Major Release (#815) * Create the DatabaseConnector * Implement DatabaseConnector for the DB connectors * Add DatabaseConnector to std imports * Flake8 fix * Remove reference to padding in copy() * Add database_discover and fix inheritance * Remove strict_length from copy() * Put strict_length back in original order * Remove strict_length stub from BQ * Fix discover_database export statement * Add return annotation to mysql table_exists * Black formatter pass * Add more documentation on when you should use * Add developer notes. * Fix code block documentation * Enhance discover database * Add unit tests for discover database * Fix unit tests * Add two more tests * Reverse Postgres string_length change --------- Co-authored-by: Jason Walker <[email protected]> * add type handling * pass in updated params * move access token function * ok let's rock!! * make changes * pass access token key only * use temporary client to gen token * mock request in constructor * drop unused imports * add changes * scaffolding tests * Add multiple python versions to CI tests (#858) * Add multiple python versions to CI tests * Remove duplicate key * Combine CI jobs * Update ubuntu image and actually install Python versions * Replace pyenv with apt-get to install python versions * Remove sudo * Remove get from 'apt-get' * Update apt before attempting to install * Add ppa/deadsnakes repository * Add prereq * Fix typo * Add -y to install command * Move -y to correct spot * Add more -ys * Add some echoes to debug * Switch back to pyenv approach * Remove tests from circleci config and move to new github actions config Note: no caching yet, this is more of a proof of concept * Split out Mac tests into seaparate file * Set testing environmental variable separately * First attempt to add depdendency cache * Remove windows tests for now * Fix circleci config * Fix circleci for real this time * Add tests on merging of PRs and update readme to show we do not support for Python 3.7 * Enable passing `identifiers` to ActionNetwork `upsert_person()` (#861) * Enable passing `identifiers` to ActionNetwork upsert_person * Remove unused arguments from method self.get_page method doesn't exist and that method call doesn't return anything. The return statement works fine as-is to return all tags and handles pagination on its own. * Include deprecated per_page argument for backwards compatibility Emit a deprecation warning if this argument is used * Include examples in docstring for `identifiers` argument * Expand documentation on ActionNetwork identifiers * Add pre-commit hook config to run flake8 and black on commit (#864) Notes added to README on how to install and set up * black format * black format * jwt -> s2s oauth * scaffold new functions * add docs * return * add type handling * pass in updated params * move access token function * ok let's rock!! * make changes * pass access token key only * use temporary client to gen token * mock request in constructor * drop unused imports * add changes * scaffolding tests * write unit tests * added testing * drop typing (for now) * update docstring typing * add tests * write functions * update typing * add poll results * update table output * fix tests * uhhh run it back * add scope requirements * add to docs We can add more here if folks see fit * one for the money two for the show --------- Co-authored-by: Jason <[email protected]> Co-authored-by: Austin Weisgrau <[email protected]> Co-authored-by: Cody Gordon <[email protected]> Co-authored-by: sjwmoveon <[email protected]> Co-authored-by: Alex French <[email protected]> Co-authored-by: Kathy Nguyen <[email protected]> Co-authored-by: willyraedy <[email protected]> Co-authored-by: Jason Walker <[email protected]> Co-authored-by: Shauna <[email protected]> * Check for empty tables in zoom poll results (#897) Co-authored-by: Jason Walker <[email protected]> * Bump urllib3 from 1.26.5 to 1.26.17 (#901) Bumps [urllib3](https://github.com/urllib3/urllib3) from 1.26.5 to 1.26.17. - [Release notes](https://github.com/urllib3/urllib3/releases) - [Changelog](https://github.com/urllib3/urllib3/blob/main/CHANGES.rst) - [Commits](urllib3/urllib3@1.26.5...1.26.17) --- updated-dependencies: - dependency-name: urllib3 dependency-type: direct:production ... Signed-off-by: dependabot[bot] <[email protected]> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * Add MobileCommons Connector (#896) * mobilecommons class * Update __init__.py * get broadcasts * fix get broadcast request * Add mc_get_request method * Add annotation * Incorporate Daniel's suggestions and finish up get_broadcasts * A few more methods Need to figure out page_count issue * small fix * Remove page_count, use page record num instead * Add in page_count again Not all get responses include num param, but do include page_count. wft * Fix logging numbers * Add create_profile * Fix error message for post request * Start tests * Add some tests * Continue testing * Update test_mobilecommons.py * functionalize status_code check * break out parse_get_request function * fix test data * fix documentation typo * Add several tests * Update mobilecommons.py * Fix limit and pagination logic * debug unit testing * better commenting and logic * Documentation * Add MC to init file * Revert "Merge branch 'main' into cormac-mobilecommons-connector" This reverts commit cad250f, reversing changes made to 493e117. * Revert "Add MC to init file" This reverts commit 493e117. * Revert "Revert "Add MC to init file"" This reverts commit 8f87ec2. * Revert "Revert "Merge branch 'main' into cormac-mobilecommons-connector"" This reverts commit 8190052. * Fix init destruction * fix init yet again * Update testing docs with underscores * Lint * Lint tests * break up long responses * Fix more linting issues * Hopefully last linting issue * DGJKSNCHIVBN * Documentation fixes * Remove note to self * date format * remove random notes * Update test_mobilecommons.py --------- Co-authored-by: sharinetmc <[email protected]> * #741 : Deprecate Slack chat.postMessage `as_user` argument and allow for new authorship arguments (#891) * remove the argument and add a warning that the usage is deprecated * remove usage of as_user from sample code * add in the user customization arguments in lieu of the deprecated as_user argument * add comment regarding the permissions required to use these arguments * use kwargs * surface the whole response * allow usage of the deprecated argument but surface the failed response better * add to retry * delete test file * fix linting * formatting to fix tests * fix if style * add warning for using thread_ts * move the documentation to the optional arguments * #816 Airtable.get_records() fields argument can be either str or list (#892) * all fields to be a str object * remove newline * Nir's actionnetwork changes (#900) * working on adding a functio to an and took care of a lint issues * init * working on all get functions * actionnetwork functions batch 1 is ready * linting and black formatted compliance * removed unwanted/unsused lines * merged updated main * did some linting * added some more get functions to support all ActionNetwork objects (Advocacy Campaigns, Attendances, Campaigns, Custom Fields, Donations, Embeds, Event Campaigns, Events, Forms, Fundraising Pages, Items, Lists, Messages, Metadata, Outreaches, People, Petitions, Queries, Signatures, Submissions, Tags, Taggings, Wrappers) * worked on linting again * fix airtable.insert_records table arg (#907) * Add canales s3 functions (#885) * add raw s3 functions to parsons * add selected functions to s3.py * delte redundant functions and move drop_and_save function to redshift.py * create test file * add s3 unit tests * add rs.drop_and_unload unit test * add printing for debugging * remove testing file * unsaved changes * remove unused packages * remove unneeded module * Bump urllib3 from 1.26.17 to 1.26.18 (#904) Bumps [urllib3](https://github.com/urllib3/urllib3) from 1.26.17 to 1.26.18. - [Release notes](https://github.com/urllib3/urllib3/releases) - [Changelog](https://github.com/urllib3/urllib3/blob/main/CHANGES.rst) - [Commits](urllib3/urllib3@1.26.17...1.26.18) --- updated-dependencies: - dependency-name: urllib3 dependency-type: direct:production ... Signed-off-by: dependabot[bot] <[email protected]> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: sharinetmc <[email protected]> * New connector for working with the Catalist Match API (#912) * Enable api_connector to return error message in `text` attribute Some API error responses contain the error message in the `text` attribute, so this update makes it possible to fetch that message if it exists. * New connector to work with the Catalist Match API * Add pytest-mock to requirements to support mocking in pytests * Tests on the catalist match connector * More open ended pytest-mock version for compatibility * Expand docstring documetation based on feedback in PR * More verbose error on match failure * Parameterize template_id variable * Expand docstrings on initial setup * Include Catalist documentation rst file * Enhancement: Action Network Connector: Added unpack_statistics param in get_messages method (#917) * Adds parameter to get_messages This adds the ability to unpack the statistics which are returned as a nested dictionary in the response. * added unpack_statistics to an.get_messages() * added parameters to get_messages and built tests * changes unpack_statistics to False by default. * added tbl variable * formatted with black * fixed docs --------- Co-authored-by: mattkrausse <[email protected]> * Adding rename_columns method to Parsons Table (#923) * added rename_columns for multiple cols * linted * added clarification to docs about dict structure * updated docs --------- Co-authored-by: mattkrausse <[email protected]> * Add http response to update_mailer (#924) Without returning the response, or at least the status code, it's impossible to check for errors. * Enable passing arbitrary additional fields to NGPVAN person match API (#916) * match gcs api to s3 * wip * two different functions * use csv as default * drop unused var * add docs * use temp file * add comments * wip * add docs + replicate in gzip * boy howdy! * set timeout * Revert "Merge branch 'main' into ianferguson/gcs-pathing" This reverts commit 5b1ef6e, reversing changes made to f0eb3d6. * black format --------- Signed-off-by: dependabot[bot] <[email protected]> Co-authored-by: Kasia Hinkson <[email protected]> Co-authored-by: Jason <[email protected]> Co-authored-by: Austin Weisgrau <[email protected]> Co-authored-by: Cody Gordon <[email protected]> Co-authored-by: sjwmoveon <[email protected]> Co-authored-by: Alex French <[email protected]> Co-authored-by: Kathy Nguyen <[email protected]> Co-authored-by: willyraedy <[email protected]> Co-authored-by: Jason Walker <[email protected]> Co-authored-by: Shauna <[email protected]> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: Cormac Martinez del Rio <[email protected]> Co-authored-by: sharinetmc <[email protected]> Co-authored-by: Angela Gloyna <[email protected]> Co-authored-by: NirTatcher <[email protected]> Co-authored-by: justicehaze <[email protected]> Co-authored-by: mattkrausse <[email protected]> Co-authored-by: mkrausse-ggtx <[email protected]> Co-authored-by: Sophie Waldman <[email protected]>
move-coop · Dec 1, 2023 · 8b843f2 · 8b843f2
1 parent 0e2d848
commit 8b843f2
Show file tree

Hide file tree

Showing 2 changed files with 114 additions and 22 deletions.
diff --git a/parsons/google/google_bigquery.py b/parsons/google/google_bigquery.py
@@ -324,6 +324,9 @@ def copy_from_gcs(
         quote: Optional[str] = None,
         schema: Optional[List[dict]] = None,
         job_config: Optional[LoadJobConfig] = None,
+        force_unzip_blobs: bool = False,
+        compression_type: str = "gzip",
+        new_file_extension: str = "csv",
         **load_kwargs,
     ):
         """
@@ -371,6 +374,12 @@ def copy_from_gcs(
                 on the BigQuery client. The function will create its own if not provided. Note
                 if there are any conflicts between the job_config and other parameters, the
                 job_config values are preferred.
+            force_unzip_blobs: bool
+                If True, target blobs will be unzipped before being loaded to BigQuery.
+            compression_type: str
+                Accepts `zip` or `gzip` values to differentially unzip a compressed blob in cloud storage.
+            new_file_extension: str
+                Provides a file extension if a blob is decompressed and rewritten to cloud storage.
             **load_kwargs: kwargs
                 Other arguments to pass to the underlying load_table_from_uri call on the BigQuery
                 client.
@@ -406,15 +415,32 @@ def copy_from_gcs(
         # load CSV from Cloud Storage into BigQuery
         table_ref = get_table_ref(self.client, table_name)
 
-        load_job = self.client.load_table_from_uri(
-            source_uris=gcs_blob_uri,
-            destination=table_ref,
-            job_config=job_config,
-            **load_kwargs,
-        )
-
         try:
-            load_job.result()
+            if force_unzip_blobs:
+                self.copy_large_compressed_file_from_gcs(
+                    gcs_blob_uri=gcs_blob_uri,
+                    table_name=table_name,
+                    if_exists=if_exists,
+                    max_errors=max_errors,
+                    data_type=data_type,
+                    csv_delimiter=csv_delimiter,
+                    ignoreheader=ignoreheader,
+                    nullas=nullas,
+                    allow_quoted_newlines=allow_quoted_newlines,
+                    quote=quote,
+                    schema=schema,
+                    job_config=job_config,
+                    compression_type=compression_type,
+                    new_file_extension=new_file_extension,
+                )
+            else:
+                load_job = self.client.load_table_from_uri(
+                    source_uris=gcs_blob_uri,
+                    destination=table_ref,
+                    job_config=job_config,
+                    **load_kwargs,
+                )
+                load_job.result()
         except exceptions.BadRequest as e:
             if "one of the files is larger than the maximum allowed size." in str(e):
                 logger.debug(
@@ -433,6 +459,8 @@ def copy_from_gcs(
                     quote=quote,
                     schema=schema,
                     job_config=job_config,
+                    compression_type=compression_type,
+                    new_file_extension=new_file_extension,
                 )
             elif "Schema has no field" in str(e):
                 logger.debug(f"{gcs_blob_uri.split('/')[-1]} is empty, skipping file")
@@ -465,6 +493,8 @@ def copy_large_compressed_file_from_gcs(
         quote: Optional[str] = None,
         schema: Optional[List[dict]] = None,
         job_config: Optional[LoadJobConfig] = None,
+        compression_type: str = "gzip",
+        new_file_extension: str = "csv",
         **load_kwargs,
     ):
         """
@@ -513,6 +543,10 @@ def copy_large_compressed_file_from_gcs(
                 on the BigQuery client. The function will create its own if not provided. Note
                 if there are any conflicts between the job_config and other parameters, the
                 job_config values are preferred.
+            compression_type: str
+                Accepts `zip` or `gzip` values to differentially unzip a compressed blob in cloud storage.
+            new_file_extension: str
+                Provides a file extension if a blob is decompressed and rewritten to cloud storage.
             **load_kwargs: kwargs
                 Other arguments to pass to the underlying load_table_from_uri call on the BigQuery
                 client.
@@ -557,7 +591,8 @@ def copy_large_compressed_file_from_gcs(
             uncompressed_gcs_uri = gcs.unzip_blob(
                 bucket_name=old_bucket_name,
                 blob_name=old_blob_name,
-                new_file_extension="csv",
+                new_file_extension=new_file_extension,
+                compression_type=compression_type,
             )
 
             logger.debug(

diff --git a/parsons/google/google_cloud_storage.py b/parsons/google/google_cloud_storage.py
@@ -8,7 +8,7 @@
 import time
 import uuid
 import gzip
-import shutil
+import zipfile
 from typing import Optional
 
 logger = logging.getLogger(__name__)
@@ -431,6 +431,7 @@ def copy_bucket_to_gcs(
                 },
                 "gcs_data_sink": {
                     "bucket_name": gcs_sink_bucket,
+                    "path": destination_path,
                 },
             }
 
@@ -452,7 +453,6 @@ def copy_bucket_to_gcs(
 
         while polling:
             if latest_operation_name:
-
                 operation = client.get_operation({"name": latest_operation_name})
 
                 if not operation.done:
@@ -520,6 +520,7 @@ def unzip_blob(
         self,
         bucket_name: str,
         blob_name: str,
+        compression_type: str = "gzip",
         new_filename: Optional[str] = None,
         new_file_extension: Optional[str] = None,
     ) -> str:
@@ -535,6 +536,9 @@ def unzip_blob(
             blob_name: str
                 Blob name in GCS bucket
 
+            compression_type: str
+                Either `zip` or `gzip`
+
             new_filename: str
                 If provided, replaces the existing blob name
                 when the decompressed file is uploaded
@@ -547,29 +551,82 @@ def unzip_blob(
             String representation of decompressed GCS URI
         """
 
+        compression_params = {
+            "zip": {
+                "file_extension": ".zip",
+                "compression_function": self.__zip_decompress_and_write_to_gcs,
+                "read": "r",
+            },
+            "gzip": {
+                "file_extension": ".gz",
+                "compression_function": self.__gzip_decompress_and_write_to_gcs,
+            },
+        }
+
+        file_extension = compression_params[compression_type]["file_extension"]
+        compression_function = compression_params[compression_type][
+            "compression_function"
+        ]
+
         compressed_filepath = self.download_blob(
             bucket_name=bucket_name, blob_name=blob_name
         )
 
-        decompressed_filepath = compressed_filepath.replace(".gz", "")
+        decompressed_filepath = compressed_filepath.replace(file_extension, "")
         decompressed_blob_name = (
-            new_filename if new_filename else blob_name.replace(".gz", "")
+            new_filename if new_filename else blob_name.replace(file_extension, "")
         )
         if new_file_extension:
             decompressed_filepath += f".{new_file_extension}"
             decompressed_blob_name += f".{new_file_extension}"
 
         logger.debug("Decompressing file...")
+        compression_function(
+            compressed_filepath=compressed_filepath,
+            decompressed_filepath=decompressed_filepath,
+            decompressed_blob_name=decompressed_blob_name,
+            bucket_name=bucket_name,
+            new_file_extension=new_file_extension,
+        )
+
+        return self.format_uri(bucket=bucket_name, name=decompressed_blob_name)
+
+    def __gzip_decompress_and_write_to_gcs(self, **kwargs):
+        """
+        Handles `.gzip` decompression and streams blob contents
+        to a decompressed storage object
+        """
+
+        compressed_filepath = kwargs.pop("compressed_filepath")
+        decompressed_blob_name = kwargs.pop("decompressed_blob_name")
+        bucket_name = kwargs.pop("bucket_name")
+
         with gzip.open(compressed_filepath, "rb") as f_in:
-            with open(decompressed_filepath, "wb") as f_out:
-                shutil.copyfileobj(f_in, f_out)
+            logger.debug(
+                f"Uploading uncompressed file to GCS: {decompressed_blob_name}"
+            )
+            bucket = self.get_bucket(bucket_name=bucket_name)
+            blob = storage.Blob(name=decompressed_blob_name, bucket=bucket)
+            blob.upload_from_file(file_obj=f_in, rewind=True, timeout=3600)
+
+    def __zip_decompress_and_write_to_gcs(self, **kwargs):
+        """
+        Handles `.zip` decompression and streams blob contents
+        to a decompressed storage object
+        """
+
+        compressed_filepath = kwargs.pop("compressed_filepath")
+        decompressed_blob_name = kwargs.pop("decompressed_blob_name")
+        decompressed_blob_in_archive = decompressed_blob_name.split("/")[-1]
+        bucket_name = kwargs.pop("bucket_name")
+
+        # Unzip the archive
+        with zipfile.ZipFile(compressed_filepath) as path_:
+            # Open the underlying file
+            with path_.open(decompressed_blob_in_archive) as f_in:
                 logger.debug(
                     f"Uploading uncompressed file to GCS: {decompressed_blob_name}"
                 )
-                self.put_blob(
-                    bucket_name=bucket_name,
-                    blob_name=decompressed_blob_name,
-                    local_path=decompressed_filepath,
-                )
-
-        return self.format_uri(bucket=bucket_name, name=decompressed_blob_name)
+                bucket = self.get_bucket(bucket_name=bucket_name)
+                blob = storage.Blob(name=decompressed_blob_name, bucket=bucket)
+                blob.upload_from_file(file_obj=f_in, rewind=True, timeout=3600)