py_tasks_melange: comparison app/django/db/backends/oracle/base.py

equal deleted inserted replaced

-:6641e941ef1e
+:ff1a9aa48cfd
 Requires cx_Oracle: http://www.python.net/crew/atuining/cx_Oracle/
 """
 import os
+import datetime
-from django.db.backends import BaseDatabaseWrapper, BaseDatabaseFeatures, BaseDatabaseOperations, util
+import time
-from django.db.backends.oracle import query
-from django.utils.datastructures import SortedDict
-from django.utils.encoding import smart_str, force_unicode
 # Oracle takes client-side character set encoding from the environment.
 os.environ['NLS_LANG'] = '.UTF8'
 try:
 import cx_Oracle as Database
 except ImportError, e:
 from django.core.exceptions import ImproperlyConfigured
 raise ImproperlyConfigured("Error loading cx_Oracle module: %s" % e)
-DatabaseError = Database.Error
+from django.db.backends import *
+from django.db.backends.oracle import query
+from django.db.backends.oracle.client import DatabaseClient
+from django.db.backends.oracle.creation import DatabaseCreation
+from django.db.backends.oracle.introspection import DatabaseIntrospection
+from django.utils.encoding import smart_str, force_unicode
+DatabaseError = Database.DatabaseError
 IntegrityError = Database.IntegrityError
 class DatabaseFeatures(BaseDatabaseFeatures):
-allows_group_by_ordinal = False
-allows_unique_and_pk = False        # Suppress UNIQUE/PK for Oracle (ORA-02259)
 empty_fetchmany_value = ()
 needs_datetime_string_cast = False
-needs_upper_for_iops = True
-supports_tablespaces = True
-uses_case_insensitive_names = True
 uses_custom_query_class = True
+interprets_empty_strings_as_nulls = True
 class DatabaseOperations(BaseDatabaseOperations):
 def autoinc_sql(self, table, column):
 # To simulate auto-incrementing primary keys in Oracle, we have to
 # create a sequence and a trigger.
 sq_name = get_sequence_name(table)
 tr_name = get_trigger_name(table)
 tbl_name = self.quote_name(table)
 col_name = self.quote_name(column)
-sequence_sql = 'CREATE SEQUENCE %s;' % sq_name
+sequence_sql = """
+DECLARE
+i INTEGER;
+BEGIN
+SELECT COUNT(*) INTO i FROM USER_CATALOG
+WHERE TABLE_NAME = '%(sq_name)s' AND TABLE_TYPE = 'SEQUENCE';
+IF i = 0 THEN
+EXECUTE IMMEDIATE 'CREATE SEQUENCE %(sq_name)s';
+END IF;
+END;
+/""" % locals()
 trigger_sql = """
 CREATE OR REPLACE TRIGGER %(tr_name)s
 BEFORE INSERT ON %(tbl_name)s
 FOR EACH ROW
 WHEN (new.%(col_name)s IS NULL)
 def last_insert_id(self, cursor, table_name, pk_name):
 sq_name = util.truncate_name(table_name, self.max_name_length() - 3)
 cursor.execute('SELECT %s_sq.currval FROM dual' % sq_name)
 return cursor.fetchone()[0]
-def limit_offset_sql(self, limit, offset=None):
-# Limits and offset are too complicated to be handled here.
-# Instead, they are handled in django/db/backends/oracle/query.py.
-return ""
 def lookup_cast(self, lookup_type):
 if lookup_type in ('iexact', 'icontains', 'istartswith', 'iendswith'):
 return "UPPER(%s)"
 return "%s"
 def max_name_length(self):
 return 30
+def prep_for_iexact_query(self, x):
+return x
 def query_class(self, DefaultQueryClass):
 return query.query_class(DefaultQueryClass, Database)
 def quote_name(self, name):
 style.SQL_FIELD(self.quote_name(table))
 ) for table in tables]
 # Since we've just deleted all the rows, running our sequence
 # ALTER code will reset the sequence to 0.
 for sequence_info in sequences:
-table_name = sequence_info['table']
+sequence_name = get_sequence_name(sequence_info['table'])
-seq_name = get_sequence_name(table_name)
+table_name = self.quote_name(sequence_info['table'])
 column_name = self.quote_name(sequence_info['column'] or 'id')
-query = _get_sequence_reset_sql() % {'sequence': seq_name,
+query = _get_sequence_reset_sql() % {'sequence': sequence_name,
-'table': self.quote_name(table_name),
+'table': table_name,
 'column': column_name}
 sql.append(query)
 return sql
 else:
 return []
 def sequence_reset_sql(self, style, model_list):
 from django.db import models
 output = []
 query = _get_sequence_reset_sql()
 for model in model_list:
-for f in model._meta.fields:
+for f in model._meta.local_fields:
 if isinstance(f, models.AutoField):
+table_name = self.quote_name(model._meta.db_table)
 sequence_name = get_sequence_name(model._meta.db_table)
-column_name = self.quote_name(f.db_column or f.name)
+column_name = self.quote_name(f.column)
 output.append(query % {'sequence': sequence_name,
-'table': model._meta.db_table,
+'table': table_name,
 'column': column_name})
 break # Only one AutoField is allowed per model, so don't bother continuing.
 for f in model._meta.many_to_many:
+table_name = self.quote_name(f.m2m_db_table())
 sequence_name = get_sequence_name(f.m2m_db_table())
+column_name = self.quote_name('id')
 output.append(query % {'sequence': sequence_name,
-'table': f.m2m_db_table(),
+'table': table_name,
-'column': self.quote_name('id')})
+'column': column_name})
 return output
 def start_transaction_sql(self):
 return ''
 def tablespace_sql(self, tablespace, inline=False):
 return "%sTABLESPACE %s" % ((inline and "USING INDEX " or ""), self.quote_name(tablespace))
+def value_to_db_time(self, value):
+if value is None:
+return None
+if isinstance(value, basestring):
+return datetime.datetime(*(time.strptime(value, '%H:%M:%S')[:6]))
+return datetime.datetime(1900, 1, 1, value.hour, value.minute,
+value.second, value.microsecond)
+def year_lookup_bounds_for_date_field(self, value):
+first = '%s-01-01'
+second = '%s-12-31'
+return [first % value, second % value]
 class DatabaseWrapper(BaseDatabaseWrapper):
-features = DatabaseFeatures()
-ops = DatabaseOperations()
 operators = {
 'exact': '= %s',
 'iexact': '= UPPER(%s)',
 'contains': "LIKEC %s ESCAPE '\\'",
 'icontains': "LIKEC UPPER(%s) ESCAPE '\\'",
 'endswith': "LIKEC %s ESCAPE '\\'",
 'istartswith': "LIKEC UPPER(%s) ESCAPE '\\'",
 'iendswith': "LIKEC UPPER(%s) ESCAPE '\\'",
 }
 oracle_version = None
+def __init__(self, *args, **kwargs):
+super(DatabaseWrapper, self).__init__(*args, **kwargs)
+self.features = DatabaseFeatures()
+self.ops = DatabaseOperations()
+self.client = DatabaseClient()
+self.creation = DatabaseCreation(self)
+self.introspection = DatabaseIntrospection(self)
+self.validation = BaseDatabaseValidation()
 def _valid_connection(self):
 return self.connection is not None
 def _cursor(self, settings):
 cursor = FormatStylePlaceholderCursor(self.connection)
 # Default arraysize of 1 is highly sub-optimal.
 cursor.arraysize = 100
 return cursor
+class OracleParam(object):
+"""
+Wrapper object for formatting parameters for Oracle. If the string
+representation of the value is large enough (greater than 4000 characters)
+the input size needs to be set as NCLOB. Alternatively, if the parameter has
+an `input_size` attribute, then the value of the `input_size` attribute will
+be used instead. Otherwise, no input size will be set for the parameter when
+executing the query.
+"""
+def __init__(self, param, charset, strings_only=False):
+self.smart_str = smart_str(param, charset, strings_only)
+if hasattr(param, 'input_size'):
+# If parameter has `input_size` attribute, use that.
+self.input_size = param.input_size
+elif isinstance(param, basestring) and len(param) > 4000:
+# Mark any string parameter greater than 4000 characters as an NCLOB.
+self.input_size = Database.NCLOB
+else:
+self.input_size = None
 class FormatStylePlaceholderCursor(Database.Cursor):
 """
 Django uses "format" (e.g. '%s') style placeholders, but Oracle uses ":var"
 style. This fixes it -- but note that if you want to use a literal "%s" in
 a query, you'll need to use "%%s".
 charset = 'utf-8'
 def _format_params(self, params):
 if isinstance(params, dict):
 result = {}
-charset = self.charset
 for key, value in params.items():
-result[smart_str(key, charset)] = smart_str(value, charset)
+result[smart_str(key, self.charset)] = OracleParam(param, self.charset)
 return result
 else:
-return tuple([smart_str(p, self.charset, True) for p in params])
+return tuple([OracleParam(p, self.charset, True) for p in params])
 def _guess_input_sizes(self, params_list):
-# Mark any string parameter greater than 4000 characters as an NCLOB.
 if isinstance(params_list[0], dict):
 sizes = {}
 iterators = [params.iteritems() for params in params_list]
 else:
 sizes = [None] * len(params_list[0])
 iterators = [enumerate(params) for params in params_list]
 for iterator in iterators:
 for key, value in iterator:
-if isinstance(value, basestring) and len(value) > 4000:
+if value.input_size: sizes[key] = value.input_size
-sizes[key] = Database.NCLOB
 if isinstance(sizes, dict):
 self.setinputsizes(**sizes)
 else:
 self.setinputsizes(*sizes)
+def _param_generator(self, params):
+if isinstance(params, dict):
+return dict([(k, p.smart_str) for k, p in params.iteritems()])
+else:
+return [p.smart_str for p in params]
 def execute(self, query, params=None):
 if params is None:
 params = []
 else:
 # is being passed to SQL*Plus.
 if query.endswith(';') or query.endswith('/'):
 query = query[:-1]
 query = smart_str(query, self.charset) % tuple(args)
 self._guess_input_sizes([params])
-return Database.Cursor.execute(self, query, params)
+try:
+return Database.Cursor.execute(self, query, self._param_generator(params))
+except DatabaseError, e:
+# cx_Oracle <= 4.4.0 wrongly raises a DatabaseError for ORA-01400.
+if e.args[0].code == 1400 and not isinstance(e, IntegrityError):
+e = IntegrityError(e.args[0])
+raise e
 def executemany(self, query, params=None):
 try:
 args = [(':arg%d' % i) for i in range(len(params[0]))]
 except (IndexError, TypeError):
 # characters must be included in the original query in case the query
 # is being passed to SQL*Plus.
 if query.endswith(';') or query.endswith('/'):
 query = query[:-1]
 query = smart_str(query, self.charset) % tuple(args)
-new_param_list = [self._format_params(i) for i in params]
+formatted = [self._format_params(i) for i in params]
-self._guess_input_sizes(new_param_list)
+self._guess_input_sizes(formatted)
-return Database.Cursor.executemany(self, query, new_param_list)
+try:
+return Database.Cursor.executemany(self, query, [self._param_generator(p) for p in formatted])
+except DatabaseError, e:
+# cx_Oracle <= 4.4.0 wrongly raises a DatabaseError for ORA-01400.
+if e.args[0].code == 1400 and not isinstance(e, IntegrityError):
+e = IntegrityError(e.args[0])
+raise e
 def fetchone(self):
 row = Database.Cursor.fetchone(self)
 if row is None:
 return row

changeset 323	ff1a9aa48cfd
parent 54	03e267d67478