if __name__ == '__main__':
    from sqlalchemy import create_engine
    from sqlalchemy import text
    db_url = "oracle://user:password@ip:port/?service_name=xxx"
    engine = create_engine(db_url, echo=True)
 
    connection = engine.connect()
    trans = connection.begin()
    param = '"errorCode":431335,"description":"clean_up_queue"'
    sql = "insert into Table_TEST(BUG_NO) values('{0}')".format(param).decode('ISO-8859-1')
    # connection.execute(text(sql))  # insert failure, because text() func will take ':xxx' as a parameter
    # connection.execute(sql)  # insert success, but need pay attention to sql inject
    trans.commit()

Reference:

https://kite.com/python/docs/sqlalchemy.text

更新于 阅读次数

请我喝[茶]~( ̄▽ ̄)~*

Jalen Chu 微信支付

微信支付

Jalen Chu 支付宝

支付宝

Jalen Chu 公众号

公众号