chore(compute): migrate region tags step 1 in metadata (watch maintenance notices) (#13072)

* chore(compute): migrate regions step 1 - add regions to metadata/main.py

* chore(compute): remove whitespace in metadata/vm_identity.py

* chore(compute): fix code style to comply with PEP-8 and code-review-assist

Author: eapl-gemugami

compute/metadata/main.py

@@ -19,8 +19,8 @@
 For more information, see the README.md under /compute.
 """
 
+# [START compute_metadata_watch_maintenance_notices]
 # [START all]
-
 import time
 from typing import Callable, NoReturn, Optional
 
@@ -32,8 +32,7 @@
 
 
 def wait_for_maintenance(callback: Callable[[Optional[str]], None]) -> NoReturn:
-    """
-    Start an infinite loop waiting for maintenance signal.
+    """Start an infinite loop waiting for maintenance signal.
 
     Args:
         callback: Function to be called when a maintenance is scheduled.
@@ -44,6 +43,7 @@ def wait_for_maintenance(callback: Callable[[Optional[str]], None]) -> NoReturn:
     url = METADATA_URL + "instance/maintenance-event"
     last_maintenance_event = None
     # [START hanging_get]
+    # [START compute_metadata_hanging_get_etag]
     last_etag = "0"
 
     while True:
@@ -62,6 +62,7 @@ def wait_for_maintenance(callback: Callable[[Optional[str]], None]) -> NoReturn:
 
         last_etag = r.headers["etag"]
         # [END hanging_get]
+        # [END compute_metadata_hanging_get_etag]
 
         if r.text == "NONE":
             maintenance_event = None
@@ -74,8 +75,7 @@ def wait_for_maintenance(callback: Callable[[Optional[str]], None]) -> NoReturn:
 
 
 def maintenance_callback(event: Optional[str]) -> None:
-    """
-    Example callback function to handle the maintenance event.
+    """Example callback function to handle the maintenance event.
 
     Args:
         event: details about scheduled maintenance.
@@ -93,3 +93,4 @@ def main():
 if __name__ == "__main__":
     main()
 # [END all]
+# [END compute_metadata_watch_maintenance_notices]

compute/metadata/vm_identity.py

@@ -12,20 +12,20 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-"""
-Example of verifying Google Compute Engine virtual machine identity.
+"""Example of verifying Google Compute Engine virtual machine identity.
 
 This sample will work only on a GCE virtual machine, as it relies on
-communication with metadata server (https://cloud.google.com/compute/docs/storing-retrieving-metadata).
+communication with metadata server
+(https://cloud.google.com/compute/docs/storing-retrieving-metadata).
 
-Example is used on: https://cloud.google.com/compute/docs/instances/verifying-instance-identity
+Example is used on:
+https://cloud.google.com/compute/docs/instances/verifying-instance-identity
 """
 import pprint
 
 # [START compute_vm_identity_verify_token]
 import google.auth.transport.requests
 from google.oauth2 import id_token
-
 # [END compute_vm_identity_verify_token]
 
 # [START compute_vm_identity_acquire_token]
@@ -78,17 +78,12 @@ def acquire_token(
     # Extract and return the token from the response.
     r.raise_for_status()
     return r.text
-
-
 # [END compute_vm_identity_acquire_token]
 
 
 # [START compute_vm_identity_verify_token]
-
-
 def verify_token(token: str, audience: str) -> dict:
-    """
-    Verify token signature and return the token payload.
+    """Verify token signature and return the token payload.
 
     Args:
         token: the JSON Web Token received from the metadata server to
@@ -102,8 +97,6 @@ def verify_token(token: str, audience: str) -> dict:
     request = google.auth.transport.requests.Request()
     payload = id_token.verify_token(token, request=request, audience=audience)
     return payload
-
-
 # [END compute_vm_identity_verify_token]
 
 

compute/metadata/vm_identity_test.py

@@ -19,6 +19,7 @@
 
 import vm_identity
 
+
 AUDIENCE = "http://www.testing.com"