Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- ##################################
- #in separate module
- from sqlalchemy.engine.url import URL
- from sqlalchemy.ext.sqlsoup import SqlSoup
- import sqlalchemy as sa
- class SQLAlchemyTablesDict(object):
- drivermap = {
- "sqlite3": "sqlite",
- "postgresql_psycopg2": "postgres",
- "mysql": "mysql",
- }
- def __init__(self, drivername, username=None, password=None, host=None, port=None, database=None, query=None):
- url = URL(SQLAlchemyTablesDict.drivermap[drivername], username=username, password=password, host=host, port=None, database=database, query=query)
- self.engine = sa.create_engine(url)
- self._soup = SqlSoup(self.engine)
- def __getitem__(self, key):
- mapped = self._soup.__getattr__(key)
- return mapped._table
- ##################################
- #in settings.py
- SA_TABLES = SQLAlchemyTablesDict(DATABASE_ENGINE, username=DATABASE_USER, password=DATABASE_PASSWORD, host=DATABASE_HOST, port=DATABASE_PORT, database=DATABASE_NAME)
- ##################################
- #in dao module
- def my_dao_fun(par1):
- mytable = SA_TABLES["myapp_mytable"]
- conn = SA_TABLES.engine.connect()
- #...do query
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement