mirror of
https://github.com/yweber/lodel2.git
synced 2026-05-02 13:10:58 +02:00
85 lines
2.8 KiB
Python
85 lines
2.8 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import os
|
|
import logging as logger
|
|
|
|
import sqlalchemy as sql
|
|
|
|
from Database.sqlwrapper import SqlWrapper
|
|
|
|
class SqlQueryBuilder():
|
|
|
|
def __init__(self, sqlwrapper, table):
|
|
if not type(sqlwrapper) is SqlWrapper:
|
|
logger.error("Unable to instanciate, bad argument...")
|
|
raise TypeError('Excepted a SqlWrapper not a '+str(type(sqlwrapper)))
|
|
self.table = table
|
|
self.sqlwrapper = sqlwrapper
|
|
self.proxy = None
|
|
|
|
|
|
def Select(self, arg):
|
|
""" Alias for select clause
|
|
@param arg iterable: arg must be a Python list or other iterable and contain eather table name/type or colum/literal_column
|
|
"""
|
|
|
|
self.proxy = sql.select(arg)
|
|
return self.proxy
|
|
|
|
def Where(self, arg):
|
|
""" Alias for where clause
|
|
@param arg SQL expression object or string
|
|
"""
|
|
self.proxy = self.proxy.where(arg)
|
|
|
|
def From(self, arg):
|
|
""" Alias for select_from clause
|
|
@param arg Table or table('tablename') or join clause
|
|
"""
|
|
self.proxy = self.proxy.select_from(arg)
|
|
|
|
def Update(self):
|
|
self.proxy = self.table.update()
|
|
|
|
def Insert(self):
|
|
self.proxy = self.table.insert()
|
|
|
|
def Delete(self):
|
|
self.proxy = self.proxy.delete()
|
|
|
|
def Value(self, arg):
|
|
"""
|
|
Allow you to specifies the VALUES or SET clause of the statement.
|
|
@param arg: VALUES or SET clause
|
|
"""
|
|
self.proxy = self.proxy.values(arg)
|
|
|
|
def Execute(self, bindedparam):
|
|
"""
|
|
Execute the sql query constructed in the proxy and return the result.
|
|
If no query then return False.
|
|
@return: query result on success else False
|
|
"""
|
|
if(self.proxy.__str__().split() == 'SELECT'):
|
|
if('bindparam' in self.proxy.__str__()):
|
|
#on test separement la présence de la clause bindparam et le type de l'argument correspondant
|
|
#car si la clause est présente mais que l'argument est defectueux on doit renvoyer False et non pas executer la requete
|
|
if(type(bindedparam) is list and type(bindedparam[0]) is dict):
|
|
return self.sqlwrapper.rconn.execute(self.proxy, bindedparam)
|
|
else:
|
|
return False
|
|
else:
|
|
return self.sqlwrapper.rconn.execute(self.proxy)
|
|
elif(self.proxy is not None):
|
|
if('bindparam' in self.proxy.__str__()):
|
|
if(type(bindedparam) is list and type(bindedparam[0]) is dict):
|
|
return self.sqlwrapper.wconn.execute(self.proxy, bindedparam)
|
|
else:
|
|
return False
|
|
else:
|
|
return self.sqlwrapper.wconn.execute(self.proxy)
|
|
else:
|
|
return False
|
|
|
|
|
|
|