Update properties of a GCS bucket. Args: bucket_name (str): The name of the GCS bucket to update. credentials (Credentials): The credentials to use for the request. versioning_enabled (bool, optional): Whether to enable versioning for the bucket. uniform_bucket_lev
(
*,
bucket_name: str,
credentials: Credentials,
versioning_enabled: bool | None = None,
uniform_bucket_level_access_enabled: bool | None = None,
)
| 110 | |
| 111 | |
| 112 | def update_bucket( |
| 113 | *, |
| 114 | bucket_name: str, |
| 115 | credentials: Credentials, |
| 116 | versioning_enabled: bool | None = None, |
| 117 | uniform_bucket_level_access_enabled: bool | None = None, |
| 118 | ) -> dict: |
| 119 | """Update properties of a GCS bucket. |
| 120 | |
| 121 | Args: |
| 122 | bucket_name (str): The name of the GCS bucket to update. |
| 123 | credentials (Credentials): The credentials to use for the request. |
| 124 | versioning_enabled (bool, optional): Whether to enable versioning for the |
| 125 | bucket. |
| 126 | uniform_bucket_level_access_enabled (bool, optional): Whether to enable |
| 127 | uniform bucket-level access. |
| 128 | |
| 129 | Returns: |
| 130 | dict: Dictionary indicating success or error. |
| 131 | """ |
| 132 | try: |
| 133 | gcs_client = client.get_gcs_client(credentials=credentials) |
| 134 | bucket = gcs_client.get_bucket(bucket_name) |
| 135 | if versioning_enabled is not None: |
| 136 | bucket.versioning_enabled = versioning_enabled |
| 137 | if uniform_bucket_level_access_enabled is not None: |
| 138 | bucket.iam_configuration.uniform_bucket_level_access_enabled = ( |
| 139 | uniform_bucket_level_access_enabled |
| 140 | ) |
| 141 | |
| 142 | if ( |
| 143 | versioning_enabled is not None |
| 144 | or uniform_bucket_level_access_enabled is not None |
| 145 | ): |
| 146 | bucket.patch() |
| 147 | |
| 148 | return { |
| 149 | "status": "SUCCESS", |
| 150 | "results": f"Bucket {bucket.name} updated successfully.", |
| 151 | } |
| 152 | except Exception as ex: |
| 153 | return { |
| 154 | "status": "ERROR", |
| 155 | "error_details": str(ex), |
| 156 | } |
| 157 | |
| 158 | |
| 159 | def delete_bucket(*, bucket_name: str, credentials: Credentials) -> dict: |