Connect using the AlloyDB Language Connectors

  import 
  
 com.zaxxer.hikari.HikariConfig 
 ; 
 import 
  
 com.zaxxer.hikari.HikariDataSource 
 ; 
 public 
  
 class 
 AlloyDbJdbcConnectorDataSourceFactory 
  
 { 
  
 public 
  
 static 
  
 final 
  
 String 
  
 ALLOYDB_DB 
  
 = 
  
 System 
 . 
 getenv 
 ( 
 "ALLOYDB_DB" 
 ); 
  
 public 
  
 static 
  
 final 
  
 String 
  
 ALLOYDB_USER 
  
 = 
  
 System 
 . 
 getenv 
 ( 
 "ALLOYDB_USER" 
 ); 
  
 public 
  
 static 
  
 final 
  
 String 
  
 ALLOYDB_PASS 
  
 = 
  
 System 
 . 
 getenv 
 ( 
 "ALLOYDB_PASS" 
 ); 
  
 public 
  
 static 
  
 final 
  
 String 
  
 ALLOYDB_INSTANCE_NAME 
  
 = 
  
 System 
 . 
 getenv 
 ( 
 "ALLOYDB_INSTANCE_NAME" 
 ); 
  
 static 
  
 HikariDataSource 
  
 createDataSource 
 () 
  
 { 
  
 HikariConfig 
  
 config 
  
 = 
  
 new 
  
 HikariConfig 
 (); 
  
 config 
 . 
 setJdbcUrl 
 ( 
 String 
 . 
 format 
 ( 
 "jdbc:postgresql:///%s" 
 , 
  
 ALLOYDB_DB 
 )); 
  
 config 
 . 
 setUsername 
 ( 
 ALLOYDB_USER 
 ); 
  
 // e.g., "postgres" 
  
 config 
 . 
 setPassword 
 ( 
 ALLOYDB_PASS 
 ); 
  
 // e.g., "secret-password" 
  
 config 
 . 
 addDataSourceProperty 
 ( 
 "socketFactory" 
 , 
  
 "com.google.cloud.alloydb.SocketFactory" 
 ); 
  
 // e.g., "projects/my-project/locations/us-central1/clusters/my-cluster/instances/my-instance" 
  
 config 
 . 
 addDataSourceProperty 
 ( 
 "alloydbInstanceName" 
 , 
  
 ALLOYDB_INSTANCE_NAME 
 ); 
  
 return 
  
 new 
  
 HikariDataSource 
 ( 
 config 
 ); 
  
 } 
 } 
 

  import 
  
 sqlalchemy 
 from 
  
 google.cloud.alloydbconnector 
  
 import 
 Connector 
 def 
  
 create_sqlalchemy_engine 
 ( 
 inst_uri 
 : 
 str 
 , 
 user 
 : 
 str 
 , 
 password 
 : 
 str 
 , 
 db 
 : 
 str 
 , 
 refresh_strategy 
 : 
 str 
 = 
 "background" 
 , 
 ) 
 - 
> tuple 
 [ 
 sqlalchemy 
 . 
 engine 
 . 
 Engine 
 , 
 Connector 
 ]: 
  
 """Creates a connection pool for an AlloyDB instance and returns the pool 
 and the connector. Callers are responsible for closing the pool and the 
 connector. 
 A sample invocation looks like: 
 engine, connector = create_sqlalchemy_engine( 
 inst_uri, 
 user, 
 password, 
 db, 
 ) 
 with engine.connect() as conn: 
 time = conn.execute(sqlalchemy.text("SELECT NOW()")).fetchone() 
 conn.commit() 
 curr_time = time[0] 
 # do something with query result 
 connector.close() 
 Args: 
 instance_uri (str): 
 The instance URI specifies the instance relative to the project, 
 region, and cluster. For example: 
 "projects/my-project/locations/us-central1/clusters/my-cluster/instances/my-instance" 
 user (str): 
 The database user name, e.g., postgres 
 password (str): 
 The database user's password, e.g., secret-password 
 db (str): 
 The name of the database, e.g., mydb 
 refresh_strategy (Optional[str]): 
 Refresh strategy for the AlloyDB Connector. Can be one of "lazy" 
 or "background". For serverless environments use "lazy" to avoid 
 errors resulting from CPU being throttled. 
 """ 
 connector 
 = 
 Connector 
 ( 
 refresh_strategy 
 = 
 refresh_strategy 
 ) 
 # create SQLAlchemy connection pool 
 engine 
 = 
 sqlalchemy 
 . 
 create_engine 
 ( 
 "postgresql+pg8000://" 
 , 
 creator 
 = 
 lambda 
 : 
 connector 
 . 
 connect 
 ( 
 inst_uri 
 , 
 "pg8000" 
 , 
 user 
 = 
 user 
 , 
 password 
 = 
 password 
 , 
 db 
 = 
 db 
 , 
 ), 
 ) 
 engine 
 . 
 dialect 
 . 
 description_encoding 
 = 
 None 
 return 
 engine 
 , 
 connector 
 

  import 
  
 asyncpg 
 import 
  
 sqlalchemy 
 import 
  
 sqlalchemy.ext.asyncio 
 from 
  
 google.cloud.alloydbconnector 
  
 import 
 AsyncConnector 
 async 
 def 
  
 create_sqlalchemy_engine 
 ( 
 inst_uri 
 : 
 str 
 , 
 user 
 : 
 str 
 , 
 password 
 : 
 str 
 , 
 db 
 : 
 str 
 , 
 refresh_strategy 
 : 
 str 
 = 
 "background" 
 , 
 ) 
 - 
> tuple 
 [ 
 sqlalchemy 
 . 
 ext 
 . 
 asyncio 
 . 
 engine 
 . 
 AsyncEngine 
 , 
 AsyncConnector 
 ]: 
  
 """Creates a connection pool for an AlloyDB instance and returns the pool 
 and the connector. Callers are responsible for closing the pool and the 
 connector. 
 A sample invocation looks like: 
 engine, connector = await create_sqlalchemy_engine( 
 inst_uri, 
 user, 
 password, 
 db, 
 ) 
 async with engine.connect() as conn: 
 time = await conn.execute(sqlalchemy.text("SELECT NOW()")).fetchone() 
 curr_time = time[0] 
 # do something with query result 
 await connector.close() 
 Args: 
 instance_uri (str): 
 The instance URI specifies the instance relative to the project, 
 region, and cluster. For example: 
 "projects/my-project/locations/us-central1/clusters/my-cluster/instances/my-instance" 
 user (str): 
 The database user name, e.g., postgres 
 password (str): 
 The database user's password, e.g., secret-password 
 db (str): 
 The name of the database, e.g., mydb 
 refresh_strategy (Optional[str]): 
 Refresh strategy for the AlloyDB Connector. Can be one of "lazy" 
 or "background". For serverless environments use "lazy" to avoid 
 errors resulting from CPU being throttled. 
 """ 
 connector 
 = 
 AsyncConnector 
 ( 
 refresh_strategy 
 = 
 refresh_strategy 
 ) 
 # create SQLAlchemy connection pool 
 engine 
 = 
 sqlalchemy 
 . 
 ext 
 . 
 asyncio 
 . 
 create_async_engine 
 ( 
 "postgresql+asyncpg://" 
 , 
 async_creator 
 = 
 lambda 
 : 
 connector 
 . 
 connect 
 ( 
 inst_uri 
 , 
 "asyncpg" 
 , 
 user 
 = 
 user 
 , 
 password 
 = 
 password 
 , 
 db 
 = 
 db 
 , 
 ), 
 execution_options 
 = 
 { 
 "isolation_level" 
 : 
 "AUTOCOMMIT" 
 }, 
 ) 
 return 
 engine 
 , 
 connector 
 

  import 
  
 ( 
  
 "context" 
  
 "fmt" 
  
 "net" 
  
 "cloud.google.com/go/alloydbconn" 
  
 "github.com/jackc/pgx/v5/pgxpool" 
 ) 
 // connectPgx establishes a connection to your database using pgxpool and the 
 // AlloyDB Go Connector (aka alloydbconn.Dialer) 
 // 
 // The function takes an instance URI, a username, a password, and a database 
 // name. Usage looks like this: 
 // 
 //	pool, cleanup, err := connectPgx( 
 //	  context.Background(), 
 //	  "projects/myproject/locations/us-central1/clusters/mycluster/instances/myinstance", 
 //	  "postgres", 
 //	  "secretpassword", 
 //	  "mydb", 
 //	) 
 // 
 // In addition to a *pgxpool.Pool type, the function returns a cleanup function 
 // that should be called when you're done with the database connection. 
 func 
  
 connectPgx 
 ( 
  
 ctx 
  
 context 
 . 
 Context 
 , 
  
 instURI 
 , 
  
 user 
 , 
  
 pass 
 , 
  
 dbname 
  
 string 
 , 
  
 opts 
  
 ... 
 alloydbconn 
 . 
  Option 
 
 , 
 ) 
  
 ( 
 * 
 pgxpool 
 . 
 Pool 
 , 
  
 func 
 () 
  
 error 
 , 
  
 error 
 ) 
  
 { 
  
 // First initialize the dialer. alloydbconn.NewDialer accepts additional 
  
 // options to configure credentials, timeouts, etc. 
  
 // 
  
 // For details, see: 
  
 // https://pkg.go.dev/cloud.google.com/go/alloydbconn#Option 
  
 d 
 , 
  
 err 
  
 := 
  
 alloydbconn 
 . 
  NewDialer 
 
 ( 
 ctx 
 , 
  
 opts 
 ... 
 ) 
  
 if 
  
 err 
  
 != 
  
 nil 
  
 { 
  
 noop 
  
 := 
  
 func 
 () 
  
 error 
  
 { 
  
 return 
  
 nil 
  
 } 
  
 return 
  
 nil 
 , 
  
 noop 
 , 
  
 fmt 
 . 
 Errorf 
 ( 
 "failed to init Dialer: %v" 
 , 
  
 err 
 ) 
  
 } 
  
 // The cleanup function will stop the dialer's background refresh 
  
 // goroutines. Call it when you're done with your database connection to 
  
 // avoid a goroutine leak. 
  
 cleanup 
  
 := 
  
 func 
 () 
  
 error 
  
 { 
  
 return 
  
 d 
 . 
  Close 
 
 () 
  
 } 
  
 dsn 
  
 := 
  
 fmt 
 . 
 Sprintf 
 ( 
  
 // sslmode is disabled, because the Dialer will handle the SSL 
  
 // connection instead. 
  
 "user=%s password=%s dbname=%s sslmode=disable" 
 , 
  
 user 
 , 
  
 pass 
 , 
  
 dbname 
 , 
  
 ) 
  
 // Prefer pgxpool for applications. 
  
 // For more information, see: 
  
 // https://github.com/jackc/pgx/wiki/Getting-started-with-pgx#using-a-connection-pool 
  
 config 
 , 
  
 err 
  
 := 
  
 pgxpool 
 . 
 ParseConfig 
 ( 
 dsn 
 ) 
  
 if 
  
 err 
  
 != 
  
 nil 
  
 { 
  
 return 
  
 nil 
 , 
  
 cleanup 
 , 
  
 fmt 
 . 
 Errorf 
 ( 
 "failed to parse pgx config: %v" 
 , 
  
 err 
 ) 
  
 } 
  
 // Tell pgx to use alloydbconn.Dialer to connect to the instance. 
  
 config 
 . 
 ConnConfig 
 . 
 DialFunc 
  
 = 
  
 func 
 ( 
 ctx 
  
 context 
 . 
 Context 
 , 
  
 _ 
  
 string 
 , 
  
 _ 
  
 string 
 ) 
  
 ( 
 net 
 . 
 Conn 
 , 
  
 error 
 ) 
  
 { 
  
 return 
  
 d 
 . 
  Dial 
 
 ( 
 ctx 
 , 
  
 instURI 
 ) 
  
 } 
  
 // Establish the connection. 
  
 pool 
 , 
  
 connErr 
  
 := 
  
 pgxpool 
 . 
 NewWithConfig 
 ( 
 ctx 
 , 
  
 config 
 ) 
  
 if 
  
 connErr 
  
 != 
  
 nil 
  
 { 
  
 return 
  
 nil 
 , 
  
 cleanup 
 , 
  
 fmt 
 . 
 Errorf 
 ( 
 "failed to connect: %s" 
 , 
  
 connErr 
 ) 
  
 } 
  
 return 
  
 pool 
 , 
  
 cleanup 
 , 
  
 nil 
 } 
 

  import 
  
 ( 
  
 "database/sql" 
  
 "fmt" 
  
 "cloud.google.com/go/alloydbconn" 
  
 "cloud.google.com/go/alloydbconn/driver/pgxv5" 
 ) 
 // connectDatabaseSQL establishes a connection to your database using the Go 
 // standard library's database/sql package and using the AlloyDB Go Connector 
 // (aka alloydbconn.Dialer) 
 // 
 // The function takes an instance URI, a username, a password, and a database 
 // name. Usage looks like this: 
 // 
 //	db, cleanup, err := connectDatabaseSQL( 
 //	  "projects/myproject/locations/us-central1/clusters/mycluster/instances/myinstance", 
 //	  "postgres", 
 //	  "secretpassword", 
 //	  "mydb", 
 //	) 
 // 
 // In addition to a *db.SQL type, the function returns a cleanup function that 
 // should be called when you're done with the database connection. 
 func 
  
 connectDatabaseSQL 
 ( 
  
 instURI 
 , 
  
 user 
 , 
  
 pass 
 , 
  
 dbname 
  
 string 
 , 
  
 opts 
  
 ... 
 alloydbconn 
 . 
  Option 
 
 , 
 ) 
  
 ( 
 * 
 sql 
 . 
 DB 
 , 
  
 func 
 () 
  
 error 
 , 
  
 error 
 ) 
  
 { 
  
 // First, register the AlloyDB driver. Note, the driver's name is arbitrary 
  
 // and must only match what you use below in sql.Open. Also, 
  
 // pgxv5.RegisterDriver accepts options to configure credentials, timeouts, 
  
 // etc. 
  
 // 
  
 // For details, see: 
  
 // https://pkg.go.dev/cloud.google.com/go/alloydbconn#Option 
  
 // 
  
 // The cleanup function will stop the dialer's background refresh 
  
 // goroutines. Call it when you're done with your database connection to 
  
 // avoid a goroutine leak. 
  
 cleanup 
 , 
  
 err 
  
 := 
  
 pgxv5 
 . 
  RegisterDriver 
 
 ( 
 "alloydb" 
 , 
  
 opts 
 ... 
 ) 
  
 if 
  
 err 
  
 != 
  
 nil 
  
 { 
  
 return 
  
 nil 
 , 
  
 cleanup 
 , 
  
 err 
  
 } 
  
 db 
 , 
  
 err 
  
 := 
  
 sql 
 . 
 Open 
 ( 
  
 "alloydb" 
 , 
  
 fmt 
 . 
 Sprintf 
 ( 
  
 // sslmode is disabled, because the Dialer will handle the SSL 
  
 // connection instead. 
  
 "host=%s user=%s password=%s dbname=%s sslmode=disable" 
 , 
  
 instURI 
 , 
  
 user 
 , 
  
 pass 
 , 
  
 dbname 
 , 
  
 ), 
  
 ) 
  
 return 
  
 db 
 , 
  
 cleanup 
 , 
  
 err 
 }