|
| 1 | +""" |
| 2 | +Example demonstrating full lifecycle of Reserved Fixed IPs resource using async client. |
| 3 | +Includes create, get, list, update (i.e. toggle VIP), and delete operations. |
| 4 | +""" |
| 5 | + |
| 6 | +import asyncio |
| 7 | + |
| 8 | +from gcore import AsyncGcore |
| 9 | +from gcore.pagination import AsyncOffsetPage |
| 10 | +from gcore.types.cloud import ReservedFixedIP |
| 11 | + |
| 12 | + |
| 13 | +async def create_reserved_fixed_ip(*, client: AsyncGcore) -> ReservedFixedIP: |
| 14 | + print("\n=== CREATE RESERVED FIXED IP ===") |
| 15 | + task_list = await client.cloud.reserved_fixed_ips.create( |
| 16 | + type="external", |
| 17 | + ip_family="ipv4", |
| 18 | + is_vip=False, |
| 19 | + ) |
| 20 | + task = await client.cloud.tasks.poll(task_list.tasks[0]) |
| 21 | + if not task.created_resources or not task.created_resources.ports or len(task.created_resources.ports) != 1: |
| 22 | + raise RuntimeError("Expected exactly one port created in the task result") |
| 23 | + port = await client.cloud.reserved_fixed_ips.get(task.created_resources.ports[0]) |
| 24 | + print(f"Created reserved fixed IP: ID={port.port_id}, name={port.name}, IP={port.fixed_ip_address}") |
| 25 | + print("========================") |
| 26 | + return port |
| 27 | + |
| 28 | + |
| 29 | +async def list_reserved_fixed_ips(*, client: AsyncGcore) -> AsyncOffsetPage[ReservedFixedIP]: |
| 30 | + print("\n=== LIST RESERVED FIXED IPS ===") |
| 31 | + reserved_ips = await client.cloud.reserved_fixed_ips.list() |
| 32 | + count = 1 |
| 33 | + async for ip in reserved_ips: |
| 34 | + print(f"{count}. Reserved fixed IP: ID={ip.port_id}, name={ip.name}, status={ip.status}") |
| 35 | + count += 1 |
| 36 | + print("========================") |
| 37 | + return reserved_ips |
| 38 | + |
| 39 | + |
| 40 | +async def get_reserved_fixed_ip(*, client: AsyncGcore, port_id: str) -> ReservedFixedIP: |
| 41 | + print("\n=== GET RESERVED FIXED IP ===") |
| 42 | + reserved_ip = await client.cloud.reserved_fixed_ips.get(port_id) |
| 43 | + print(f"Reserved fixed IP: ID={reserved_ip.port_id}, name={reserved_ip.name}, status={reserved_ip.status}") |
| 44 | + print("========================") |
| 45 | + return reserved_ip |
| 46 | + |
| 47 | + |
| 48 | +async def toggle_reserved_fixed_ip_vip(*, client: AsyncGcore, port_id: str, is_vip: bool) -> ReservedFixedIP: |
| 49 | + print("\n=== TOGGLE RESERVED FIXED IP VIP ===") |
| 50 | + reserved_ip = await client.cloud.reserved_fixed_ips.vip.toggle(port_id, is_vip=is_vip) |
| 51 | + print( |
| 52 | + f"Toggled reserved fixed IP VIP: ID={reserved_ip.port_id}, name={reserved_ip.name}, is_vip={reserved_ip.is_vip}" |
| 53 | + ) |
| 54 | + print("========================") |
| 55 | + return reserved_ip |
| 56 | + |
| 57 | + |
| 58 | +async def list_candidate_ports(*, client: AsyncGcore, port_id: str) -> None: |
| 59 | + print("\n=== LIST CANDIDATE PORTS ===") |
| 60 | + candidate_ports = await client.cloud.reserved_fixed_ips.vip.list_candidate_ports(port_id) |
| 61 | + for count, port in enumerate(candidate_ports.results, 1): |
| 62 | + print(f"{count}. Candidate port: ID={port.port_id}, instance name={port.instance_name}") |
| 63 | + print("========================") |
| 64 | + |
| 65 | + |
| 66 | +async def list_connected_ports(*, client: AsyncGcore, port_id: str) -> None: |
| 67 | + print("\n=== LIST CONNECTED PORTS ===") |
| 68 | + connected_ports = await client.cloud.reserved_fixed_ips.vip.list_connected_ports(port_id) |
| 69 | + for count, port in enumerate(connected_ports.results, 1): |
| 70 | + print(f"{count}. Connected port: ID={port.port_id}, instance name={port.instance_name}") |
| 71 | + print("========================") |
| 72 | + |
| 73 | + |
| 74 | +async def delete_reserved_fixed_ip(*, client: AsyncGcore, port_id: str) -> None: |
| 75 | + print("\n=== DELETE RESERVED FIXED IP ===") |
| 76 | + task_list = await client.cloud.reserved_fixed_ips.delete(port_id) |
| 77 | + await client.cloud.tasks.poll(task_list.tasks[0]) |
| 78 | + print(f"Deleted reserved fixed IP: ID={port_id}") |
| 79 | + print("========================") |
| 80 | + |
| 81 | + |
| 82 | +async def main() -> None: |
| 83 | + # No need to pass the API key explicitly — it will automatically be read from the GCORE_API_KEY environment variable if omitted |
| 84 | + # api_key = os.environ.get("GCORE_API_KEY") |
| 85 | + # Will use Production API URL if omitted |
| 86 | + # base_url = os.environ.get("GCORE_BASE_URL") |
| 87 | + |
| 88 | + gcore = AsyncGcore( |
| 89 | + # api_key=api_key, |
| 90 | + # base_url=base_url, |
| 91 | + ) |
| 92 | + |
| 93 | + fixed_ip = await create_reserved_fixed_ip(client=gcore) |
| 94 | + await list_reserved_fixed_ips(client=gcore) |
| 95 | + await get_reserved_fixed_ip(client=gcore, port_id=fixed_ip.port_id) |
| 96 | + |
| 97 | + # VIP |
| 98 | + await toggle_reserved_fixed_ip_vip(client=gcore, port_id=fixed_ip.port_id, is_vip=True) |
| 99 | + await list_candidate_ports(client=gcore, port_id=fixed_ip.port_id) |
| 100 | + await list_connected_ports(client=gcore, port_id=fixed_ip.port_id) |
| 101 | + # is_vip needs to be false to delete the reserved fixed IP |
| 102 | + await toggle_reserved_fixed_ip_vip(client=gcore, port_id=fixed_ip.port_id, is_vip=False) |
| 103 | + |
| 104 | + await delete_reserved_fixed_ip(client=gcore, port_id=fixed_ip.port_id) |
| 105 | + |
| 106 | + |
| 107 | +if __name__ == "__main__": |
| 108 | + asyncio.run(main()) |
0 commit comments