Clase para conexion con base de datos OpenSource MIT

conexion.py

#!/usr/bin/python
# Libreria de conexion a base de datos
# Autor: Dario R. Garcia G. DR2GSistemas
# Villeta - Paraguay
#
# Version: 1.02
# - Agregado runtime exception
# Version: 1.01
# - Bug fix: control de host y port
# Version: 1.00
# - Inicial
#
# Requerimientos:
# - SQLAlchemy, pg8000, pysqlite, pymysql, fdb, cx-oracle

from sqlalchemy import *
from sqlalchemy.engine import RowProxy


class DatabaseBase(object):
    """Clase base para conexion a base de datos"""
    _username = None
    _userpass = None
    _host = None
    _port = None
    _database_name = None
    _url = None
    _metadata = None
    _engine = None
    _conexion = None

    def __init__(self, *args, **kwargs):
        self._username = kwargs.get('username', None)
        self._userpass = kwargs.get('userpass', None)
        self._host = kwargs.get('host', None)
        self._port = kwargs.get('port', None)
        self._database_name = kwargs.get('database_name', None)

        if self._host is None or self._username is None or self._userpass is None:
            raise RuntimeError("host no definido")

    def gettable(self, tablename):
        return self._metadata.tables[tablename]

    def select_one(self, query, *args, **kwargs):
        return self._conexion.execute(text(query), *args, **kwargs).fetchone()

    def select_many(self, query, *args, **kwargs):
        obj = self._conexion.execute(text(query), *args, **kwargs).fetchall()
        return dict(obj)

    def select_scalar(self, query, *args, **kwargs):
        return self._conexion.execute(text(query), *args, **kwargs).scalar()

    @staticmethod
    def todict(obj):
        if isinstance(obj, (int,)):
            return dict(value=obj)
        else:
            if obj:
                return [dict(item) for item in obj]
            else:
                return []

    @staticmethod
    def rowproxy2dict(obj):
        if isinstance(obj, (RowProxy,)):
            return dict(obj)
        else:
            return obj

    # @event.listens_for(Engine, "connect")
    # def set_sqlite_pragma(dbapi_connection, connection_record):
    #     cursor = dbapi_connection.cursor()
    #     cursor.execute('PRAGMA foreign_keys=ON')
    #     cursor.close()


class DatabaseSQLite(DatabaseBase):

    def __init__(self, url=None, *args, **kwargs):
        super().__init__(*args, **kwargs)
        if url is None:
            raise Exception('Url no definida')

        self.url = 'sqlite:///{}'.format(url)
        self.engine = create_engine(self.url)
        self.conexion = self.engine.connect()
        self.metadata = MetaData(self.engine)
        self.metadata.reflect(bind=self.engine)