Log in using IAM database authentication

import(
"context"
"database/sql"
"fmt"
"log"
"net"
"os"
"cloud.google.com/go/cloudsqlconn"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/stdlib"
)
funcconnectWithConnectorIAMAuthN()(*sql.DB,error){
mustGetenv:=func(kstring)string{
v:=os.Getenv(k)
ifv==""{
log.Fatalf("Warning: %s environment variable not set.",k)
}
returnv
}
// Note: Saving credentials in environment variables is convenient, but not
// secure - consider a more secure solution such as
// Cloud Secret Manager (https://cloud.google.com/secret-manager) to help
// keep secrets safe.
var(
dbUser=mustGetenv("DB_IAM_USER")// e.g. 'service-account-name@project-id.iam'
dbName=mustGetenv("DB_NAME")// e.g. 'my-database'
instanceConnectionName=mustGetenv("INSTANCE_CONNECTION_NAME")// e.g. 'project:region:instance'
usePrivate=os.Getenv("PRIVATE_IP")
)
// WithLazyRefresh() Option is used to perform refresh
// when needed, rather than on a scheduled interval.
// This is recommended for serverless environments to
// avoid background refreshes from throttling CPU.
d,err:=cloudsqlconn.NewDialer (
context.Background(),
cloudsqlconn.WithIAMAuthN (),
cloudsqlconn.WithLazyRefresh (),
)
iferr!=nil{
returnnil,fmt.Errorf("cloudsqlconn.NewDialer: %w",err)
}
varopts[]cloudsqlconn.DialOption
ifusePrivate!=""{
opts=append(opts,cloudsqlconn.WithPrivateIP ())
}
dsn:=fmt.Sprintf("user=%s database=%s",dbUser,dbName)
config,err:=pgx.ParseConfig(dsn)
iferr!=nil{
returnnil,err
}
config.DialFunc=func(ctxcontext.Context,network,instancestring)(net.Conn,error){
returnd.Dial (ctx,instanceConnectionName,opts...)
}
dbURI:=stdlib.RegisterConnConfig(config)
dbPool,err:=sql.Open("pgx",dbURI)
iferr!=nil{
returnnil,fmt.Errorf("sql.Open: %w",err)
}
returndbPool,nil
}

importcom.zaxxer.hikari.HikariConfig;
importcom.zaxxer.hikari.HikariDataSource;
importjavax.sql.DataSource;
publicclass ConnectorIamAuthnConnectionPoolFactoryextendsConnectionPoolFactory{
// Note: Saving credentials in environment variables is convenient, but not
// secure - consider a more secure solution such as
// Cloud Secret Manager (https://cloud.google.com/secret-manager) to help
// keep secrets safe.
privatestaticfinalStringINSTANCE_CONNECTION_NAME=
System.getenv("INSTANCE_CONNECTION_NAME");
privatestaticfinalStringDB_IAM_USER=System.getenv("DB_IAM_USER");
privatestaticfinalStringDB_NAME=System.getenv("DB_NAME");
publicstaticDataSourcecreateConnectionPool(){
// The configuration object specifies behaviors for the connection pool.
HikariConfigconfig=newHikariConfig();
// The following URL is equivalent to setting the config options below:
// jdbc:postgresql:///<DB_NAME>?cloudSqlInstance=<INSTANCE_CONNECTION_NAME>&
// socketFactory=com.google.cloud.sql.postgres.SocketFactory&user=<DB_IAM_USER>&
// password=password
// See the link below for more info on building a JDBC URL for the Cloud SQL JDBC Socket Factory
// https://github.com/GoogleCloudPlatform/cloud-sql-jdbc-socket-factory#creating-the-jdbc-url
// Configure which instance and what database to connect with.
config.setJdbcUrl(String.format("jdbc:postgresql:///%s",DB_NAME));
config.addDataSourceProperty("socketFactory","com.google.cloud.sql.postgres.SocketFactory");
config.addDataSourceProperty("cloudSqlInstance",INSTANCE_CONNECTION_NAME);
// If connecting using automatic database authentication, follow the instructions for
// connecting using the connector, but set the DB_IAM_USER value to an IAM user or
// service account that has been given access to the database.
// See https://cloud.google.com/sql/docs/postgres/iam-logins for more details.
config.addDataSourceProperty("enableIamAuth","true");
config.addDataSourceProperty("user",DB_IAM_USER);
// Password must be set to a nonempty value to bypass driver validation errors.
config.addDataSourceProperty("password","password");
// Explicitly set sslmode to disable to prevent driver from hanging.
// The Java Connector will handle SSL so it is unneccesary to enable it at the driver level.
config.addDataSourceProperty("sslmode","disable");
// cloudSqlRefreshStrategy set to "lazy" is used to perform a
// refresh when needed, rather than on a scheduled interval.
// This is recommended for serverless environments to
// avoid background refreshes from throttling CPU.
config.addDataSourceProperty("cloudSqlRefreshStrategy","lazy");
// ... Specify additional connection properties here.
// ...
// Initialize the connection pool using the configuration object.
returnnewHikariDataSource(config);
}
}

privatestaticfinalStringCONNECTION_NAME=System.getenv("POSTGRES_CONNECTION_NAME");
privatestaticfinalStringDB_NAME=System.getenv("POSTGRES_DB");
privatestaticfinalStringDB_USER=System.getenv("POSTGRES_IAM_USER");
privatestaticfinalStringIP_TYPE=
System.getenv("IP_TYPE")==null?"PUBLIC":System.getenv("IP_TYPE");
// Set up ConnectionFactoryOptions
ConnectionFactoryOptionsoptions=
ConnectionFactoryOptions.builder()
.option(DRIVER,"gcp")
.option(PROTOCOL,"postgresql")
// Password must be set to a nonempty value to bypass driver validation errors
.option(PASSWORD,"password")
.option(USER,DB_USER)
.option(DATABASE,DB_NAME)
.option(HOST,CONNECTION_NAME)
.option(IP_TYPES,IP_TYPE)
.option(ENABLE_IAM_AUTH,true)
.build();
// Initialize connection pool
ConnectionFactoryconnectionFactory=ConnectionFactories.get(options);
ConnectionPoolConfigurationconfiguration=
ConnectionPoolConfiguration.builder(connectionFactory).build();
this.connectionPool=newConnectionPool(configuration);

importos
fromgoogle.cloud.sql.connectorimport Connector, IPTypes
importpg8000
importsqlalchemy
defconnect_with_connector_auto_iam_authn() -> sqlalchemy.engine.base.Engine:
"""
 Initializes a connection pool for a Cloud SQL instance of Postgres.
 Uses the Cloud SQL Python Connector with Automatic IAM Database Authentication.
 """
 # Note: Saving credentials in environment variables is convenient, but not
 # secure - consider a more secure solution such as
 # Cloud Secret Manager (https://cloud.google.com/secret-manager) to help
 # keep secrets safe.
 instance_connection_name = os.environ[
 "INSTANCE_CONNECTION_NAME"
 ] # e.g. 'project:region:instance'
 db_iam_user = os.environ["DB_IAM_USER"] # e.g. 'sa-name@project-id.iam'
 db_name = os.environ["DB_NAME"] # e.g. 'my-database'
 ip_type = IPTypes.PRIVATE if os.environ.get("PRIVATE_IP") else IPTypes.PUBLIC
 # initialize Cloud SQL Python Connector object
 connector = Connector(refresh_strategy="LAZY")
 defgetconn() -> pg8000.dbapi.Connection:
 conn: pg8000.dbapi.Connection = connector.connect(
 instance_connection_name,
 "pg8000",
 user=db_iam_user,
 db=db_name,
 enable_iam_auth=True,
 ip_type=ip_type,
 )
 return conn
 # The Cloud SQL Python Connector can be used with SQLAlchemy
 # using the 'creator' argument to 'create_engine'
 pool = sqlalchemy.create_engine(
 "postgresql+pg8000://",
 creator=getconn,
 # ...
 )
 return pool