AlloyDB Private IP Connection test

test_connectivity.py

from datetime import datetime
import os

import asyncpg

async def main() -> None:
    """Basic test to get time from database using native asyncpg connection."""
    async with asyncpg.create_pool(
        user="postgres",
        password="my-password",
        host="10.59.16.2",  # your AlloyDB instance IP address
        port=5432,
        database="postgres",  # your database name
        ssl="require",
    ) as pool:
        # acquire connection from native asyncpg connection pool
        async with pool.acquire() as conn:
            time = await conn.fetchrow("SELECT NOW()")
            print("Current time is ", time[0])
# run main
asyncio.run(main())