Hi,
I though I post my current work here for comments and also if someone
needs it :-D
I hope to be able to integrate this work in to SQLObject at some point,
or help out
with a similar effort. (I've just started looking at the SQLObject devel
process.)
Notes: I've think this (most) work, but I've not written test.
Strange things happen with things like count or sum when I use groupBy,
but given the SQL create it is expected.
(with groupBy you get a list of count/sum etc. SelectResults assumes a
single item. I'm hoping to imrpove this
in the future, but is a bit more tricky and will involve changing core
SQLObject)
Best Regards,
Johan Carlsson
from sqlobject.dbconnection import Iteration
from sqlobject.sresults import SelectResults
from sqlobject.sqlbuilder import True
from sqlobject import *
from sqlobject.sqlbuilder import *
class MyIteration(Iteration):
def __init__(self, dbconn, rawconn, select, keepConnection=False):
self.dbconn = dbconn
self.rawconn = rawconn
self.select = select
self.keepConnection = keepConnection
self.cursor = rawconn.cursor()
self.query = dbconn.sqlrepr(select.queryForSelect())
if dbconn.debug:
dbconn.printDebug(rawconn, self.query, 'Select')
self.dbconn._executeRetry(self.rawconn, self.cursor, self.query)
class BuilderResults(SelectResults):
IterationClass = MyIteration
def __init__(self, sourceClass, select, clause=True,
clauseTables=None, **ops):
super(BuilderResults, self).__init__(sourceClass, clause,
clauseTables=clauseTables, **ops)
conn=self._getConnection()
self.select=select
def queryForSelect(self):
conn=self._getConnection()
ops = self.ops.copy()
del ops['dbOrderBy']
select= self.select.clone(**ops)
return select
def clone(self, **newOps):
ops = self.ops.copy()
ops.update(newOps)
return self.__class__(self.sourceClass, self.select,
self.clause, self.clauseTables, **ops)
def newClause(self, new_clause):
return self.__class__(self.sourceClass, self.select, new_clause,
self.clauseTables, **self.ops)
sqlhub = dbconnection.ConnectionHub()
class Brain:
_connection = sqlhub
columnList=[]
sqlmeta = sqlmeta
def _init(self, id, connection=None, selectResults=None):
self.id=id
if connection is not None:
self._connection = connection
if selectResults and len(selectResults)>=len(self.columnList):
i=0
for name in self.columnList:
setattr(self, name, selectResults[i])
i+=1
def get(cls, id, connection=None, selectResults=None):
assert id is not None, 'None is not a possible id for %s' %
cls.__name__
self=cls()
self._init(id, connection, selectResults)
return self
get = classmethod(get)
-------------------------------------------------------------------------
This SF.net email is sponsored by: Splunk Inc.
Still grepping through log files to find problems? Stop.
Now Search log events and configuration files using AJAX and a browser.
Download your FREE copy of Splunk now >> http://get.splunk.com/
_______________________________________________
sqlobject-discuss mailing list
sqlobject-discuss@lists.sourceforge.net
https://lists.sourceforge.net/lists/listinfo/sqlobject-discuss
