fix: customer warehouse bucket (#704)

## Summary

## Checklist
- [ ] Added Unit Tests
- [ ] Covered by existing CI
- [ ] Integration tested
- [ ] Documentation update
<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

- **New Features**
- Improved handling of Google Cloud Storage bucket selection for file
uploads, now automatically using the appropriate warehouse bucket for
each customer.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

<!-- av pr metadata
This information is embedded by the av CLI when creating PRs to track
the status of stacks when using Aviator. Please do not delete or edit
this section of the PR.
```
{"parent":"main","parentHead":"","trunk":"main"}
```
-->

---------

Co-authored-by: Thomas Chow <[email protected]>

Author: tchow-zlai

api/python/ai/chronon/repo/gcp.py

@@ -16,6 +16,7 @@
     check_call,
     check_output,
     extract_filename_from_path,
+    get_customer_warehouse_bucket,
     get_environ_arg,
     retry_decorator,
     split_date_range,
@@ -217,7 +218,6 @@ def generate_dataproc_submitter_args(
     ):
 
         parsed = urlparse(customer_artifact_prefix)
-        customer_bucket_name = parsed.netloc
         source_blob_name = parsed.path.lstrip("/")
 
         if local_files_to_upload is None:
@@ -233,7 +233,7 @@ def generate_dataproc_submitter_args(
             )
             gcs_files.append(
                 GcpRunner.upload_gcs_blob(
-                    customer_bucket_name, source_file, destination_file_path
+                    get_customer_warehouse_bucket(), source_file, destination_file_path
                 )
             )
         gcs_file_args = ",".join(gcs_files)

api/python/ai/chronon/repo/utils.py

@@ -52,6 +52,8 @@ def get_environ_arg(env_name, ignoreError=False) -> str:
         raise ValueError(f"Please set {env_name} environment variable")
     return value
 
+def get_customer_warehouse_bucket() -> str:
+    return f"zipline-warehouse-{get_customer_id()}"
 
 def get_customer_id() -> str:
     return get_environ_arg("CUSTOMER_ID")