I don't have Oracle, but here's a complete runnable example using MySQL
(hopefully the formatting will survive this time):
import sqlalchemy as sa
import MySQLdb
# start with the wrong password to force a connection error
engine = sa.create_engine("mysql://user:wrong@db/db")
@sa.event.listens_for(engine, "do_connect")
def receive_do_connect(dialect, conn_rec, cargs, cparams):
print(f" do_connect ")
print(f"cargs: {cargs}")
print(f"cparams: {cparams}")
try:
return MySQLdb.connect(**cparams)
except Exception as e:
print(f"EXCEPTION: {e}")
# Store the correct password in cparams for future connection
attempts
cparams["passwd"] = "correct"
return MySQLdb.connect(**cparams)
conn1 = engine.connect()
print(f"conn1: {conn1}")
conn2 = engine.connect()
print(f"conn2: {conn2}")
And here's the output:
do_connect
cargs: []
cparams: {'host': 'db', 'db': 'db', 'user': 'user', 'passwd': 'wrong',
'client_flag': 2}
EXCEPTION: (1045, "Access denied for user 'user'@'172.18.0.9' (using
password: YES)")
conn1:
do_connect
cargs: []
cparams: {'host': 'db', 'db': 'db', 'user': 'user', 'passwd': 'correct',
'client_flag': 2}
conn2:
As you can hopefully see, the first connection attempt triggered an
exception. The correct password was then stored in cparams, and the next
time we called engine.connect(), the correct password was already passed in
and no exception was raised.
Simon
On Wed, Mar 23, 2022 at 4:36 AM Srinu Chp wrote:
> Hello Simon,
>
> I tried your suggestion as POC:
>
> def setup_event_handlers(engine):
> @event.listens_for(engine, 'do_connect')
> def receive_do_connect(dialect, conn_rec, cargs, cparams):
> print("inside do_connect")
> print('password %s' % cparams['password'])
> try:
> print("inside try")
> return cx_Oracle.connect(*cargs, **cparams)
> except Exception as e:
> print("inside catch")
> cparams['password'] = "NewPassword"
> return cx_Oracle.connect(*cargs, **cparams)
>
> Every time except block is triggered even I set correct password in Except
> block. As per document once cparams password set it should pass new
> password for new request. Can you please suggest if I miss anything here?
>
> Regards,
> Pydi
>
> On Tuesday, March 22, 2022 at 9:33:57 AM UTC-7 Srinu Chp wrote:
>
>> Hello Simon,
>>
>> Thank you very much for detail information.
>>
>> Regards,
>> Pydi
>>
>> On Tuesday, March 22, 2022 at 3:11:34 AM UTC-7 Simon King wrote:
>>
>>> I don't know anything about Airflow. Are you sure that each of these
>>> tasks is running inside the same Python interpreter/process? I see
>>> Airflow can distribute tasks among workers:
>>>
>>> https://airflow.apache.org/docs/apache-airflow/stable/executor/index.html
>>>
>>> This sounds like a problem that is going to be very specific to your
>>> deployment environment. If you have multiple worker processes, you're
>>> going to need some way to distribute the new password to each of the
>>> workers (eg. a shared cache)
>>>
>>> But regardless of that, you're still not following the pattern from
>>> the documentation. I don't understand why you are creating a new
>>> engine inside your do_connect handler. You should be creating a
>>> cx-Oracle connection and returning it. The parameters passed to the
>>> do_connect handler have already been parsed out of the connection
>>> string. So for example, if your connection string includes a
>>> "password=some_password" parameter, then cparams will have a
>>> "password" key with the value "some_password". The same cparams
>>> dictionary will be passed to the do_connect handler each time, so if
>>> you mutate the dictionary (eg. by updating the "password" key), the
>>> next call to the handler will contain the new value.
>>>
>>> If each task invocation is creating a new engine using a connection
>>> string that is out of date, then none of that will help you, but that
>>> would be an Airflow problem, not an SQLAlchemy problem.
>>>
>>> Simon
>>>
>>> On Mon, Mar 21, 2022 at 8:27 PM Srinu Chp wrote:
>>> >
>>> > Hello Simon,
>>> >
>>> > I tried POC approach in my project where we are using Airflow using
>>> sqlalchemy to connect to db. Here is the event code:
>>> >
>>> > @event.listens_for(engine, "do_connect")
>>> > def receive_do_connect(dialect, conn_rec, cargs, cparams):
>>> > global SQL_ALCHEMY_CONN
>>> > log.info("receive_do_connect called for user AIRFLOW.")
>>> > log.info("user details DB: {}".format(SQL_ALCHEMY_CONN))
>>> > # creating new engine to valide using cx_oracle driver
>>> > engine_new =
>>> create_engine(f'{SQL_ALCHEMY_CONN}/?encoding=UTF-8=UTF-8',
>>> max_identifier_length=128)
>>> > try:
>>> > with engine_new.connect() as conn:
>>> > log.info(conn.scalar("select sysdate from dual"))
>>> > SQL_ALCHEMY_CONN = "testNew_try"
>>> > except Exception as e:
>>> > # check for invalid user/pwd error
>>> > if search('ORA-01017', str(e)):
>>> > log.info("receive_do_connect exception occurred during engine
>>> connection
