47 lines
1.3 KiB
Python
47 lines
1.3 KiB
Python
"""Oracle 11.2 thick 连接。环境变量与 rpa_oms / rpa_selection 相同。"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
|
|
CREATE_USER_NAME = "rpa-cart"
|
|
|
|
_client_ready = False
|
|
|
|
|
|
class RepoUnavailable(Exception):
|
|
def __init__(self, message: str = "数据库不可用"):
|
|
self.message = message
|
|
super().__init__(message)
|
|
|
|
|
|
def oracle_dsn() -> tuple[str, str, str]:
|
|
user = os.getenv("RPA_ORACLE_USER", "")
|
|
password = os.getenv("RPA_ORACLE_PASSWORD", "")
|
|
host = os.getenv("RPA_ORACLE_HOST", "")
|
|
if not user or not password or not host:
|
|
raise RepoUnavailable("未配置 RPA_ORACLE_HOST / RPA_ORACLE_USER / RPA_ORACLE_PASSWORD")
|
|
port = os.getenv("RPA_ORACLE_PORT", "1521")
|
|
service = os.getenv("RPA_ORACLE_SERVICE", "rpa")
|
|
return user, password, f"{host}:{port}/{service}"
|
|
|
|
|
|
def connect_from_env():
|
|
user, password, dsn = oracle_dsn()
|
|
_ensure_client()
|
|
import oracledb
|
|
|
|
return oracledb.connect(user=user, password=password, dsn=dsn)
|
|
|
|
|
|
def _ensure_client() -> None:
|
|
global _client_ready
|
|
if _client_ready:
|
|
return
|
|
import oracledb
|
|
|
|
if os.getenv("RPA_ORACLE_THICK", "1") == "1":
|
|
lib_dir = os.getenv("RPA_ORACLE_CLIENT_LIB", "")
|
|
kwargs = {"lib_dir": lib_dir} if lib_dir else {}
|
|
oracledb.init_oracle_client(**kwargs)
|
|
_client_ready = True
|