Cloud SQL Connector Private IP

connector_private_ip.py

from google.cloud.sql.connector import Connector, IPTypes
import sqlalchemy

def init_connection_pool(connector: Connector):
    def getconn():
       connection = connector.connect(
           "PROJECT:REGION:INSTANCE_NAME",
           "pg8000",
           user="my-user",
           password="my-password",
           db="my-database",
       )
       return connection
       
    pool = sqlalchemy.create_engine(
        "postgresql+pg8000://",
        creator=getconn,
    )
    return pool
    
def main():
  # create Connector for Private IP connections
  connector = Connector(ip_type=IPTypes.PRIVATE)
  pool = init_connection_pool(connector)
  # interact with Cloud SQL database using connection pool
  with pool.connect() as db_conn:
    # insert statement
    insert_stmt = sqlalchemy.text(
        "INSERT INTO my_table (id, title) VALUES (:id, :title)",
    )
    # insert into database
    db_conn.execute(insert_stmt, parameters={"id": "book1", "title": "Book One"})

    # commit transaction (SQLAlchemy v2.X.X is commit as you go)
    db_conn.commit()
  connector.close()