Skip to content

Instantly share code, notes, and snippets.

@dpnova
Created August 25, 2014 02:54
Show Gist options
  • Save dpnova/7e3dd4f6de2b7748bec1 to your computer and use it in GitHub Desktop.
Save dpnova/7e3dd4f6de2b7748bec1 to your computer and use it in GitHub Desktop.
Create sqlalchemy engines and metadata from a Django-ish set of settings.
from sqlalchemy import create_engine, Table
from sqlalchemy.schema import MetaData
from somewhere import settings
from collections import defaultdict
_engines = {}
_metadata = defaultdict(lambda: MetaData())
_tables = defaultdict(dict)
def db_table(content_type_name, which="default"):
global _metadata
global _tables
global _engines
if content_type_name not in _tables[which]:
engine = get_engine(which)
table = Table(
content_type_name,
_metadata[which],
autoload=True,
autoload_with=engine
)
_tables[which][content_type_name] = table
return _tables[which][content_type_name]
def create_connect_string(which):
tmp = '%(engine)s://%(user)s:%(password)s@%(host)s%(port)s/%(database)s'
db = settings.DATABASES[which]
if "postgresql" in db['ENGINE']:
engine = "postgresql"
elif "sqlite3" in db['ENGINE']:
engine = "sqlite3"
data = {
"engine": engine,
"host": db['HOST'] or "localhost",
"user": db['USER'],
"password": db['PASSWORD'],
"port": ":" + db['PORT'] if db['PORT'] else "",
"database": db['NAME']
}
return tmp % data
def get_engine(which="default"):
global _engines
global _metadatas
if not _engines:
connect_string = create_connect_string(which)
_engines[which] = create_engine(connect_string)
_engines[which].connect()
_metadata[which].bind = _engines[which]
return _engines[which]
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment