Skip to content

Instantly share code, notes, and snippets.

@ar45
Created February 13, 2017 01:59
Show Gist options
  • Save ar45/25258af16565409ea569d492ee49690b to your computer and use it in GitHub Desktop.
Save ar45/25258af16565409ea569d492ee49690b to your computer and use it in GitHub Desktop.
Quick dirty SQLAlchemy middleware for django < 1.9
from sqlalchemy.engine import create_engine
from sqlalchemy.orm import sessionmaker
def create_engine_from_django_settings(database='default'):
from django.conf import settings
db = settings.DATABASES[database]
engine = db['ENGINE'].lower()
if 'postgresql' in engine:
sqlalchemy_engine = 'postgresql'
elif 'mysql' in engine:
sqlalchemy_engine = 'mysql'
elif 'sqlite' in engine:
sqlalchemy_engine = 'sqlite'
else:
raise NotImplementedError('Could not determine what engine to use for "%s" automatically' % engine)
host = db['HOST']
name = db['NAME']
credentials = [db['USER']]
password = db['PASSWORD']
if password:
credentials.append(password)
user_password = ':'.join(credentials)
dsn = '{engine}://{user_password}@{host}/{name}'.format(
engine=sqlalchemy_engine, user_password=user_password, host=host, name=name)
return create_engine(dsn)
class SqlAlchemySession:
def __init__(self):
self.Session = sessionmaker(bind=create_engine_from_django_settings('postgres_db'))
def process_request(self, request):
request.db_session = self.Session()
return None
def process_exception(self, request, exception):
setattr(request, '_exception', exception)
try:
session = request.db_session
except AttributeError:
pass
else:
session.rollback()
return None
def process_response(self, request, response):
try:
session = request.db_session
except AttributeError:
return response
else:
if getattr(request, '_exception', None) is None:
session.commit()
session.close()
return response
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment