| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414 |
- from collections import namedtuple
- from django.db import models
- from django.db.backends.base.introspection import BaseDatabaseIntrospection
- from django.db.backends.base.introspection import FieldInfo as BaseFieldInfo
- from django.db.backends.base.introspection import TableInfo as BaseTableInfo
- from django.db.backends.oracle.oracledb_any import oracledb
- FieldInfo = namedtuple(
- "FieldInfo", BaseFieldInfo._fields + ("is_autofield", "is_json", "comment")
- )
- TableInfo = namedtuple("TableInfo", BaseTableInfo._fields + ("comment",))
- class DatabaseIntrospection(BaseDatabaseIntrospection):
- cache_bust_counter = 1
- # Maps type objects to Django Field types.
- data_types_reverse = {
- oracledb.DB_TYPE_DATE: "DateField",
- oracledb.DB_TYPE_BINARY_DOUBLE: "FloatField",
- oracledb.DB_TYPE_BLOB: "BinaryField",
- oracledb.DB_TYPE_CHAR: "CharField",
- oracledb.DB_TYPE_CLOB: "TextField",
- oracledb.DB_TYPE_INTERVAL_DS: "DurationField",
- oracledb.DB_TYPE_NCHAR: "CharField",
- oracledb.DB_TYPE_NCLOB: "TextField",
- oracledb.DB_TYPE_NVARCHAR: "CharField",
- oracledb.DB_TYPE_NUMBER: "DecimalField",
- oracledb.DB_TYPE_TIMESTAMP: "DateTimeField",
- oracledb.DB_TYPE_VARCHAR: "CharField",
- }
- def get_field_type(self, data_type, description):
- if data_type == oracledb.NUMBER:
- precision, scale = description[4:6]
- if scale == 0:
- if precision > 11:
- return (
- "BigAutoField"
- if description.is_autofield
- else "BigIntegerField"
- )
- elif 1 < precision < 6 and description.is_autofield:
- return "SmallAutoField"
- elif precision == 1:
- return "BooleanField"
- elif description.is_autofield:
- return "AutoField"
- else:
- return "IntegerField"
- elif scale == -127:
- return "FloatField"
- elif data_type == oracledb.NCLOB and description.is_json:
- return "JSONField"
- return super().get_field_type(data_type, description)
- def get_table_list(self, cursor):
- """Return a list of table and view names in the current database."""
- cursor.execute(
- """
- SELECT
- user_tables.table_name,
- 't',
- user_tab_comments.comments
- FROM user_tables
- LEFT OUTER JOIN
- user_tab_comments
- ON user_tab_comments.table_name = user_tables.table_name
- WHERE
- NOT EXISTS (
- SELECT 1
- FROM user_mviews
- WHERE user_mviews.mview_name = user_tables.table_name
- )
- UNION ALL
- SELECT view_name, 'v', NULL FROM user_views
- UNION ALL
- SELECT mview_name, 'v', NULL FROM user_mviews
- """
- )
- return [
- TableInfo(self.identifier_converter(row[0]), row[1], row[2])
- for row in cursor.fetchall()
- ]
- def get_table_description(self, cursor, table_name):
- """
- Return a description of the table with the DB-API cursor.description
- interface.
- """
- # A default collation for the given table/view/materialized view.
- cursor.execute(
- """
- SELECT user_tables.default_collation
- FROM user_tables
- WHERE
- user_tables.table_name = UPPER(%s) AND
- NOT EXISTS (
- SELECT 1
- FROM user_mviews
- WHERE user_mviews.mview_name = user_tables.table_name
- )
- UNION ALL
- SELECT user_views.default_collation
- FROM user_views
- WHERE user_views.view_name = UPPER(%s)
- UNION ALL
- SELECT user_mviews.default_collation
- FROM user_mviews
- WHERE user_mviews.mview_name = UPPER(%s)
- """,
- [table_name, table_name, table_name],
- )
- row = cursor.fetchone()
- default_table_collation = row[0] if row else ""
- # user_tab_columns gives data default for columns
- cursor.execute(
- """
- SELECT
- user_tab_cols.column_name,
- user_tab_cols.data_default,
- CASE
- WHEN user_tab_cols.collation = %s
- THEN NULL
- ELSE user_tab_cols.collation
- END collation,
- CASE
- WHEN user_tab_cols.char_used IS NULL
- THEN user_tab_cols.data_length
- ELSE user_tab_cols.char_length
- END as display_size,
- CASE
- WHEN user_tab_cols.identity_column = 'YES' THEN 1
- ELSE 0
- END as is_autofield,
- CASE
- WHEN EXISTS (
- SELECT 1
- FROM user_json_columns
- WHERE
- user_json_columns.table_name = user_tab_cols.table_name AND
- user_json_columns.column_name = user_tab_cols.column_name
- )
- THEN 1
- ELSE 0
- END as is_json,
- user_col_comments.comments as col_comment
- FROM user_tab_cols
- LEFT OUTER JOIN
- user_col_comments ON
- user_col_comments.column_name = user_tab_cols.column_name AND
- user_col_comments.table_name = user_tab_cols.table_name
- WHERE user_tab_cols.table_name = UPPER(%s)
- """,
- [default_table_collation, table_name],
- )
- field_map = {
- column: (
- display_size,
- default.rstrip() if default and default != "NULL" else None,
- collation,
- is_autofield,
- is_json,
- comment,
- )
- for (
- column,
- default,
- collation,
- display_size,
- is_autofield,
- is_json,
- comment,
- ) in cursor.fetchall()
- }
- self.cache_bust_counter += 1
- cursor.execute(
- "SELECT * FROM {} WHERE ROWNUM < 2 AND {} > 0".format(
- self.connection.ops.quote_name(table_name), self.cache_bust_counter
- )
- )
- description = []
- for desc in cursor.description:
- name = desc[0]
- (
- display_size,
- default,
- collation,
- is_autofield,
- is_json,
- comment,
- ) = field_map[name]
- name %= {} # oracledb, for some reason, doubles percent signs.
- description.append(
- FieldInfo(
- self.identifier_converter(name),
- desc[1],
- display_size,
- desc[3],
- desc[4] or 0,
- desc[5] or 0,
- *desc[6:],
- default,
- collation,
- is_autofield,
- is_json,
- comment,
- )
- )
- return description
- def identifier_converter(self, name):
- """Identifier comparison is case insensitive under Oracle."""
- return name.lower()
- def get_sequences(self, cursor, table_name, table_fields=()):
- cursor.execute(
- """
- SELECT
- user_tab_identity_cols.sequence_name,
- user_tab_identity_cols.column_name
- FROM
- user_tab_identity_cols,
- user_constraints,
- user_cons_columns cols
- WHERE
- user_constraints.constraint_name = cols.constraint_name
- AND user_constraints.table_name = user_tab_identity_cols.table_name
- AND cols.column_name = user_tab_identity_cols.column_name
- AND user_constraints.constraint_type = 'P'
- AND user_tab_identity_cols.table_name = UPPER(%s)
- """,
- [table_name],
- )
- # Oracle allows only one identity column per table.
- row = cursor.fetchone()
- if row:
- return [
- {
- "name": self.identifier_converter(row[0]),
- "table": self.identifier_converter(table_name),
- "column": self.identifier_converter(row[1]),
- }
- ]
- # To keep backward compatibility for AutoFields that aren't Oracle
- # identity columns.
- for f in table_fields:
- if isinstance(f, models.AutoField):
- return [{"table": table_name, "column": f.column}]
- return []
- def get_relations(self, cursor, table_name):
- """
- Return a dictionary of {field_name: (field_name_other_table, other_table)}
- representing all foreign keys in the given table.
- """
- table_name = table_name.upper()
- cursor.execute(
- """
- SELECT ca.column_name, cb.table_name, cb.column_name
- FROM user_constraints, USER_CONS_COLUMNS ca, USER_CONS_COLUMNS cb
- WHERE user_constraints.table_name = %s AND
- user_constraints.constraint_name = ca.constraint_name AND
- user_constraints.r_constraint_name = cb.constraint_name AND
- ca.position = cb.position""",
- [table_name],
- )
- return {
- self.identifier_converter(field_name): (
- self.identifier_converter(rel_field_name),
- self.identifier_converter(rel_table_name),
- )
- for field_name, rel_table_name, rel_field_name in cursor.fetchall()
- }
- def get_primary_key_columns(self, cursor, table_name):
- cursor.execute(
- """
- SELECT
- cols.column_name
- FROM
- user_constraints,
- user_cons_columns cols
- WHERE
- user_constraints.constraint_name = cols.constraint_name AND
- user_constraints.constraint_type = 'P' AND
- user_constraints.table_name = UPPER(%s)
- ORDER BY
- cols.position
- """,
- [table_name],
- )
- return [self.identifier_converter(row[0]) for row in cursor.fetchall()]
- def get_constraints(self, cursor, table_name):
- """
- Retrieve any constraints or keys (unique, pk, fk, check, index) across
- one or more columns.
- """
- constraints = {}
- # Loop over the constraints, getting PKs, uniques, and checks
- cursor.execute(
- """
- SELECT
- user_constraints.constraint_name,
- LISTAGG(LOWER(cols.column_name), ',')
- WITHIN GROUP (ORDER BY cols.position),
- CASE user_constraints.constraint_type
- WHEN 'P' THEN 1
- ELSE 0
- END AS is_primary_key,
- CASE
- WHEN user_constraints.constraint_type IN ('P', 'U') THEN 1
- ELSE 0
- END AS is_unique,
- CASE user_constraints.constraint_type
- WHEN 'C' THEN 1
- ELSE 0
- END AS is_check_constraint
- FROM
- user_constraints
- LEFT OUTER JOIN
- user_cons_columns cols
- ON user_constraints.constraint_name = cols.constraint_name
- WHERE
- user_constraints.constraint_type = ANY('P', 'U', 'C')
- AND user_constraints.table_name = UPPER(%s)
- GROUP BY user_constraints.constraint_name, user_constraints.constraint_type
- """,
- [table_name],
- )
- for constraint, columns, pk, unique, check in cursor.fetchall():
- constraint = self.identifier_converter(constraint)
- constraints[constraint] = {
- "columns": columns.split(","),
- "primary_key": pk,
- "unique": unique,
- "foreign_key": None,
- "check": check,
- "index": unique, # All uniques come with an index
- }
- # Foreign key constraints
- cursor.execute(
- """
- SELECT
- cons.constraint_name,
- LISTAGG(LOWER(cols.column_name), ',')
- WITHIN GROUP (ORDER BY cols.position),
- LOWER(rcols.table_name),
- LOWER(rcols.column_name)
- FROM
- user_constraints cons
- INNER JOIN
- user_cons_columns rcols
- ON rcols.constraint_name = cons.r_constraint_name AND rcols.position = 1
- LEFT OUTER JOIN
- user_cons_columns cols
- ON cons.constraint_name = cols.constraint_name
- WHERE
- cons.constraint_type = 'R' AND
- cons.table_name = UPPER(%s)
- GROUP BY cons.constraint_name, rcols.table_name, rcols.column_name
- """,
- [table_name],
- )
- for constraint, columns, other_table, other_column in cursor.fetchall():
- constraint = self.identifier_converter(constraint)
- constraints[constraint] = {
- "primary_key": False,
- "unique": False,
- "foreign_key": (other_table, other_column),
- "check": False,
- "index": False,
- "columns": columns.split(","),
- }
- # Now get indexes
- cursor.execute(
- """
- SELECT
- ind.index_name,
- LOWER(ind.index_type),
- LOWER(ind.uniqueness),
- LISTAGG(LOWER(cols.column_name), ',')
- WITHIN GROUP (ORDER BY cols.column_position),
- LISTAGG(cols.descend, ',') WITHIN GROUP (ORDER BY cols.column_position)
- FROM
- user_ind_columns cols, user_indexes ind
- WHERE
- cols.table_name = UPPER(%s) AND
- NOT EXISTS (
- SELECT 1
- FROM user_constraints cons
- WHERE ind.index_name = cons.index_name
- ) AND cols.index_name = ind.index_name
- GROUP BY ind.index_name, ind.index_type, ind.uniqueness
- """,
- [table_name],
- )
- for constraint, type_, unique, columns, orders in cursor.fetchall():
- constraint = self.identifier_converter(constraint)
- constraints[constraint] = {
- "primary_key": False,
- "unique": unique == "unique",
- "foreign_key": None,
- "check": False,
- "index": True,
- "type": "idx" if type_ == "normal" else type_,
- "columns": columns.split(","),
- "orders": orders.split(","),
- }
- return constraints
|