Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

docs: add direct psycopg2 IAM authentication test #119

Merged
merged 20 commits into from
Sep 18, 2023
Merged

Conversation

jackwotherspoon
Copy link
Collaborator

@jackwotherspoon jackwotherspoon commented Sep 18, 2023

Adding sample for direct TCP connection for psycopg2 with automatic IAM database authentication through the use of SQLAlchemy's do_connect event listener.

Region tags should result in the following code sample:

import sqlalchemy
from sqlalchemy import event

import google.auth
from google.auth.credentials import Credentials
from google.auth.transport.requests import Request

# initialize Google Auth creds
creds, _ = google.auth.default(
    scopes=["https://www.googleapis.com/auth/cloud-platform"]
)

def get_authentication_token(credentials: Credentials) -> str:
    """Get OAuth2 access token to be used for IAM database authentication"""
    # refresh credentials if expired
    if not credentials.valid:
        request = Request()
        credentials.refresh(request)
    return credentials.token

engine = sqlalchemy.create_engine(
    # Equivalent URL:
    # postgresql+psycopg2://<user>:empty@<host>:<port>/<db_name>
    sqlalchemy.engine.url.URL.create(
        drivername="postgresql+psycopg2",
        username=user,  # IAM db user, [email protected]
        password="",  # placeholder to be replaced with OAuth2 token
        host=ip_address,  # AlloyDB instance IP address
        port=5432,
        database=db_name,  # "my-database-name"
    ),
    connect_args={"sslmode": "require"},
)

# set 'do_connect' event listener to replace password with OAuth2 token
@event.listens_for(engine, "do_connect")
def auto_iam_authentication(dialect, conn_rec, cargs, cparams) -> None:
    cparams["password"] = get_authentication_token(creds)

# use connection from connection pool to query AlloyDB database
with engine.connect() as conn:
    time = conn.execute(sqlalchemy.text("SELECT NOW()")).fetchone()
    conn.commit()
    print("Current time is ", time[0])

@jackwotherspoon jackwotherspoon self-assigned this Sep 18, 2023
@jackwotherspoon jackwotherspoon requested a review from a team as a code owner September 18, 2023 16:02
@enocom enocom changed the title chore: add direct psycopg2 IAM authentication test docs: add direct psycopg2 IAM authentication test Sep 18, 2023
@jackwotherspoon jackwotherspoon merged commit b15a3ba into main Sep 18, 2023
15 checks passed
@jackwotherspoon jackwotherspoon deleted the direct-samples branch September 18, 2023 21:00
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

Successfully merging this pull request may close these issues.

2 participants