introspection.py 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414
  1. from collections import namedtuple
  2. from django.db import models
  3. from django.db.backends.base.introspection import BaseDatabaseIntrospection
  4. from django.db.backends.base.introspection import FieldInfo as BaseFieldInfo
  5. from django.db.backends.base.introspection import TableInfo as BaseTableInfo
  6. from django.db.backends.oracle.oracledb_any import oracledb
  7. FieldInfo = namedtuple(
  8. "FieldInfo", BaseFieldInfo._fields + ("is_autofield", "is_json", "comment")
  9. )
  10. TableInfo = namedtuple("TableInfo", BaseTableInfo._fields + ("comment",))
  11. class DatabaseIntrospection(BaseDatabaseIntrospection):
  12. cache_bust_counter = 1
  13. # Maps type objects to Django Field types.
  14. data_types_reverse = {
  15. oracledb.DB_TYPE_DATE: "DateField",
  16. oracledb.DB_TYPE_BINARY_DOUBLE: "FloatField",
  17. oracledb.DB_TYPE_BLOB: "BinaryField",
  18. oracledb.DB_TYPE_CHAR: "CharField",
  19. oracledb.DB_TYPE_CLOB: "TextField",
  20. oracledb.DB_TYPE_INTERVAL_DS: "DurationField",
  21. oracledb.DB_TYPE_NCHAR: "CharField",
  22. oracledb.DB_TYPE_NCLOB: "TextField",
  23. oracledb.DB_TYPE_NVARCHAR: "CharField",
  24. oracledb.DB_TYPE_NUMBER: "DecimalField",
  25. oracledb.DB_TYPE_TIMESTAMP: "DateTimeField",
  26. oracledb.DB_TYPE_VARCHAR: "CharField",
  27. }
  28. def get_field_type(self, data_type, description):
  29. if data_type == oracledb.NUMBER:
  30. precision, scale = description[4:6]
  31. if scale == 0:
  32. if precision > 11:
  33. return (
  34. "BigAutoField"
  35. if description.is_autofield
  36. else "BigIntegerField"
  37. )
  38. elif 1 < precision < 6 and description.is_autofield:
  39. return "SmallAutoField"
  40. elif precision == 1:
  41. return "BooleanField"
  42. elif description.is_autofield:
  43. return "AutoField"
  44. else:
  45. return "IntegerField"
  46. elif scale == -127:
  47. return "FloatField"
  48. elif data_type == oracledb.NCLOB and description.is_json:
  49. return "JSONField"
  50. return super().get_field_type(data_type, description)
  51. def get_table_list(self, cursor):
  52. """Return a list of table and view names in the current database."""
  53. cursor.execute(
  54. """
  55. SELECT
  56. user_tables.table_name,
  57. 't',
  58. user_tab_comments.comments
  59. FROM user_tables
  60. LEFT OUTER JOIN
  61. user_tab_comments
  62. ON user_tab_comments.table_name = user_tables.table_name
  63. WHERE
  64. NOT EXISTS (
  65. SELECT 1
  66. FROM user_mviews
  67. WHERE user_mviews.mview_name = user_tables.table_name
  68. )
  69. UNION ALL
  70. SELECT view_name, 'v', NULL FROM user_views
  71. UNION ALL
  72. SELECT mview_name, 'v', NULL FROM user_mviews
  73. """
  74. )
  75. return [
  76. TableInfo(self.identifier_converter(row[0]), row[1], row[2])
  77. for row in cursor.fetchall()
  78. ]
  79. def get_table_description(self, cursor, table_name):
  80. """
  81. Return a description of the table with the DB-API cursor.description
  82. interface.
  83. """
  84. # A default collation for the given table/view/materialized view.
  85. cursor.execute(
  86. """
  87. SELECT user_tables.default_collation
  88. FROM user_tables
  89. WHERE
  90. user_tables.table_name = UPPER(%s) AND
  91. NOT EXISTS (
  92. SELECT 1
  93. FROM user_mviews
  94. WHERE user_mviews.mview_name = user_tables.table_name
  95. )
  96. UNION ALL
  97. SELECT user_views.default_collation
  98. FROM user_views
  99. WHERE user_views.view_name = UPPER(%s)
  100. UNION ALL
  101. SELECT user_mviews.default_collation
  102. FROM user_mviews
  103. WHERE user_mviews.mview_name = UPPER(%s)
  104. """,
  105. [table_name, table_name, table_name],
  106. )
  107. row = cursor.fetchone()
  108. default_table_collation = row[0] if row else ""
  109. # user_tab_columns gives data default for columns
  110. cursor.execute(
  111. """
  112. SELECT
  113. user_tab_cols.column_name,
  114. user_tab_cols.data_default,
  115. CASE
  116. WHEN user_tab_cols.collation = %s
  117. THEN NULL
  118. ELSE user_tab_cols.collation
  119. END collation,
  120. CASE
  121. WHEN user_tab_cols.char_used IS NULL
  122. THEN user_tab_cols.data_length
  123. ELSE user_tab_cols.char_length
  124. END as display_size,
  125. CASE
  126. WHEN user_tab_cols.identity_column = 'YES' THEN 1
  127. ELSE 0
  128. END as is_autofield,
  129. CASE
  130. WHEN EXISTS (
  131. SELECT 1
  132. FROM user_json_columns
  133. WHERE
  134. user_json_columns.table_name = user_tab_cols.table_name AND
  135. user_json_columns.column_name = user_tab_cols.column_name
  136. )
  137. THEN 1
  138. ELSE 0
  139. END as is_json,
  140. user_col_comments.comments as col_comment
  141. FROM user_tab_cols
  142. LEFT OUTER JOIN
  143. user_col_comments ON
  144. user_col_comments.column_name = user_tab_cols.column_name AND
  145. user_col_comments.table_name = user_tab_cols.table_name
  146. WHERE user_tab_cols.table_name = UPPER(%s)
  147. """,
  148. [default_table_collation, table_name],
  149. )
  150. field_map = {
  151. column: (
  152. display_size,
  153. default.rstrip() if default and default != "NULL" else None,
  154. collation,
  155. is_autofield,
  156. is_json,
  157. comment,
  158. )
  159. for (
  160. column,
  161. default,
  162. collation,
  163. display_size,
  164. is_autofield,
  165. is_json,
  166. comment,
  167. ) in cursor.fetchall()
  168. }
  169. self.cache_bust_counter += 1
  170. cursor.execute(
  171. "SELECT * FROM {} WHERE ROWNUM < 2 AND {} > 0".format(
  172. self.connection.ops.quote_name(table_name), self.cache_bust_counter
  173. )
  174. )
  175. description = []
  176. for desc in cursor.description:
  177. name = desc[0]
  178. (
  179. display_size,
  180. default,
  181. collation,
  182. is_autofield,
  183. is_json,
  184. comment,
  185. ) = field_map[name]
  186. name %= {} # oracledb, for some reason, doubles percent signs.
  187. description.append(
  188. FieldInfo(
  189. self.identifier_converter(name),
  190. desc[1],
  191. display_size,
  192. desc[3],
  193. desc[4] or 0,
  194. desc[5] or 0,
  195. *desc[6:],
  196. default,
  197. collation,
  198. is_autofield,
  199. is_json,
  200. comment,
  201. )
  202. )
  203. return description
  204. def identifier_converter(self, name):
  205. """Identifier comparison is case insensitive under Oracle."""
  206. return name.lower()
  207. def get_sequences(self, cursor, table_name, table_fields=()):
  208. cursor.execute(
  209. """
  210. SELECT
  211. user_tab_identity_cols.sequence_name,
  212. user_tab_identity_cols.column_name
  213. FROM
  214. user_tab_identity_cols,
  215. user_constraints,
  216. user_cons_columns cols
  217. WHERE
  218. user_constraints.constraint_name = cols.constraint_name
  219. AND user_constraints.table_name = user_tab_identity_cols.table_name
  220. AND cols.column_name = user_tab_identity_cols.column_name
  221. AND user_constraints.constraint_type = 'P'
  222. AND user_tab_identity_cols.table_name = UPPER(%s)
  223. """,
  224. [table_name],
  225. )
  226. # Oracle allows only one identity column per table.
  227. row = cursor.fetchone()
  228. if row:
  229. return [
  230. {
  231. "name": self.identifier_converter(row[0]),
  232. "table": self.identifier_converter(table_name),
  233. "column": self.identifier_converter(row[1]),
  234. }
  235. ]
  236. # To keep backward compatibility for AutoFields that aren't Oracle
  237. # identity columns.
  238. for f in table_fields:
  239. if isinstance(f, models.AutoField):
  240. return [{"table": table_name, "column": f.column}]
  241. return []
  242. def get_relations(self, cursor, table_name):
  243. """
  244. Return a dictionary of {field_name: (field_name_other_table, other_table)}
  245. representing all foreign keys in the given table.
  246. """
  247. table_name = table_name.upper()
  248. cursor.execute(
  249. """
  250. SELECT ca.column_name, cb.table_name, cb.column_name
  251. FROM user_constraints, USER_CONS_COLUMNS ca, USER_CONS_COLUMNS cb
  252. WHERE user_constraints.table_name = %s AND
  253. user_constraints.constraint_name = ca.constraint_name AND
  254. user_constraints.r_constraint_name = cb.constraint_name AND
  255. ca.position = cb.position""",
  256. [table_name],
  257. )
  258. return {
  259. self.identifier_converter(field_name): (
  260. self.identifier_converter(rel_field_name),
  261. self.identifier_converter(rel_table_name),
  262. )
  263. for field_name, rel_table_name, rel_field_name in cursor.fetchall()
  264. }
  265. def get_primary_key_columns(self, cursor, table_name):
  266. cursor.execute(
  267. """
  268. SELECT
  269. cols.column_name
  270. FROM
  271. user_constraints,
  272. user_cons_columns cols
  273. WHERE
  274. user_constraints.constraint_name = cols.constraint_name AND
  275. user_constraints.constraint_type = 'P' AND
  276. user_constraints.table_name = UPPER(%s)
  277. ORDER BY
  278. cols.position
  279. """,
  280. [table_name],
  281. )
  282. return [self.identifier_converter(row[0]) for row in cursor.fetchall()]
  283. def get_constraints(self, cursor, table_name):
  284. """
  285. Retrieve any constraints or keys (unique, pk, fk, check, index) across
  286. one or more columns.
  287. """
  288. constraints = {}
  289. # Loop over the constraints, getting PKs, uniques, and checks
  290. cursor.execute(
  291. """
  292. SELECT
  293. user_constraints.constraint_name,
  294. LISTAGG(LOWER(cols.column_name), ',')
  295. WITHIN GROUP (ORDER BY cols.position),
  296. CASE user_constraints.constraint_type
  297. WHEN 'P' THEN 1
  298. ELSE 0
  299. END AS is_primary_key,
  300. CASE
  301. WHEN user_constraints.constraint_type IN ('P', 'U') THEN 1
  302. ELSE 0
  303. END AS is_unique,
  304. CASE user_constraints.constraint_type
  305. WHEN 'C' THEN 1
  306. ELSE 0
  307. END AS is_check_constraint
  308. FROM
  309. user_constraints
  310. LEFT OUTER JOIN
  311. user_cons_columns cols
  312. ON user_constraints.constraint_name = cols.constraint_name
  313. WHERE
  314. user_constraints.constraint_type = ANY('P', 'U', 'C')
  315. AND user_constraints.table_name = UPPER(%s)
  316. GROUP BY user_constraints.constraint_name, user_constraints.constraint_type
  317. """,
  318. [table_name],
  319. )
  320. for constraint, columns, pk, unique, check in cursor.fetchall():
  321. constraint = self.identifier_converter(constraint)
  322. constraints[constraint] = {
  323. "columns": columns.split(","),
  324. "primary_key": pk,
  325. "unique": unique,
  326. "foreign_key": None,
  327. "check": check,
  328. "index": unique, # All uniques come with an index
  329. }
  330. # Foreign key constraints
  331. cursor.execute(
  332. """
  333. SELECT
  334. cons.constraint_name,
  335. LISTAGG(LOWER(cols.column_name), ',')
  336. WITHIN GROUP (ORDER BY cols.position),
  337. LOWER(rcols.table_name),
  338. LOWER(rcols.column_name)
  339. FROM
  340. user_constraints cons
  341. INNER JOIN
  342. user_cons_columns rcols
  343. ON rcols.constraint_name = cons.r_constraint_name AND rcols.position = 1
  344. LEFT OUTER JOIN
  345. user_cons_columns cols
  346. ON cons.constraint_name = cols.constraint_name
  347. WHERE
  348. cons.constraint_type = 'R' AND
  349. cons.table_name = UPPER(%s)
  350. GROUP BY cons.constraint_name, rcols.table_name, rcols.column_name
  351. """,
  352. [table_name],
  353. )
  354. for constraint, columns, other_table, other_column in cursor.fetchall():
  355. constraint = self.identifier_converter(constraint)
  356. constraints[constraint] = {
  357. "primary_key": False,
  358. "unique": False,
  359. "foreign_key": (other_table, other_column),
  360. "check": False,
  361. "index": False,
  362. "columns": columns.split(","),
  363. }
  364. # Now get indexes
  365. cursor.execute(
  366. """
  367. SELECT
  368. ind.index_name,
  369. LOWER(ind.index_type),
  370. LOWER(ind.uniqueness),
  371. LISTAGG(LOWER(cols.column_name), ',')
  372. WITHIN GROUP (ORDER BY cols.column_position),
  373. LISTAGG(cols.descend, ',') WITHIN GROUP (ORDER BY cols.column_position)
  374. FROM
  375. user_ind_columns cols, user_indexes ind
  376. WHERE
  377. cols.table_name = UPPER(%s) AND
  378. NOT EXISTS (
  379. SELECT 1
  380. FROM user_constraints cons
  381. WHERE ind.index_name = cons.index_name
  382. ) AND cols.index_name = ind.index_name
  383. GROUP BY ind.index_name, ind.index_type, ind.uniqueness
  384. """,
  385. [table_name],
  386. )
  387. for constraint, type_, unique, columns, orders in cursor.fetchall():
  388. constraint = self.identifier_converter(constraint)
  389. constraints[constraint] = {
  390. "primary_key": False,
  391. "unique": unique == "unique",
  392. "foreign_key": None,
  393. "check": False,
  394. "index": True,
  395. "type": "idx" if type_ == "normal" else type_,
  396. "columns": columns.split(","),
  397. "orders": orders.split(","),
  398. }
  399. return constraints