1+ from sqlalchemy .sql import text
2+ from .base import BaseInspector
3+ from sqlalchemy .engine import reflection
4+ from urllib .parse import quote_plus
5+
6+ class OracleInspector (BaseInspector ):
7+ """Oracle元数据获取实现"""
8+
9+ def __init__ (self , host : str , port : int , database : str ,
10+ username : str , password : str , schema_name : str = None , ** kwargs ):
11+ super ().__init__ (host , port , database , username , password , schema_name )
12+ self .schema_name = username .upper () # Oracle模式名通常与用户名一致[3,8](@ref)
13+
14+ def build_conn_str (self , host : str , port : int , database : str ,
15+ username : str , password : str ) -> str :
16+ # 使用cx_Oracle驱动,支持SID或Service Name[6,8](@ref)
17+ return (
18+ f"oracle+cx_oracle://{ quote_plus (username )} :{ quote_plus (password )} "
19+ f"@{ host } :{ port } /?service_name={ database } "
20+ )
21+
22+ def get_table_names (self , inspector : reflection .Inspector ) -> list [str ]:
23+ return inspector .get_table_names (schema = self .schema_name ) # 需指定schema[3](@ref)
24+
25+ def get_table_comment (self , inspector : reflection .Inspector ,
26+ table_name : str ) -> str :
27+ # 查询ALL_TAB_COMMENTS视图[3,5](@ref)
28+ with self .engine .connect () as conn :
29+ sql = text ("""
30+ SELECT COMMENTS
31+ FROM ALL_TAB_COMMENTS
32+ WHERE OWNER = :owner
33+ AND TABLE_NAME = :table_name
34+ """ )
35+ return conn .execute (sql , {
36+ 'owner' : self .schema_name ,
37+ 'table_name' : table_name
38+ }).scalar () or ""
39+
40+ def get_column_comment (self , inspector : reflection .Inspector ,
41+ table_name : str , column_name : str ) -> str :
42+ # 查询ALL_COL_COMMENTS视图[3](@ref)
43+ with self .engine .connect () as conn :
44+ sql = text ("""
45+ SELECT COMMENTS
46+ FROM ALL_COL_COMMENTS
47+ WHERE OWNER = :owner
48+ AND TABLE_NAME = :table_name
49+ AND COLUMN_NAME = :column_name
50+ """ )
51+ return conn .execute (sql , {
52+ 'owner' : self .schema_name ,
53+ 'table_name' : table_name ,
54+ 'column_name' : column_name
55+ }).scalar () or ""
56+
57+ def normalize_type (self , raw_type : str ) -> str :
58+ # 标准化Oracle类型(如去除精度信息)[3,5](@ref)
59+ return raw_type .split ('(' )[0 ].split ('%' )[0 ].upper ()
0 commit comments