Home
Reading
Searching
Subscribe
Sponsors
Statistics
Posting
Contact
Spam
Lists
Links
About
Hosting
Filtering
Features Download
Marketing
Archives
FAQ
Blog
 
Gmane
From: jt <jt <at> udev.org>
Subject: Re: sqlite connection and user-defined aggregate
Newsgroups: gmane.comp.python.sqlobject
Date: Thursday 22nd June 2006 09:08:34 UTC (over 11 years ago)
Thanks Oleg. Dirty hack indeed :-)

Looking at sqliteconnection.py and dbconnection.py I can see that what
I want should work out of the box since I do not use multiple threads:
getConnection should return the same sqliteconnection if you have not
released it.
I was also mislead because I called getConnection() on transaction
object but I shall use the _connection attribute so that the
sqliteconnection survives with the transaction object.

I also tried the factory argument of pysqlite connection (could you
add it to the trunk?):
============================================================
--- sqliteconnection.py (revision 1824)
+++ sqliteconnection.py (working copy)
@@ -54,6 +54,8 @@
             if sqlite2_Binary is None:
                 sqlite2_Binary = sqlite.Binary
                 sqlite.Binary = lambda s: sqlite2_Binary(sqlite.encode(s))
+            if 'factory' in kw:
+                opts['factory'] = popKey(kw, 'factory')(sqlite)
         else:
             opts['autocommit'] = bool(autoCommit)
             if 'encoding' in kw:
============================================================

and in my code:
============================================================
def SQLiteConnectionFactory(sqlite):

        class MyConnection(sqlite.Connection):
                def regexp(expr, item):
                        r = re.compile(expr)
                        return r.match(item) is not None

                class group_concat:
                        def __init__(self):
                                self.acc = []
                        def step(self, value):
                                if isinstance(value, basestring):
                                        self.acc.append(value)
                                else:
                                        self.acc.append(str(value))
                        def finalize(self):
                                self.acc.sort()
                                return ", ".join(self.acc)

                def __init__(self, *args, **kwargs):
                        super(MyConnection, self).__init__(*args, **kwargs)
                        self.create_function("regexp", 2, self.regexp)
                        self.create_aggregate("group_concat", 1,
self.group_concat)

        return MyConnection

con = SQLiteConnection('data.db', factory=SQLiteConnectionFactory)
============================================================

I hope this makes possible to clean up the RLIKE test.

-- 
jt
 
CD: 18ms