Skip to content

Instantly share code, notes, and snippets.

@EBNull
Created October 16, 2012 20:56
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save EBNull/3901984 to your computer and use it in GitHub Desktop.
Save EBNull/3901984 to your computer and use it in GitHub Desktop.
In django, create a database definition that only exists for the current session.
"""
Example:
import django.db
from cStringIO import StringIO
def inspect_mdb(filename):
db_dict = {'ENGINE': 'access.pyodbc', 'OPTIONS': {'driver': 'access'}, 'NAME': filename}
with temp_db(db_dict, 'test') as using:
django.db.connections[using].cursor() #Connect immediately
io = StringIO()
from django.core.management.commands.inspectdb import Command as InspectCommand
c = InspectCommand()
c.stdout = io
c.handle_noargs(database=using)
return io.getvalue()
if __name__ == '__main__':
print inspect_mdb(os.argv[1])
"""
import django.db
import contextlib
#WARNING: There is a race condition based on usingname among threads. You should wrap the using name in a threadlocal or equivalent.
@contextlib.contextmanager
def temp_db(db_dict, usingname='_temp_db'):
insert_db(db_dict, usingname)
try:
yield usingname
finally:
remove_db(usingname)
def insert_db(definition, using):
if using in django.db.connections.databases:
raise ValueError('%s already exists'%(using))
if using in django.db.connections:
raise ValueError('%s already exists'%(using))
django.db.connections.databases[using] = definition
return using
def remove_db(using):
del django.db.connections.databases[using]
conn = None
try:
conn = django.db.connections[using]
conn.close()
except Exception:
pass
try:
delattr(django.db.connections._connections, using)
except Exception:
pass
assert not hasattr(django.db.connections._connections, using), "Could not remove database wrapper %s"%(using)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment