-
-
Save foxcarlos/c0b6f63dbd91fb09de633e8a6662be7f to your computer and use it in GitHub Desktop.
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
import pymssql | |
import pymysql | |
class CursorMixin(object): | |
def __init__(self, dbname, dsn): | |
"""If you have implemented this method you will need define | |
self._cnx with the connection to database | |
self._obj with the cursor | |
""" | |
raise NotImplementYet | |
def execute(self, query, params=None): | |
self._obj.execute(query, params) | |
def __del__(self): | |
if not self.connect.closed: | |
self.execute("ROLLBACK") | |
self.close() | |
class CursorMSSQL(CursorMixin): | |
def __init__(self, dbname, dsn): | |
self._cnx = pymssql.connect(dbname, **dsn) | |
self._obj = self._cnx.cursor() | |
def close(self): | |
pymssql.close() | |
class CursorMySQL(CursorMixin): | |
def __init__(self, dbname, dsn): | |
self._cnx = mysql.connect(dbname, **dsn) | |
self._obj = self._cnx.cursor() | |
def close(self): | |
pymssql.close() | |
class DbDriver(models.Model): | |
_name = 'db.connection' | |
name = fields.Char() | |
driver = fields.Selection([ | |
('my_sql', 'My SQL'), | |
('ms_sql', 'MS SQL'), | |
]) | |
user = fields.Char() | |
password = fields.Char() | |
hostname = fields.Char() | |
database = fields.Char(help="Optional parameter to connect by default if it not defined a database name from connect method") | |
# def connect_mysql(self): | |
# return "conneted to My-SQL" | |
# def connect_mssql(self): | |
# return "connected to MS SQL" | |
def connect(self, driver, database=None): | |
api.ensure_one() | |
if database is None: | |
database = self.database | |
if driver == 'my_sql': | |
return CursorMySQL(database, self.username, self.password) | |
if driver == 'ms_sql': | |
return CursorMSSQL(database, self.username, self.password) | |
raise "Driver %s not implement yet" % (driver) | |
# method_name = "connect_%s" % driver | |
# print("Method to call %s" % method_name) | |
# method = getattr(self, method_name) | |
# return method() | |
# obj = DbDriver() | |
# res = obj.connect('mysql') | |
# print(res) | |
def _get_customer(self) | |
connect = self.env['db.connection'].browse(1).connect('myssql', 'JSAT') | |
connect.execute('SELECT * FROM table') | |
def get_customer(self): | |
return sudo()._get_customer() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment