/usr/lib/python2.7/dist-packages/storm/django/backend/base.py is in python-storm 0.19-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | __metaclass__ = type
__all__ = [
'DatabaseWrapper', 'DatabaseError', 'IntegrityError',
]
from django.conf import settings
import transaction
from storm.django.stores import get_store, get_store_uri
from storm.exceptions import DatabaseError, IntegrityError
class StormDatabaseWrapperMixin(object):
_store = None
def _get_connection(self):
if self._store is None:
self._store = get_store(settings.DATABASE_NAME)
# Make sure that the store is registered with the transaction
# manager: we don't know what the connection will be used for.
self._store._event.emit("register-transaction")
self._store._connection._ensure_connected()
return self._store._connection._raw_connection
def _set_connection(self, connection):
# Ignore attempts to set the connection.
pass
connection = property(_get_connection, _set_connection)
def _valid_connection(self):
# Storm handles the connection liveness checks.
return True
def _cursor(self, *args):
cursor = super(StormDatabaseWrapperMixin, self)._cursor(*args)
return StormCursorWrapper(self._store, cursor)
def _commit(self):
#print "commit"
transaction.commit()
def _rollback(self):
#print "rollback"
transaction.abort()
def close(self):
# As we are borrowing Storm's connection, we shouldn't close
# it behind Storm's back.
self._store = None
class StormCursorWrapper(object):
"""A cursor wrapper that checks for disconnection errors."""
def __init__(self, store, cursor):
self._connection = store._connection
self._cursor = cursor
def _check_disconnect(self, *args, **kwargs):
from django.db import DatabaseError as DjangoDatabaseError
kwargs['extra_disconnection_errors'] = DjangoDatabaseError
return self._connection._check_disconnect(*args, **kwargs)
def execute(self, statement, *args):
"""Execute an SQL statement."""
return self._check_disconnect(self._cursor.execute, statement, *args)
def fetchone(self):
"""Fetch one row from the result."""
return self._check_disconnect(self._cursor.fetchone)
def fetchall(self):
"""Fetch all rows from the result."""
return self._check_disconnect(self._cursor.fetchall)
def fetchmany(self, *args):
"""Fetch multiple rows from the result."""
return self._check_disconnect(self._cursor.fetchmany, *args)
@property
def description(self):
"""Fetch the description of the result."""
return self._check_disconnect(getattr, self._cursor, "description")
@property
def rowcount(self):
"""Fetch the number of rows in the result."""
return self._check_disconnect(getattr, self._cursor, "rowcount")
@property
def query(self):
"""Fetch the last executed query."""
return self._check_disconnect(getattr, self._cursor, "query")
PostgresStormDatabaseWrapper = None
MySQLStormDatabaseWrapper = None
def DatabaseWrapper(*args, **kwargs):
store_uri = get_store_uri(settings.DATABASE_NAME)
# Create a DatabaseWrapper class that uses an underlying Storm
# connection. We don't support sqlite here because Django expects
# a bunch of special setup on the connection that Storm doesn't
# do.
if store_uri.startswith('postgres:'):
global PostgresStormDatabaseWrapper
if PostgresStormDatabaseWrapper is None:
from django.db.backends.postgresql_psycopg2.base import (
DatabaseWrapper as PostgresDatabaseWrapper)
class PostgresStormDatabaseWrapper(StormDatabaseWrapperMixin,
PostgresDatabaseWrapper):
pass
DatabaseWrapper = PostgresStormDatabaseWrapper
elif store_uri.startswith('mysql:'):
global MySQLStormDatabaseWrapper
if MySQLStormDatabaseWrapper is None:
from django.db.backends.mysql.base import (
DatabaseWrapper as MySQLDatabaseWrapper)
class MySQLStormDatabaseWrapper(StormDatabaseWrapperMixin,
MySQLDatabaseWrapper):
pass
DatabaseWrapper = MySQLStormDatabaseWrapper
else:
assert False, (
"Unsupported database backend: %s" % store_uri)
return DatabaseWrapper(*args, **kwargs)
|