[Python] Connecting to MySQL using a token

I've been trying to connect to an RDS instance using the sqlalchemy library using a token. According to the docs it should be possible via the cparams['token'] variable, but doing so I get an error of an unexpected argument.

Going deeper into the docs, there was this code where the connection could be established using the keyword argument attrs_before, but in this case such argument also does not exist.

Does anyone has any idea on how to connect to the DB using a token?

I've trying to do it via such code but with no effect:

import boto3
import struct
from sqlalchemy import create_engine, event
from sqlalchemy.engine.url import URL
from sqlalchemy.engine.url import URL
from sqlalchemy.orm import sessionmaker

SQL_COPT_SS_ACCESS_TOKEN = 1256
session = boto3.session.Session(profile_name='xxx')
rds = session.client('rds')

def get_authentication_token():
return rds.generate_db_auth_token(DBHostname='xxxx', Port=3306, DBUsername='xxx')

engine_url = URL.create(drivername='mysql+pymysql', host='xxxx')
engine = create_engine(engine_url)

#@event.listens_for(engine, 'do_connect')
#def provide_token(dialect, conn_rec, cargs, cparams):
# token_struct = struct.pack(f'<I{len(token)}s', len(token), token)
# cparams['attrs_before'] = {SQL_COPT_SS_ACCESS_TOKEN: token_struct}

@event.listens_for(engine, "do_connect")
def provide_token(dialect, conn_rec, cargs, cparams):
cparams['token'] = get_authentication_token()

Session = sessionmaker(bind=engine)

with Session() as session:
result = session.execute('select now()').first()



To be more precise I get the error: TypeError: __init__() got an unexpected keyword argument 'token'

Continue reading...