153 lines
4.7 KiB
Python
153 lines
4.7 KiB
Python
from __future__ import unicode_literals
|
|
|
|
import datetime
|
|
import decimal
|
|
import hashlib
|
|
import logging
|
|
from time import time
|
|
|
|
from django.conf import settings
|
|
from django.utils.encoding import force_bytes
|
|
from django.utils.timezone import utc
|
|
|
|
|
|
logger = logging.getLogger('django.db.backends')
|
|
|
|
|
|
class CursorWrapper(object):
|
|
def __init__(self, cursor, db):
|
|
self.cursor = cursor
|
|
self.db = db
|
|
|
|
def set_dirty(self):
|
|
if self.db.is_managed():
|
|
self.db.set_dirty()
|
|
|
|
def __getattr__(self, attr):
|
|
if attr in ('execute', 'executemany', 'callproc'):
|
|
self.set_dirty()
|
|
return getattr(self.cursor, attr)
|
|
|
|
def __iter__(self):
|
|
return iter(self.cursor)
|
|
|
|
|
|
class CursorDebugWrapper(CursorWrapper):
|
|
|
|
def execute(self, sql, params=()):
|
|
self.set_dirty()
|
|
start = time()
|
|
try:
|
|
return self.cursor.execute(sql, params)
|
|
finally:
|
|
stop = time()
|
|
duration = stop - start
|
|
sql = self.db.ops.last_executed_query(self.cursor, sql, params)
|
|
self.db.queries.append({
|
|
'sql': sql,
|
|
'time': "%.3f" % duration,
|
|
})
|
|
logger.debug('(%.3f) %s; args=%s' % (duration, sql, params),
|
|
extra={'duration': duration, 'sql': sql, 'params': params}
|
|
)
|
|
|
|
def executemany(self, sql, param_list):
|
|
self.set_dirty()
|
|
start = time()
|
|
try:
|
|
return self.cursor.executemany(sql, param_list)
|
|
finally:
|
|
stop = time()
|
|
duration = stop - start
|
|
try:
|
|
times = len(param_list)
|
|
except TypeError: # param_list could be an iterator
|
|
times = '?'
|
|
self.db.queries.append({
|
|
'sql': '%s times: %s' % (times, sql),
|
|
'time': "%.3f" % duration,
|
|
})
|
|
logger.debug('(%.3f) %s; args=%s' % (duration, sql, param_list),
|
|
extra={'duration': duration, 'sql': sql, 'params': param_list}
|
|
)
|
|
|
|
|
|
###############################################
|
|
# Converters from database (string) to Python #
|
|
###############################################
|
|
|
|
def typecast_date(s):
|
|
return s and datetime.date(*map(int, s.split('-'))) or None # returns None if s is null
|
|
|
|
def typecast_time(s): # does NOT store time zone information
|
|
if not s: return None
|
|
hour, minutes, seconds = s.split(':')
|
|
if '.' in seconds: # check whether seconds have a fractional part
|
|
seconds, microseconds = seconds.split('.')
|
|
else:
|
|
microseconds = '0'
|
|
return datetime.time(int(hour), int(minutes), int(seconds), int(float('.'+microseconds) * 1000000))
|
|
|
|
def typecast_timestamp(s): # does NOT store time zone information
|
|
# "2005-07-29 15:48:00.590358-05"
|
|
# "2005-07-29 09:56:00-05"
|
|
if not s: return None
|
|
if not ' ' in s: return typecast_date(s)
|
|
d, t = s.split()
|
|
# Extract timezone information, if it exists. Currently we just throw
|
|
# it away, but in the future we may make use of it.
|
|
if '-' in t:
|
|
t, tz = t.split('-', 1)
|
|
tz = '-' + tz
|
|
elif '+' in t:
|
|
t, tz = t.split('+', 1)
|
|
tz = '+' + tz
|
|
else:
|
|
tz = ''
|
|
dates = d.split('-')
|
|
times = t.split(':')
|
|
seconds = times[2]
|
|
if '.' in seconds: # check whether seconds have a fractional part
|
|
seconds, microseconds = seconds.split('.')
|
|
else:
|
|
microseconds = '0'
|
|
tzinfo = utc if settings.USE_TZ else None
|
|
return datetime.datetime(int(dates[0]), int(dates[1]), int(dates[2]),
|
|
int(times[0]), int(times[1]), int(seconds),
|
|
int((microseconds + '000000')[:6]), tzinfo)
|
|
|
|
def typecast_decimal(s):
|
|
if s is None or s == '':
|
|
return None
|
|
return decimal.Decimal(s)
|
|
|
|
###############################################
|
|
# Converters from Python to database (string) #
|
|
###############################################
|
|
|
|
def rev_typecast_decimal(d):
|
|
if d is None:
|
|
return None
|
|
return str(d)
|
|
|
|
def truncate_name(name, length=None, hash_len=4):
|
|
"""Shortens a string to a repeatable mangled version with the given length.
|
|
"""
|
|
if length is None or len(name) <= length:
|
|
return name
|
|
|
|
hsh = hashlib.md5(force_bytes(name)).hexdigest()[:hash_len]
|
|
return '%s%s' % (name[:length-hash_len], hsh)
|
|
|
|
def format_number(value, max_digits, decimal_places):
|
|
"""
|
|
Formats a number into a string with the requisite number of digits and
|
|
decimal places.
|
|
"""
|
|
if isinstance(value, decimal.Decimal):
|
|
context = decimal.getcontext().copy()
|
|
context.prec = max_digits
|
|
return '%s' % str(value.quantize(decimal.Decimal(".1") ** decimal_places, context=context))
|
|
else:
|
|
return "%.*f" % (decimal_places, value)
|