38 lines
1.0 KiB
Python
38 lines
1.0 KiB
Python
import oracledb
|
|
from logging_config import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
# Initialize thick mode (uses Oracle Instant Client)
|
|
oracledb.init_oracle_client()
|
|
|
|
class OracleDB:
|
|
def __init__(self, config):
|
|
self.config = config
|
|
self.conn = None
|
|
|
|
def connect(self):
|
|
try:
|
|
self.conn = oracledb.connect(
|
|
user=self.config.db_user,
|
|
password=self.config.db_password,
|
|
dsn=self.config.db_dsn
|
|
)
|
|
logger.info("Connected to Oracle DB using oracledb")
|
|
except Exception as e:
|
|
logger.error("Oracle connection failed", exc_info=True)
|
|
raise
|
|
|
|
def fetch_all(self, sql):
|
|
cur = self.conn.cursor()
|
|
cur.execute(sql)
|
|
cols = [c[0] for c in cur.description]
|
|
rows = [dict(zip(cols, r)) for r in cur.fetchall()]
|
|
cur.close()
|
|
return rows
|
|
|
|
def execute(self, sql, params):
|
|
cur = self.conn.cursor()
|
|
cur.execute(sql, params)
|
|
self.conn.commit()
|
|
cur.close() |