[Python] How can I create a Google Cloud Virtual Machine in Python using a Google Cloud...

My code can create Virtual Machines from a Instance Template. However I need to use a reserved regional IP.

I have created the regional IP with the CLI like this:

gcloud compute addresses create test-regional-ip --region europe-north1 --project=sindre-dev


I list the IPs and see the IP:

gcloud compute addresses list --project=sindre-dev


Result.

NAME ADDRESS/RANGE TYPE PURPOSE NETWORK REGION SUBNET STATUS
test-regional-ip 35.228.123.45 EXTERNAL europe-north1 RESERVED


Now when I try to set the regional IP in my script with external_ip = "35.228.123.45" and run it then I get this error:

TypeError: bad argument type for built-in operation
​

Can anyone help me create a VM from a instance template with regional IP using Python?

create_instance_from_template()

from google.cloud import compute_v1

from src.routes.run_fw_engagements.c_instance_template.helpers.wait_for_extended_operation import \
wait_for_extended_operation


def create_instance_from_template(
project_id: str,
zone: str,
instance_name: str,
instance_template_url: str,
external_ip: str = None) -> compute_v1.Instance:
"""
Creates a Compute Engine VM instance from an instance template and optionally assigns an external IP.

Args:
project_id: ID or number of the project you want to use.
zone: Name of the zone you want to check, for example: us-west3-b
instance_name: Name of the new instance.
instance_template_url: URL of the instance template used for creating the new instance.
It can be a full or partial URL.
external_ip: (Optional) External IP address to be assigned to the instance.

Returns:
Instance object.
"""
instance_client = compute_v1.InstancesClient()

# Initialize the instance resource with a name
instance_resource = compute_v1.Instance(name=instance_name)

# Construct the request for instance insertion
instance_insert_request = compute_v1.InsertInstanceRequest(
project=project_id,
zone=zone,
instance_resource=instance_resource, # Assign the instance resource here
)
instance_insert_request.source_instance_template = instance_template_url

# Check if an external IP was provided
if external_ip:
# Define the network interface with the external IP
network_interface = compute_v1.NetworkInterface()

# Create the AccessConfig for the external IP
print(f"Creating AccessConfig with external_ip: {external_ip}")
access_config = compute_v1.AccessConfig(
name="External NAT",
type_=compute_v1.AccessConfig.Type.ONE_TO_ONE_NAT,
# Instead of directly setting nat_ip, we do it through initialization
)

# Assign the nat_ip directly to the access_config after initialization
access_config.nat_ip = external_ip # Ensure this is a string

# Assign the access config to the network interface
network_interface.access_configs = [access_config]

# Set the network interface to the instance resource
instance_resource.network_interfaces = [network_interface]

# Call the API to create the instance
operation = instance_client.insert(instance_insert_request)

# Wait for the operation to complete
wait_for_extended_operation(operation, "instance creation")

# Return the created instance details
return instance_client.get(project=project_id, zone=zone, instance=instance_name)


if __name__ == '__main__':
# Example usage
project_id = "sindre-dev"
zone = "europe-north1-a"
instance_name = "gno-collector-1"
instance_template_url = f"https://www.googleapis.com/compute/v1/projects/{project_id}/global/instanceTemplates/{instance_name}-v1"
external_ip = "35.228.123.45"

create_instance_from_template(
project_id=project_id,
zone=zone,
instance_name=instance_name,
instance_template_url=instance_template_url,
external_ip=external_ip,
)


wait_for_extended_operation()

from __future__ import annotations

import sys
from typing import Any

from google.api_core.extended_operation import ExtendedOperation


def wait_for_extended_operation(
operation: ExtendedOperation, verbose_name: str = "operation", timeout: int = 300
) -> Any:
"""
Waits for the extended (long-running) operation to complete.

If the operation is successful, it will return its result.
If the operation ends with an error, an exception will be raised.
If there were any warnings during the execution of the operation
they will be printed to sys.stderr.

Args:
operation: a long-running operation you want to wait on.
verbose_name: (optional) a more verbose name of the operation,
used only during error and warning reporting.
timeout: how long (in seconds) to wait for operation to finish.
If None, wait indefinitely.

Returns:
Whatever the operation.result() returns.

Raises:
This method will raise the exception received from `operation.exception()`
or RuntimeError if there is no exception set, but there is an `error_code`
set for the `operation`.

In case of an operation taking longer than `timeout` seconds to complete,
a `concurrent.futures.TimeoutError` will be raised.
"""
result = operation.result(timeout=timeout)

if operation.error_code:
print(
f"Error during {verbose_name}: [Code: {operation.error_code}]: {operation.error_message}",
file=sys.stderr,
flush=True,
)
print(f"Operation ID: {operation.name}", file=sys.stderr, flush=True)
raise operation.exception() or RuntimeError(operation.error_message)

if operation.warnings:
print(f"Warnings during {verbose_name}:\n", file=sys.stderr, flush=True)
for warning in operation.warnings:
print(f" - {warning.code}: {warning.message}", file=sys.stderr, flush=True)

return result

Continue reading...