Created
August 25, 2014 02:54
-
-
Save dpnova/7e3dd4f6de2b7748bec1 to your computer and use it in GitHub Desktop.
Create sqlalchemy engines and metadata from a Django-ish set of settings.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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