diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
new file mode 100644
index 0000000..d9536f6
--- /dev/null
+++ b/.github/workflows/test.yml
@@ -0,0 +1,26 @@
+name: CI
+
+on:
+ push:
+ branches: [ master ]
+
+jobs:
+ build:
+
+ runs-on: ubuntu-latest
+ strategy:
+ matrix:
+ python-version: [2.7]
+
+ steps:
+ - uses: actions/checkout@v2
+ - name: Set up Python ${{ matrix.python-version }}
+ uses: actions/setup-python@v2
+ with:
+ python-version: ${{ matrix.python-version }}
+ - name: Install dependencies
+ run: python -m pip install -q Twisted
+ - name: Run txconnectionpool tests
+ run: trial txconnectionpool
+ - name: Run test
+ run: trial twistar
diff --git a/.gitignore b/.gitignore
index ba31055..384eda4 100644
--- a/.gitignore
+++ b/.gitignore
@@ -4,4 +4,3 @@ doc/*.html
_trial_temp
build
dist
-twistar.egg-info
\ No newline at end of file
diff --git a/INSTALL b/INSTALL
deleted file mode 100644
index b11190f..0000000
--- a/INSTALL
+++ /dev/null
@@ -1,24 +0,0 @@
-= Prerequisites =
-You must first install Twisted core. If you intent on generating API documentation,
-you will need pydoctor. If you want to generate the user documentation, you will
-need to install Twisted Lore. You will need to install the sqlite3 python package
-if you want to run tests.
-
-Your database must be one of: MySQL, PostgreSQL, or SQLite. The only DBAPI modules
-supported by Twistar are: MySQLdb, psycopg2, and sqlite3 - at least one of these
-must be installed.
-
-
-= Installation =
-$> sudo make install
-
-To generate documentation:
-$> sudo make docs
-$> firefox docs/index.html
-
-= Testing =
-Just run:
-$> sudo make test
-
-See the README in the twistar/tests folder for more information on testing with
-different database types.
\ No newline at end of file
diff --git a/README.markdown b/README.markdown
new file mode 100644
index 0000000..b15e839
--- /dev/null
+++ b/README.markdown
@@ -0,0 +1,80 @@
+# twistar: Asynchronous Python ORM
+[](https://travis-ci.org/xadhoom/twistar)
+
+The Twistar Project provides an ActiveRecord (ORM) pattern interface to the Twisted Project's RDBMS library. This file contains minimal documentation - see the project home page at http://findingscience.com/twistar for more information.
+
+## Installation
+
+```
+easy_install twistar
+```
+
+## Usage
+Your database must be one of: MySQL, PostgreSQL, or SQLite. The only DBAPI modules supported by Twistar are: MySQLdb, psycopg2, and sqlite3 - at least one of these must be installed.
+
+Here's the obligatory TL;DR example of creating a User record, assuming that there is a table named "users" with varchar columns for first_name and last_name and an int age column:
+
+```python
+#!/usr/bin/env python
+from twisted.enterprise import adbapi
+from twistar.registry import Registry
+from twistar.dbobject import DBObject
+from twisted.internet import reactor
+
+class User(DBObject):
+ pass
+
+def done(user):
+ print "A user was just created with the name %s" % user.first_name
+ reactor.stop()
+
+# Connect to the DB
+Registry.DBPOOL = adbapi.ConnectionPool('MySQLdb', user="twistar", passwd="apass", db="twistar")
+
+# make a user
+u = User()
+u.first_name = "John"
+u.last_name = "Smith"
+u.age = 25
+
+# Or, use this shorter version:
+u = User(first_name="John", last_name="Smith", age=25)
+
+# save the user
+u.save().addCallback(done)
+
+reactor.run()
+```
+
+Then, finding this user is easy:
+
+```python
+def found(users):
+ print "I found %i users!" % len(users)
+ for user in users:
+ print "User: %s %s" % (user.first_name, user.last_name)
+
+u = User.findBy(first_name="John", age=25).addCallback(found)
+```
+
+This is a very simple example - see http://findingscience.com/twistar for more complicated examples and additional uses.
+
+## Testing
+You will need to install the sqlite3 python package if you want to run the default tests. To run the tests:
+
+```
+trial twistar
+```
+
+See the README in the twistar/tests folder for more information on testing with different database types.
+
+## Documenation
+If you intent on generating API documentation, you will need pydoctor. If you want to generate the user documentation, you will need to install Twisted Lore.
+
+To generate documentation:
+
+```
+make docs
+```
+
+Then open the docs/index.html file in a browser.
diff --git a/README b/README.rst
similarity index 78%
rename from README
rename to README.rst
index b711e6e..fc61f50 100644
--- a/README
+++ b/README.rst
@@ -1,3 +1,8 @@
+Twistar
+=======
+
+.. image:: https://secure.travis-ci.org/xadhoom/twistar.png?branch=master
+
The Twistar Project provides an ActiveRecord (ORM) pattern interface to
the Twisted Project's RDBMS library.
@@ -6,3 +11,4 @@ those instructions are the steps to generating API, docs, and HOWTO examples.
For more information, see the project home page at:
http://findingscience.com/twistar
+
diff --git a/doc/examples.xhtml b/doc/examples.xhtml
index 547c148..df94088 100644
--- a/doc/examples.xhtml
+++ b/doc/examples.xhtml
@@ -9,13 +9,13 @@
Simple Examples
- Since Twistar uses Twisted, most methodand function calls willreturn a twisted.internet.defer.Deferred
- object. You must understand the propper use of these objects before most of this documentationwill make sense.
+ Since Twistar uses Twisted, most methods and function calls will return a twisted.internet.defer.Deferred
+ object. You must understand the proper use of these objects before most of this documentation will make sense.
Initialization
- Before using Twistar a connection to the DB must first be made. The method of connecting to the database
-is to use Twisted's twisted.enterprise.adapi module and the ConnectionPool
+
Before using Twistar a connection to the DB must be made. To connect to the database
+use Twisted's twisted.enterprise.adapi module and the ConnectionPool
object. The Registry class is used to keep track of that pool. For instance, this is the
method to specify a connection to a MySQL database:
@@ -54,7 +54,7 @@ an auto-incrementing integer column named id
.
Defining and Interacting With Objects
-Object's representing rows in a database need only extend the
+Objects representing rows in a database need only extend the
DBObject class.
@@ -147,13 +147,13 @@ User.exists(['first_name = ?', "John"]).addCallback(printExists)
Validations
-Twistar also supports validations. Validations describe constraints on an objects
+Twistar also supports validations. Validations describe constraints on an object's
parameters. If those constraints are violated, then an object will not be saved (this
includes both creating and updating). In such a case a special parameter in the object
keeps track of the errors and the messages.
-As an example, let's say that we want all users to always have a first name set and a last
+As an example, let's say that we want all users to have a first name set and a last
name that has a length between 1 and 256 characters. First, we describe the class
User as above and then add restrictions on it.
diff --git a/doc/index.xhtml b/doc/index.xhtml
index f46e689..b454889 100644
--- a/doc/index.xhtml
+++ b/doc/index.xhtml
@@ -24,6 +24,7 @@ object. You must understand the propper use of these objects before most of thi
Basic Examples
Examples With Relationships
Lowlevel Database Interaction
+ Examples with Transactions
diff --git a/doc/relationship_examples.xhtml b/doc/relationship_examples.xhtml
index 4d52000..b00c17b 100644
--- a/doc/relationship_examples.xhtml
+++ b/doc/relationship_examples.xhtml
@@ -150,12 +150,12 @@ Registry.register(User, Picture)
Assuming the DB structure is the same as in the "has many" example, you can now get and set the user that
a particular picture belongs to (using get() and set()
methods on picture.user). Additionally, there is a clear() method that will clear all
-objects in a given belongs to relationship.
+objects in a given "belongs to" relationship.
Has and Belongs To Many
-In a has and belongs to many relationship two objects have a many to many relationship. For instance, users and favorite colors.
-A user can have many favorite colors and a favorite color could belong to many users. In this example, there should be three tables -
+In a "has and belongs to many" relationship two objects have a many to many relationship. For instance, users and favorite colors.
+A user can have many favorite colors and a favorite color could belong to many users. In this example, there should be three tables —
a users table, a favorite_colors table, and a favorite_colors_users table. The last on the list
is a special table that stores the relationships between the colors and users. It has no primary key, and two columns: one for
user_ids and one for favorite_color_ids. The table's name by convention should be the combination of the other
diff --git a/doc/transactions.xhtml b/doc/transactions.xhtml
new file mode 100644
index 0000000..2b9e59e
--- /dev/null
+++ b/doc/transactions.xhtml
@@ -0,0 +1,112 @@
+
+
+
+
+
+ Transaction Examples
+
+
+
+ Transaction Examples
+From Active Record Transactions:
+Transactions are protective blocks where SQL statements are only permanent if they can all succeed as one atomic action.
+The classic example is a transfer between two accounts where you can only have a deposit if the withdrawal succeeded and vice versa.
+Transactions enforce the integrity of the database and guard the data against program errors or database break-downs.
+So basically you should use transactions whenever you have a number of statements that must be executed together or not at all.
+
+
+More examples can be found in the RelationshipTest class.
+
+ Initialization
+ Given a DBObject class instance, developer can call transaction()
+ method on it, causing the start of the transaction. The method will return a transaction identifier.
+
+from twisted.internet.defer import inlineCallback
+
+@inlineCallbacks
+def do_some_pen_operations:
+ pen = Pen()
+ txn = pen.transaction()
+
+
+ Commit / Rollback
+ In this example, we will be using a Pen object, which extends DBObject class.
+ WARNING: pseudocode!
+
+from twisted.internet.defer import inlineCallback
+
+@inlineCallbacks
+def do_some_pen_operations:
+ pen = Pen()
+ pen.transaction()
+
+ pen.owner = "Luke"
+ yield pen.save()
+
+ pen.use() /* do something with the pen */
+
+ /* check if the owner is still Luke, after use */
+ if pen.owner == "Luke":
+ yield pen.commit()
+ else:
+ /* owner has changed, cannot apply ! */
+ yield pen.rollback()
+
+
+ Different classes in a single transaction
+ In this example, we use the same transaction across different objects, to bind all operations on them to same handler.
+ WARNING: pseudocode!
+
+from twisted.internet.defer import inlineCallback
+
+@inlineCallbacks
+def do_some_financial_operations:
+ money = Balance()
+ txn = money.transaction()
+
+ account = Account(transaction=txn)
+
+ money.total = 100
+ account.total = -100
+
+
+ yield money.save()
+ yield account.save()
+
+ try:
+ money.commit() /* or account.commit() */
+ except Exception, e:
+ money.rollback()
+
+ Notes
+ After a rollback or commit, the transaction is closed. You need to call transaction() method again
+ to start a new transaction.
+ Depending on DB engine, you cannot call save on a DBObject instance if same class type has an open transaction
+ (not committed yet), because the code will block waiting for the first transaction to finish. If the first open transaction will finish
+ before operation timeout, save will be done, otherwise will get executed after the transaction timeout of the DB engine. For example :
+ WARNING: pseudocode!
+
+from twisted.internet.defer import inlineCallback
+
+@inlineCallbacks
+def do_some_pen_operations:
+ pen = Pen(color="red")
+ pen.transaction() /* start transaction only on pen */
+
+ another_pen = Pen(color="blue") /* this is not in transaction */
+
+ yield pen.save()
+
+ /* code will block here utile DB engine timeout ... after that save() will throw an Exception */
+ /* after that, neither pen and another_pen are really save, the former because commit has not been called,
+ * the latter because the DB engine throwed an exception */
+ yield another_pen.save()
+
+ /* not executed at all */
+ yield pen.commit()
+ yield another_pen.commit()
+
+
+
+
+
diff --git a/setup.py b/setup.py
index 3012c12..d36b5f5 100755
--- a/setup.py
+++ b/setup.py
@@ -1,17 +1,17 @@
#!/usr/bin/env python
-try:
- from setuptools import setup, Extension
-except ImportError:
- from distutils.core import setup, Extension
+from setuptools import setup, find_packages
+
+from twistar import version
setup(
name="twistar",
- version="1.0",
+ version=version,
description="An implementation of the Active Record pattern for Twisted",
author="Brian Muller",
author_email="bamuller@gmail.com",
- license="GPLv3",
+ license="MIT",
url="http://findingscience.com/twistar",
- packages=["twistar", "twistar.dbconfig", "twistar.tests", 'BermiInflector', 'BermiInflector.Rules'],
- requires=["twisted.enterprise.adbapi"]
+ packages=find_packages(),
+ requires=["twisted.enterprise.adbapi"],
+ install_requires=['twisted >= 12.1']
)
diff --git a/twistar/__init__.py b/twistar/__init__.py
index d7441f6..48324bd 100644
--- a/twistar/__init__.py
+++ b/twistar/__init__.py
@@ -6,6 +6,5 @@
@author: Brian Muller U{bamuller@gmail.com}
"""
-
-version = "0.1"
-version_info = (0, 1)
+version_info = (1, 3)
+version = '.'.join(map(str, version_info))
diff --git a/twistar/dbconfig/base.py b/twistar/dbconfig/base.py
index 64a5542..e4ea573 100644
--- a/twistar/dbconfig/base.py
+++ b/twistar/dbconfig/base.py
@@ -1,11 +1,13 @@
"""
Base module for interfacing with databases.
"""
+import sys
from twisted.python import log
from twistar.registry import Registry
from twistar.exceptions import ImaginaryTableError, CannotRefreshError
+from twistar.exceptions import TransactionNotStartedError
class InteractionBase:
"""
@@ -43,6 +45,13 @@ def log(self, query, args, kwargs):
elif len(kwargs) > 0:
log.msg("TWISTAR kargs: %s" % str(kwargs))
+ def executeOperationInTransaction(self, query, transaction, *args, **kwargs):
+ """
+ Simply makes same C{twisted.enterprise.dbapi.ConnectionPool.runOperation} call, but
+ with call to L{log} function.
+ """
+ self.log(query, args, kwargs)
+ return Registry.DBPOOL.runOperationInTransaction(transaction, query, *args, **kwargs)
def executeOperation(self, query, *args, **kwargs):
"""
@@ -62,16 +71,26 @@ def execute(self, query, *args, **kwargs):
return Registry.DBPOOL.runQuery(query, *args, **kwargs)
- def executeTxn(self, txn, query, *args, **kwargs):
+ def startTxn(self):
+ """
+ Start a transaction.
+
+ @return: a C{t.e.a.Transaction} instance
+ @rtype: C{deferred}
+ """
+ return Registry.DBPOOL.startTransaction()
+
+
+ def executeTxn(self, transaction, query, *args, **kwargs):
"""
Execute given query within the given transaction. Also, makes call
to L{log} function.
"""
self.log(query, args, kwargs)
- return txn.execute(query, *args, **kwargs)
+ return transaction.execute(query, *args, **kwargs)
- def select(self, tablename, id=None, where=None, group=None, limit=None, orderby=None, select=None):
+ def select(self, tablename, id=None, where=None, group=None, limit=None, orderby=None, select=None, transaction=None):
"""
Select rows from a table.
@@ -92,19 +111,36 @@ def select(self, tablename, id=None, where=None, group=None, limit=None, orderby
@param select: Columns to select. Default is C{*}.
+ @param transaction: An optional C{t.e.a.Transaction} instance.
+
@return: If C{limit} is 1 or id is set, then the result is one dictionary or None if not found.
Otherwise, an array of dictionaries are returned.
"""
one = False
- select = select or "*"
if id is not None:
- where = ["id = ?", id]
one = True
if not isinstance(limit, tuple) and limit is not None and int(limit) == 1:
one = True
+ q, args = self._build_select( tablename, id, where, group, limit, orderby, select )
+
+ if transaction:
+ return Registry.DBPOOL.executeOperationInTransaction(self._doselect, transaction, q, args, tablename, one)
+ else:
+ return Registry.DBPOOL.runInteraction(self._doselect, q, args, tablename, one)
+
+
+ def _build_select(self, tablename, id=None, where=None, group=None, limit=None, orderby=None, select=None):
+ """
+ Private helper to actually build the query strings and it's args
+ """
+ select = select or "*"
+
+ if id is not None:
+ where = ["id = ?", id]
+
q = "SELECT %s FROM %s" % (select, tablename)
args = []
if where is not None:
@@ -119,26 +155,26 @@ def select(self, tablename, id=None, where=None, group=None, limit=None, orderby
q += " LIMIT %s OFFSET %s" % (limit[0], limit[1])
elif limit is not None:
q += " LIMIT " + str(limit)
-
- return Registry.DBPOOL.runInteraction(self._doselect, q, args, tablename, one)
+ return q, args
- def _doselect(self, txn, q, args, tablename, one=False):
+
+ def _doselect(self, transaction, q, args, tablename, one=False):
"""
Private callback for actual select query call.
"""
- self.executeTxn(txn, q, args)
+ self.executeTxn(transaction, q, args)
if one:
- result = txn.fetchone()
+ result = transaction.fetchone()
if not result:
return None
- vals = self.valuesToHash(txn, result, tablename)
+ vals = self.valuesToHash(transaction, result, tablename)
return vals
results = []
- for result in txn.fetchall():
- vals = self.valuesToHash(txn, result, tablename)
+ for result in transaction.fetchall():
+ vals = self.valuesToHash(transaction, result, tablename)
results.append(vals)
return results
@@ -150,7 +186,7 @@ def insertArgsToString(self, vals):
return "(" + ",".join(["%s" for _ in vals.items()]) + ")"
- def insert(self, tablename, vals, txn=None):
+ def insert(self, tablename, vals, transaction=None):
"""
Insert a row into the given table.
@@ -159,7 +195,7 @@ def insert(self, tablename, vals, txn=None):
@param vals: Values to insert. Should be a dictionary in the form of
C{{'name': value, 'othername': value}}.
- @param txn: If txn is given it will be used for the query,
+ @param transaction: If transaction is given it will be used for the query,
otherwise a typical runQuery will be used
@return: A C{Deferred} that calls a callback with the id of new row.
@@ -171,8 +207,8 @@ def insert(self, tablename, vals, txn=None):
colnames = "(" + ",".join(ecolnames) + ")"
params = "VALUES %s" % params
q = "INSERT INTO %s %s %s" % (tablename, colnames, params)
- if not txn is None:
- return self.executeTxn(txn, q, vals.values())
+ if transaction:
+ return self.executeTxn(transaction, q, vals.values())
return self.executeOperation(q, vals.values())
@@ -187,7 +223,7 @@ def escapeColNames(self, colnames):
return map(lambda x: "`%s`" % x, colnames)
- def insertMany(self, tablename, vals):
+ def insertMany(self, tablename, vals, transaction=None):
"""
Insert many values into a table.
@@ -204,28 +240,33 @@ def insertMany(self, tablename, vals):
for val in vals:
args = args + val.values()
q = "INSERT INTO %s (%s) VALUES %s" % (tablename, colnames, params)
+ if transaction is not None:
+ return self.executeTxn(transaction, q, args)
return self.executeOperation(q, args)
- def getLastInsertID(self, txn):
+ def getLastInsertID(self, transaction):
"""
- Using the given txn, get the id of the last inserted row.
+ Using the given transaction, get the id of the last inserted row.
@return: The integer id of the last inserted row.
"""
q = "SELECT LAST_INSERT_ID()"
- self.executeTxn(txn, q)
- result = txn.fetchall()
+ self.executeTxn(transaction, q)
+ result = transaction.fetchall()
return result[0][0]
- def delete(self, tablename, where=None):
+ def delete(self, tablename, where=None, transaction=None):
"""
Delete from the given tablename.
@param where: Conditional of the same form as the C{where} parameter in L{DBObject.find}.
If given, the rows deleted will be restricted to ones matching this conditional.
+ @param transaction: If transaction is given it will be used for the query,
+ otherwise a typical runQuery will be used
+
@return: A C{Deferred}.
"""
q = "DELETE FROM %s" % tablename
@@ -233,10 +274,12 @@ def delete(self, tablename, where=None):
if where is not None:
wherestr, args = self.whereToString(where)
q += " WHERE " + wherestr
+ if transaction is not None:
+ return self.executeOperationInTransaction(q, transaction, args)
return self.executeOperation(q, args)
- def update(self, tablename, args, where=None, txn=None):
+ def update(self, tablename, args, where=None, transaction=None, limit=None):
"""
Update a row into the given table.
@@ -248,9 +291,11 @@ def update(self, tablename, args, where=None, txn=None):
@param where: Conditional of the same form as the C{where} parameter in L{DBObject.find}.
If given, the rows updated will be restricted to ones matching this conditional.
- @param txn: If txn is given it will be used for the query,
+ @param transaction: If transaction is given it will be used for the query,
otherwise a typical runQuery will be used
+ @param limit: If limit is given it will limit the number of rows that are updated.
+
@return: A C{Deferred}
"""
setstring, args = self.updateArgsToString(args)
@@ -259,25 +304,27 @@ def update(self, tablename, args, where=None, txn=None):
wherestr, whereargs = self.whereToString(where)
q += " WHERE " + wherestr
args += whereargs
+ if limit is not None:
+ q += " LIMIT " + str(limit)
- if txn is not None:
- return self.executeTxn(txn, q, args)
+ if transaction is not None:
+ return self.executeTxn(transaction, q, args)
return self.executeOperation(q, args)
- def valuesToHash(self, txn, values, tablename):
+ def valuesToHash(self, transaction, values, tablename):
"""
Given a row from a database query (values), create
a hash using keys from the table schema and values from
the given values;
- @param txn: The transaction to use for the schema update query.
+ @param transaction: The transaction to use for the schema update query.
@param values: A row from a db (as a C{list}).
@param tablename: Name of the table to fetch the schema for.
"""
- cols = [row[0] for row in txn.description]
+ cols = [row[0] for row in transaction.description]
if not Registry.SCHEMAS.has_key(tablename):
Registry.SCHEMAS[tablename] = cols
h = {}
@@ -287,17 +334,17 @@ def valuesToHash(self, txn, values, tablename):
return h
- def getSchema(self, tablename, txn=None):
+ def getSchema(self, tablename, transaction=None):
"""
Get the schema (in the form of a list of column names) for
a given tablename. Use the given transaction if specified.
"""
- if not Registry.SCHEMAS.has_key(tablename) and txn is not None:
+ if not Registry.SCHEMAS.has_key(tablename) and transaction is not None:
try:
- self.executeTxn(txn, "SELECT * FROM %s LIMIT 1" % tablename)
+ self.executeTxn(transaction, "SELECT * FROM %s LIMIT 1" % tablename)
except Exception, e:
raise ImaginaryTableError, "Table %s does not exist." % tablename
- Registry.SCHEMAS[tablename] = [row[0] for row in txn.description]
+ Registry.SCHEMAS[tablename] = [row[0] for row in transaction.description]
return Registry.SCHEMAS.get(tablename, [])
@@ -307,17 +354,20 @@ def insertObj(self, obj):
@return: A C{Deferred} that sends a callback the inserted object.
"""
- def _doinsert(txn):
+ def _doinsert(transaction):
klass = obj.__class__
tablename = klass.tablename()
- cols = self.getSchema(tablename, txn)
+ cols = self.getSchema(tablename, transaction)
if len(cols) == 0:
raise ImaginaryTableError, "Table %s does not exist." % tablename
vals = obj.toHash(cols, includeBlank=self.__class__.includeBlankInInsert, exclude=['id'])
- self.insert(tablename, vals, txn)
- obj.id = self.getLastInsertID(txn)
+ self.insert(tablename, vals, transaction)
+ obj.id = self.getLastInsertID(transaction)
return obj
- return Registry.DBPOOL.runInteraction(_doinsert)
+ if obj._transaction:
+ return Registry.DBPOOL.executeOperationInTransaction(_doinsert, obj._transaction)
+ else:
+ return Registry.DBPOOL.runInteraction(_doinsert)
def updateObj(self, obj):
@@ -326,15 +376,18 @@ def updateObj(self, obj):
@return: A C{Deferred} that sends a callback the updated object.
"""
- def _doupdate(txn):
+ def _doupdate(transaction):
klass = obj.__class__
tablename = klass.tablename()
- cols = self.getSchema(tablename, txn)
-
+ cols = self.getSchema(tablename, transaction)
+
vals = obj.toHash(cols, includeBlank=True, exclude=['id'])
- return self.update(tablename, vals, where=['id = ?', obj.id], txn=txn)
+ return self.update(tablename, vals, where=['id = ?', obj.id], transaction=transaction)
# We don't want to return the cursor - so add a blank callback returning the obj
- return Registry.DBPOOL.runInteraction(_doupdate).addCallback(lambda _: obj)
+ if obj._transaction:
+ return Registry.DBPOOL.executeOperationInTransaction(_doupdate, obj._transaction).addCallback(lambda _: obj)
+ else:
+ return Registry.DBPOOL.runInteraction(_doupdate).addCallback(lambda _: obj)
def refreshObj(self, obj):
@@ -371,7 +424,7 @@ def updateArgsToString(self, args):
"""
Convert dictionary of arguments to form needed for DB update query. This method will
vary by database driver.
-
+
@param args: Values to insert. Should be a dictionary in the form of
C{{'name': value, 'othername': value}}.
@@ -382,7 +435,7 @@ def updateArgsToString(self, args):
return (setstring, args.values())
- def count(self, tablename, where=None):
+ def count(self, tablename, where=None, transaction=None):
"""
Get the number of rows in the given table (optionally, that meet the given where criteria).
@@ -392,6 +445,49 @@ def count(self, tablename, where=None):
@return: A C{Deferred} that returns the number of rows.
"""
- d = self.select(tablename, where=where, select='count(*)')
- d.addCallback(lambda res: res[0]['count(*)'])
+ q, args = self._build_select(tablename, where=where, select='count(*)')
+ if transaction is not None:
+ d = Registry.DBPOOL.runQueryInTransaction(transaction, q, args)
+ else:
+ d = self.execute(q, args)
+
+ def _parse_count_result(result):
+ val = result[0]
+ return val[0]
+
+ d.addCallback(_parse_count_result)
return d
+
+
+ def commit(self, transaction):
+ """
+ Commits current obj transaction.
+
+ @return: A C{Deferred}
+ """
+ return Registry.DBPOOL.commitTransaction(transaction)
+
+
+ def rollback(self, transaction):
+ """
+ Rollback current obj transaction.
+
+ @return: A C{Deferred}
+ """
+ return Registry.DBPOOL.rollbackTransaction(transaction)
+
+
+ def _rollback(self, transaction):
+ trans = transaction
+ conn = trans._connection
+
+ if trans._cursor is None:
+ raise TransactionNotStartedError("Cannot call rollback without a transaction")
+
+ try:
+ trans.close()
+ conn.rollback()
+ except:
+ excType, excValue, excTraceback = sys.exc_info()
+ raise excType, excValue, excTraceback
+
diff --git a/twistar/dbconfig/postgres.py b/twistar/dbconfig/postgres.py
index 1b7e13b..2748162 100644
--- a/twistar/dbconfig/postgres.py
+++ b/twistar/dbconfig/postgres.py
@@ -4,10 +4,10 @@
class PostgreSQLDBConfig(InteractionBase):
includeBlankInInsert = False
- def getLastInsertID(self, txn):
+ def getLastInsertID(self, transaction):
q = "SELECT lastval()"
- self.executeTxn(txn, q)
- result = txn.fetchall()
+ self.executeTxn(transaction, q)
+ result = transaction.fetchall()
return result[0][0]
@@ -20,8 +20,3 @@ def insertArgsToString(self, vals):
def escapeColNames(self, colnames):
return map(lambda x: '"%s"' % x, colnames)
-
- def count(self, tablename, where=None):
- d = self.select(tablename, where=where, select='count(*)')
- d.addCallback(lambda res: res[0]['count'])
- return d
diff --git a/twistar/dbconfig/sqlite.py b/twistar/dbconfig/sqlite.py
index 087c916..089474a 100644
--- a/twistar/dbconfig/sqlite.py
+++ b/twistar/dbconfig/sqlite.py
@@ -10,10 +10,10 @@ def whereToString(self, where):
return (query, args)
- def getLastInsertID(self, txn):
+ def getLastInsertID(self, transaction):
q = "SELECT last_insert_rowid()"
- self.executeTxn(txn, q)
- result = txn.fetchall()
+ self.executeTxn(transaction, q)
+ result = transaction.fetchall()
return result[0][0]
@@ -28,12 +28,14 @@ def insertArgsToString(self, vals):
## retarded sqlite can't handle multiple row inserts
- def insertMany(self, tablename, vals):
- def _insertMany(txn):
+ def insertMany(self, tablename, vals, transaction=None):
+ def _insertMany(transaction):
for val in vals:
- self.insert(tablename, val, txn)
- return Registry.DBPOOL.runInteraction(_insertMany)
-
+ self.insert(tablename, val, transaction)
+ if transaction:
+ return _insertMany(transaction)
+ else:
+ return Registry.DBPOOL.runInteraction(_insertMany)
diff --git a/twistar/dbobject.py b/twistar/dbobject.py
index 1f2749d..6b26066 100644
--- a/twistar/dbobject.py
+++ b/twistar/dbobject.py
@@ -8,6 +8,7 @@
from twistar.registry import Registry
from twistar.relationships import Relationship
from twistar.exceptions import InvalidRelationshipError, DBObjectSaveError, ReferenceNotSavedError
+from twistar.exceptions import TransactionNotStartedError, TransactionAlreadyStartedError
from twistar.utils import createInstances, deferredDict, dictToWhere
from twistar.validation import Validator, Errors
@@ -58,10 +59,16 @@ class DBObject(Validator):
# it will be of the form {'othername': , 'anothername': }
RELATIONSHIP_CACHE = None
- def __init__(self, **kwargs):
+ # this will hold an optional t.e.a.Transaction instance that can be used to put many
+ # ORM operation into a single transaction.
+ _transaction = None
+
+ def __init__(self, transaction=None, **kwargs):
"""
Constructor. DO NOT OVERWRITE. Use the L{DBObject.afterInit} method.
-
+
+ @param transaction: An optional t.e.a.Transaction object
+
@param kwargs: An optional dictionary containing the properties that
should be initially set for this object.
@@ -70,6 +77,7 @@ def __init__(self, **kwargs):
self.id = None
self._deleted = False
self.errors = Errors()
+ self._transaction = transaction
self.updateAttrs(kwargs)
self._config = Registry.getConfig()
@@ -160,8 +168,9 @@ def beforeSave(self):
def afterInit(self):
"""
- Method called when a new L{DBObject} is instantiated. Classes can overwrite this method.
- This method may return a C{Deferred}.
+ Method called when a new L{DBObject} is instantiated as a result of DB queries. If you
+ create an instance of this class on your own, you will need to call the method yourself.
+ Classes can overwrite this method. This method may return a C{Deferred}.
"""
@@ -266,7 +275,10 @@ def _delete(result):
oldid = self.id
self.id = None
self._deleted = True
- return self.__class__.deleteAll(where=["id = ?", oldid])
+ if self._transaction:
+ return self.__class__.deleteAll(where=["id = ?", oldid], transaction=self._transaction)
+ else:
+ return self.__class__.deleteAll(where=["id = ?", oldid])
def _deleteOnSuccess(result):
if result == False:
@@ -275,13 +287,13 @@ def _deleteOnSuccess(result):
ds = []
for relation in self.HABTM:
name = relation['name'] if isinstance(relation, dict) else relation
- ds.append(getattr(self, name).clear())
+ ds.append(getattr(self, name).clear(transaction=self._transaction))
return defer.DeferredList(ds).addCallback(_delete)
return defer.maybeDeferred(self.beforeDelete).addCallback(_deleteOnSuccess)
- def loadRelations(self, *relations):
+ def loadRelations(self, *relations, **kwargs):
"""
Preload a a list of relationships. For instance, if you have an instance of an
object C{User} (named C{user}) that has many C{Address}es and has one C{Avatar},
@@ -296,16 +308,18 @@ def loadRelations(self, *relations):
@return: A C{Deferred}.
"""
+ transaction = kwargs.get('transaction', None)
+
if len(relations) == 0:
klass = object.__getattribute__(self, "__class__")
allrelations = klass.RELATIONSHIP_CACHE.keys()
if len(allrelations) == 0:
return defer.succeed({})
- return self.loadRelations(*allrelations)
+ return self.loadRelations(*allrelations, transaction=transaction)
ds = {}
for relation in relations:
- ds[relation] = getattr(self, relation).get()
+ ds[relation] = getattr(self, relation).get(transaction=transaction)
return deferredDict(ds)
@@ -388,12 +402,16 @@ def findBy(klass, **attrs):
Will return all matches.
"""
+ transaction = None
+ if 'transaction' in attrs:
+ transaction = attrs['transaction']
+ del(attrs['transaction'])
where = dictToWhere(attrs)
- return klass.find(where = where)
+ return klass.find(where = where, transaction=transaction)
@classmethod
- def find(klass, id=None, where=None, group=None, limit=None, orderby=None):
+ def find(klass, id=None, where=None, group=None, limit=None, orderby=None, transaction=None):
"""
Find instances of a given class.
@@ -413,6 +431,8 @@ def find(klass, id=None, where=None, group=None, limit=None, orderby=None):
@param orderby: A C{str} describing the ordering, like C{orderby='first_name DESC'}.
+ @param orderby: A C{t.e.a.Transaction} to use for this query.
+
@return: A C{Deferred} which returns the following to a callback:
If id is specified (or C{limit} is 1) then a single
instance of C{klass} will be returned if one is found that fits the criteria, C{None}
@@ -420,12 +440,12 @@ def find(klass, id=None, where=None, group=None, limit=None, orderby=None):
be returned with all matching results.
"""
config = Registry.getConfig()
- d = config.select(klass.tablename(), id, where, group, limit, orderby)
+ d = config.select(klass.tablename(), id, where, group, limit, orderby, transaction=transaction)
return d.addCallback(createInstances, klass)
@classmethod
- def count(klass, where=None):
+ def count(klass, where=None, transaction=None):
"""
Count instances of a given class.
@@ -437,7 +457,7 @@ def count(klass, where=None):
@return: A C{Deferred} which returns the total number of db records to a callback.
"""
config = Registry.getConfig()
- return config.count(klass.tablename(), where=where)
+ return config.count(klass.tablename(), where=where, transaction=transaction)
@classmethod
@@ -453,7 +473,7 @@ def all(klass):
@classmethod
- def deleteAll(klass, where=None):
+ def deleteAll(klass, where=None, transaction=None):
"""
Delete all instances of C{klass} in the database.
@@ -464,7 +484,7 @@ def deleteAll(klass, where=None):
"""
config = Registry.getConfig()
tablename = klass.tablename()
- return config.delete(tablename, where)
+ return config.delete(tablename, where, transaction)
@classmethod
@@ -484,6 +504,71 @@ def _exists(result):
return klass.find(where=where, limit=1).addCallback(_exists)
+ def transaction(self, transaction=None):
+ """
+ Read current database transaction. If already set, returns the one active.
+
+ @return: A C{dict} containing a {t.e.a.Connection} and C{t.e.a.Transaction}
+ """
+ if self._transaction is None:
+ if transaction:
+ self._transaction = transaction
+ return self._transaction
+ else:
+ raise TransactionNotStartedError("Transaction not yet started!")
+ else:
+ if transaction and transaction != self._transaction:
+ raise TransactionAlreadyStartedError("Transaction already started, cannot set!")
+ return self._transaction
+
+
+ def startTransaction(self):
+ """
+ Init a new database transaction. If already set, raises {TransactionAlreadyStartedError}
+
+ @return: A C{dict} containing a {t.e.a.Connection} and C{t.e.a.Transaction}
+ """
+ if self._transaction is None:
+ d = self._config.startTxn()
+
+ def processTxn(result):
+ self._transaction = result
+ return self._transaction
+
+ d.addCallback(processTxn)
+ return d
+ else:
+ raise TransactionAlreadyStartedError("Transaction already started. Call commit or rollback to close it")
+
+
+ def rollback(self):
+ """
+ Rollback current object transaction(s). Clean up transaction once finished.
+ """
+ if self._transaction is None:
+ raise TransactionNotStartedError("Cannot call rollback without a transaction")
+ else:
+ def _resetTxn(result):
+ self._transaction = None
+ d = self._config.rollback(self._transaction)
+ d.addCallback(_resetTxn)
+ return d
+
+
+ def commit(self):
+ """
+ Commits current object transaction(s). Clean up transaction once finished.
+ """
+ if self._transaction is None:
+ raise TransactionNotStartedError("Cannot call commit without a transaction")
+ else:
+ def _resetTxn(result):
+ self._transaction = None
+ d = self._config.commit(self._transaction)
+ d.addCallback(_resetTxn)
+ return d
+
+
def __str__(self):
"""
Get the string version of this object.
@@ -541,6 +626,10 @@ def __neq__(self, other):
return not self == other
+ def __hash__(self):
+ return hash('%s.%d' % (type(self).__name__, self.id))
+
+
__repr__ = __str__
diff --git a/twistar/exceptions.py b/twistar/exceptions.py
index 6df69a4..68be11a 100644
--- a/twistar/exceptions.py
+++ b/twistar/exceptions.py
@@ -40,3 +40,16 @@ class DBObjectSaveError(Exception):
"""
Error saving a DBObject.
"""
+
+
+class TransactionNotStartedError(Exception):
+ """
+ Error resulting from the attempt of using a method which needs a transaction started
+ """
+
+
+class TransactionAlreadyStartedError(Exception):
+ """
+ Error resulting from the attempt of starting another transaction on same object
+ """
+
diff --git a/twistar/relationships.py b/twistar/relationships.py
index f2d0aeb..080b890 100644
--- a/twistar/relationships.py
+++ b/twistar/relationships.py
@@ -11,14 +11,14 @@
from twistar.exceptions import ReferenceNotSavedError
-class Relationship:
+class Relationship(object):
"""
Base class that all specific relationship type classes extend.
@see: L{HABTM}, L{HasOne}, L{HasMany}, L{BelongsTo}
"""
- def __init__(self, inst, propname, givenargs):
+ def __init__(self, inst, propname, givenargs, singular=False):
"""
Constructor.
@@ -38,28 +38,47 @@ def __init__(self, inst, propname, givenargs):
self.dbconfig = Registry.getConfig()
## Set args
+ if singular:
+ association_foreign_key = self.infl.foreignKey(propname)
+ else:
+ association_foreign_key = self.infl.foreignKey(self.infl.singularize(propname))
+
self.args = {
'class_name': propname,
- 'association_foreign_key': self.infl.foreignKey(self.infl.singularize(propname)),
+ 'association_foreign_key': association_foreign_key,
'foreign_key': self.infl.foreignKey(self.inst.__class__.__name__),
'polymorphic': False
}
self.args.update(givenargs)
- otherklassname = self.infl.classify(self.args['class_name'])
+ if singular:
+ otherklassname = self.infl.camelize(self.args['class_name'])
+ else:
+ otherklassname = self.infl.classify(self.args['class_name'])
+
if not self.args['polymorphic']:
self.otherklass = Registry.getClass(otherklassname)
+
self.othername = self.args['association_foreign_key']
self.thisclass = self.inst.__class__
self.thisname = self.args['foreign_key']
+ def _updateInTransaction(self, transaction, tablename, args, where):
+ return self.dbconfig.update(tablename, args, where, transaction)
+
+
+
class BelongsTo(Relationship):
"""
Class representing a belongs-to relationship.
"""
- def get(self):
+ def __init__(self, *args, **kwargs):
+ kwargs['singular'] = True
+ super(BelongsTo, self).__init__(*args, **kwargs)
+
+ def get(self, transaction=None):
"""
Get the object that belong to the caller.
@@ -69,15 +88,15 @@ def get(self):
def get_polymorphic(row):
kid = getattr(row, "%s_id" % self.args['class_name'])
kname = getattr(row, "%s_type" % self.args['class_name'])
- return Registry.getClass(kname).find(kid)
+ return Registry.getClass(kname).find(kid, transaction=transaction)
if self.args['polymorphic']:
- return self.inst.find(where=["id = ?", self.inst.id], limit=1).addCallback(get_polymorphic)
+ return self.inst.find(where=["id = ?", self.inst.id], limit=1, transaction=transaction).addCallback(get_polymorphic)
- return self.otherklass.find(where=["id = ?", getattr(self.inst, self.othername)], limit=1)
+ return self.otherklass.find(where=["id = ?", getattr(self.inst, self.othername)], limit=1, transaction=transaction)
- def set(self, other):
+ def set(self, other, transaction=None):
"""
Set the object that belongs to the caller.
@@ -86,16 +105,20 @@ def set(self, other):
if self.args['polymorphic']:
setattr(self.inst, "%s_type" % self.args['class_name'], other.__class__.__name__)
setattr(self.inst, self.othername, other.id)
+ if transaction:
+ self.inst.transaction(transaction)
return self.inst.save()
- def clear(self):
+ def clear(self, transaction=None):
"""
Remove the relationship linking the object that belongs to the caller.
@return: A C{Deferred} with a callback value of the caller.
"""
setattr(self.inst, self.othername, None)
+ if transaction:
+ self.inst.transaction(transaction)
return self.inst.save()
@@ -140,7 +163,7 @@ def _generateGetArgs(self, kwargs):
else:
where = ["%s = ?" % self.thisname, self.inst.id]
- if kwargs.has_key('where'):
+ if kwargs.has_key('where') and kwargs['where']:
kwargs['where'] = joinWheres(where, kwargs['where'])
else:
kwargs['where'] = where
@@ -160,7 +183,7 @@ def _set_polymorphic(self, others):
return defer.DeferredList(ds)
- def _update(self, _, others):
+ def _update(self, _, others, transaction):
tablename = self.otherklass.tablename()
args = {self.thisname: self.inst.id}
ids = []
@@ -170,10 +193,13 @@ def _update(self, _, others):
raise ReferenceNotSavedError, msg
ids.append(str(other.id))
where = ["id IN (%s)" % ",".join(ids)]
+ if transaction:
+ return Registry.DBPOOL.executeOperationInTransaction(self._updateInTransaction,
+ transaction, tablename, args, where)
return self.dbconfig.update(tablename, args, where)
- def set(self, others):
+ def set(self, others, transaction=None):
"""
Set the objects that caller has.
@@ -184,18 +210,22 @@ def set(self, others):
tablename = self.otherklass.tablename()
args = {self.thisname: None}
- where = ["%s = ?" % self.thisname, self.inst.id]
- d = self.dbconfig.update(tablename, args, where)
+ where = ["%s = ?" % self.thisname, self.inst.id]
+ if transaction:
+ d = Registry.DBPOOL.executeOperationInTransaction(self._updateInTransaction,
+ transaction, tablename, args, where)
+ else:
+ d = self.dbconfig.update(tablename, args, where)
if len(others) > 0:
- d.addCallback(self._update, others)
+ d.addCallback(self._update, others, transaction=transaction)
return d
- def clear(self):
+ def clear(self, transaction=None):
"""
Clear the list of all of the objects that this one has.
"""
- return self.set([])
+ return self.set([], transaction)
class HasOne(Relationship):
@@ -203,16 +233,21 @@ class HasOne(Relationship):
A class representing the has one relationship.
"""
- def get(self):
+ def __init__(self, *args, **kwargs):
+ kwargs['singular'] = True
+ super(HasOne, self).__init__(*args, **kwargs)
+
+ def get(self, transaction=None):
"""
Get the object that caller has.
@return: A C{Deferred} with a callback value of the object this one has (or c{None}).
"""
- return self.otherklass.find(where=["%s = ?" % self.thisname, self.inst.id], limit=1)
+ return self.otherklass.find(where=["%s = ?" % self.thisname, self.inst.id], limit=1,
+ transaction=transaction)
- def set(self, other):
+ def set(self, other, transaction=None):
"""
Set the object that caller has.
@@ -221,6 +256,9 @@ def set(self, other):
tablename = self.otherklass.tablename()
args = {self.thisname: self.inst.id}
where = ["id = ?", other.id]
+ if transaction:
+ return Registry.DBPOOL.executeOperationInTransaction(self._updateInTransaction,
+ transaction, tablename, args, where)
return self.dbconfig.update(tablename, args, where)
@@ -258,7 +296,8 @@ def get(self, **kwargs):
@param kwargs: These could include C{limit}, C{orderby}, or any others included in
C{InteractionBase.select}. If a C{where} parameter is included, the conditions will
- be added to the ones already imposed by default in this method.
+ be added to the ones already imposed by default in this method. The argument
+ C{join_where} will be applied to the join table, if provided.
@return: A C{Deferred} with a callback value of a list of objects.
"""
@@ -267,7 +306,7 @@ def _get(rows):
return defer.succeed([])
ids = [str(row[self.othername]) for row in rows]
where = ["id IN (%s)" % ",".join(ids)]
- if kwargs.has_key('where'):
+ if kwargs.has_key('where') and kwargs['where']:
kwargs['where'] = joinWheres(where, kwargs['where'])
else:
kwargs['where'] = where
@@ -276,6 +315,11 @@ def _get(rows):
tablename = self.tablename()
where = ["%s = ?" % self.thisname, self.inst.id]
+ if kwargs.has_key('join_where'):
+ where = joinWheres(where, kwargs.pop('join_where'))
+ if 'transaction' in kwargs:
+ return self.dbconfig.select(tablename, where=where,
+ transaction=kwargs['transaction']).addCallback(_get)
return self.dbconfig.select(tablename, where=where).addCallback(_get)
@@ -292,7 +336,7 @@ def count(self, **kwargs):
def _get(rows):
if len(rows) == 0:
return defer.succeed(0)
- if not kwargs.has_key('where'):
+ if not kwargs.has_key('where') or not kwargs['where']:
return defer.succeed(len(rows))
ids = [str(row[self.othername]) for row in rows]
where = ["id IN (%s)" % ",".join(ids)]
@@ -302,37 +346,45 @@ def _get(rows):
tablename = self.tablename()
where = ["%s = ?" % self.thisname, self.inst.id]
+ if 'transaction' in kwargs:
+ return self.dbconfig.select(tablename, where=where,
+ transaction=kwargs['transaction']).addCallback(_get)
return self.dbconfig.select(tablename, where=where).addCallback(_get)
- def _set(self, _, others):
+ def _set(self, _, others, transaction=None):
args = []
for other in others:
if other.id is None:
msg = "You must save all other instances before defining a relationship"
raise ReferenceNotSavedError, msg
args.append({self.thisname: self.inst.id, self.othername: other.id})
- return self.dbconfig.insertMany(self.tablename(), args)
-
-
- def set(self, others):
+ if transaction:
+ def _manyInTransaction(transaction, tablename, args):
+ return self.dbconfig.insertMany(tablename, args, transaction)
+ return Registry.DBPOOL.executeOperationInTransaction(_manyInTransaction,
+ transaction, self.tablename(), args)
+ return self.dbconfig.insertMany( self.tablename(), args)
+
+
+ def set(self, others, transaction=None):
"""
Set the objects that caller has.
@return: A C{Deferred}.
"""
where = ["%s = ?" % self.thisname, self.inst.id]
- d = self.dbconfig.delete(self.tablename(), where=where)
+ d = self.dbconfig.delete(self.tablename(), where=where, transaction=transaction)
if len(others) > 0:
- d.addCallback(self._set, others)
+ d.addCallback(self._set, others, transaction=transaction)
return d
- def clear(self):
+ def clear(self, transaction=None):
"""
Clear the list of all of the objects that this one has.
"""
- return self.set([])
+ return self.set([], transaction=transaction)
Relationship.TYPES = {'HASMANY': HasMany, 'HASONE': HasOne, 'BELONGSTO': BelongsTo, 'HABTM': HABTM}
diff --git a/twistar/tests/mysql_config.py b/twistar/tests/mysql_config.py
index b00a7cc..c309a47 100644
--- a/twistar/tests/mysql_config.py
+++ b/twistar/tests/mysql_config.py
@@ -3,30 +3,48 @@
from twistar.registry import Registry
-CONNECTION = Registry.DBPOOL = adbapi.ConnectionPool('MySQLdb', user="", passwd="", host="localhost", db="twistar")
+from txconnectionpool.txconnectionpool import TxConnectionPool
+CONNECTION = Registry.DBPOOL = TxConnectionPool('MySQLdb', user="root", passwd="", host="127.0.0.1", db="twistar")
def initDB(testKlass):
def runInitTxn(txn):
txn.execute("""CREATE TABLE users (id INT AUTO_INCREMENT,
- first_name VARCHAR(255), last_name VARCHAR(255), age INT, dob DATE, PRIMARY KEY (id))""")
+ first_name VARCHAR(255), last_name VARCHAR(255), age INT, dob DATE, PRIMARY KEY (id)) ENGINE=INNODB""")
txn.execute("""CREATE TABLE avatars (id INT AUTO_INCREMENT, name VARCHAR(255),
- color VARCHAR(255), user_id INT, PRIMARY KEY (id))""")
+ color VARCHAR(255), user_id INT, PRIMARY KEY (id)) ENGINE=INNODB""")
txn.execute("""CREATE TABLE pictures (id INT AUTO_INCREMENT, name VARCHAR(255),
- size INT, user_id INT, PRIMARY KEY (id))""")
- txn.execute("""CREATE TABLE favorite_colors (id INT AUTO_INCREMENT, name VARCHAR(255), PRIMARY KEY (id))""")
- txn.execute("""CREATE TABLE favorite_colors_users (favorite_color_id INT, user_id INT)""")
- txn.execute("""CREATE TABLE coltests (id INT AUTO_INCREMENT, `select` VARCHAR(255), `where` VARCHAR(255), PRIMARY KEY (id))""")
+ size INT, user_id INT, PRIMARY KEY (id)) ENGINE=INNODB""")
+ txn.execute("""CREATE TABLE comments (id INT AUTO_INCREMENT, subject VARCHAR(255),
+ body TEXT, user_id INT, PRIMARY KEY (id)) ENGINE=INNODB""")
+ txn.execute("""CREATE TABLE favorite_colors (id INT AUTO_INCREMENT, name VARCHAR(255), PRIMARY KEY (id)) ENGINE=INNODB""")
+ txn.execute("""CREATE TABLE favorite_colors_users (favorite_color_id INT, user_id INT, palette_id INT) ENGINE=INNODB""")
+ txn.execute("""CREATE TABLE coltests (id INT AUTO_INCREMENT, `select` VARCHAR(255), `where` VARCHAR(255), PRIMARY KEY (id)) ENGINE=INNODB""")
- txn.execute("""CREATE TABLE boys (id INT AUTO_INCREMENT, `name` VARCHAR(255), PRIMARY KEY (id))""")
- txn.execute("""CREATE TABLE girls (id INT AUTO_INCREMENT, `name` VARCHAR(255), PRIMARY KEY (id))""")
+ txn.execute("""CREATE TABLE boys (id INT AUTO_INCREMENT, `name` VARCHAR(255), PRIMARY KEY (id)) ENGINE=INNODB""")
+ txn.execute("""CREATE TABLE girls (id INT AUTO_INCREMENT, `name` VARCHAR(255), PRIMARY KEY (id)) ENGINE=INNODB""")
txn.execute("""CREATE TABLE nicknames (id INT AUTO_INCREMENT, `value` VARCHAR(255), `nicknameable_id` INT,
- `nicknameable_type` VARCHAR(255), PRIMARY KEY(id))""")
+ `nicknameable_type` VARCHAR(255), PRIMARY KEY(id)) ENGINE=INNODB""")
txn.execute("""CREATE TABLE blogposts (id INT AUTO_INCREMENT,
title VARCHAR(255), text VARCHAR(255), PRIMARY KEY (id))""")
txn.execute("""CREATE TABLE categories (id INT AUTO_INCREMENT,
name VARCHAR(255), PRIMARY KEY (id))""")
txn.execute("""CREATE TABLE posts_categories (category_id INT, blogpost_id INT)""")
+ txn.execute("""CREATE TABLE pens (id INT AUTO_INCREMENT,
+ color VARCHAR(255), len INT, PRIMARY KEY (id), UNIQUE(color)) ENGINE=INNODB""")
+ txn.execute("""CREATE TABLE tables (id INT AUTO_INCREMENT,
+ color VARCHAR(255), weight INT, PRIMARY KEY (id), UNIQUE(color)) ENGINE=INNODB""")
+ txn.execute("""CREATE TABLE pens_tables (pen_id INT, table_id INT) ENGINE=INNODB""")
+ txn.execute("""CREATE TABLE rubbers (id INT AUTO_INCREMENT,
+ color VARCHAR(255), table_id INT, PRIMARY KEY (id)) ENGINE=INNODB""")
+
+ txn.execute("""CREATE TABLE roles (id INT AUTO_INCREMENT,
+ description VARCHAR(255), serviceclass_id INT, PRIMARY KEY (id)) ENGINE=INNODB""")
+ txn.execute("""CREATE TABLE serviceclasses (id INT AUTO_INCREMENT,
+ description VARCHAR(255), superclass_id INT, PRIMARY KEY (id)) ENGINE=INNODB""")
+ txn.execute("""CREATE TABLE superclasses (id INT AUTO_INCREMENT,
+ description VARCHAR(255), PRIMARY KEY (id)) ENGINE=INNODB""")
+
return CONNECTION.runInteraction(runInitTxn)
@@ -35,6 +53,7 @@ def runTearDownDB(txn):
txn.execute("DROP TABLE users")
txn.execute("DROP TABLE avatars")
txn.execute("DROP TABLE pictures")
+ txn.execute("DROP TABLE comments")
txn.execute("DROP TABLE favorite_colors")
txn.execute("DROP TABLE favorite_colors_users")
txn.execute("DROP TABLE coltests")
@@ -44,5 +63,12 @@ def runTearDownDB(txn):
txn.execute("DROP TABLE blogposts")
txn.execute("DROP TABLE categories")
txn.execute("DROP TABLE posts_categories")
+ txn.execute("DROP TABLE pens")
+ txn.execute("DROP TABLE tables")
+ txn.execute("DROP TABLE rubbers")
+ txn.execute("DROP TABLE pens_tables")
+ txn.execute("DROP TABLE roles")
+ txn.execute("DROP TABLE serviceclasses")
+ txn.execute("DROP TABLE superclasses")
return CONNECTION.runInteraction(runTearDownDB)
diff --git a/twistar/tests/postgres_config.py b/twistar/tests/postgres_config.py
index fe53a9f..d8c0c84 100644
--- a/twistar/tests/postgres_config.py
+++ b/twistar/tests/postgres_config.py
@@ -3,7 +3,9 @@
from twistar.registry import Registry
-CONNECTION = Registry.DBPOOL = adbapi.ConnectionPool('psycopg2', "dbname=twistar")
+from txconnectionpool.txconnectionpool import TxConnectionPool
+
+CONNECTION = Registry.DBPOOL = TxConnectionPool('psycopg2', "dbname=twistar")
def initDB(testKlass):
def runInitTxn(txn):
@@ -13,8 +15,10 @@ def runInitTxn(txn):
color VARCHAR(255), user_id INT)""")
txn.execute("""CREATE TABLE pictures (id SERIAL PRIMARY KEY, name VARCHAR(255),
size INT, user_id INT)""")
+ txn.execute("""CREATE TABLE comments (id SERIAL PRIMARY KEY, subject VARCHAR(255),
+ body TEXT, user_id INT)""")
txn.execute("""CREATE TABLE favorite_colors (id SERIAL PRIMARY KEY, name VARCHAR(255))""")
- txn.execute("""CREATE TABLE favorite_colors_users (favorite_color_id INT, user_id INT)""")
+ txn.execute("""CREATE TABLE favorite_colors_users (favorite_color_id INT, user_id INT, palette_id INT)""")
txn.execute("""CREATE TABLE coltests (id SERIAL PRIMARY KEY, "select" VARCHAR(255), "where" VARCHAR(255))""")
txn.execute("""CREATE TABLE boys (id SERIAL PRIMARY KEY, "name" VARCHAR(255))""")
@@ -27,6 +31,15 @@ def runInitTxn(txn):
name VARCHAR(255))""")
txn.execute("""CREATE TABLE posts_categories (category_id INT, blogpost_id INT)""")
+ txn.execute("""CREATE TABLE pens (id SERIAL PRIMARY KEY, color VARCHAR(255) UNIQUE, len INT)""");
+ txn.execute("""CREATE TABLE tables (id SERIAL PRIMARY KEY, color VARCHAR(255) UNIQUE, weight INT)""");
+ txn.execute("""CREATE TABLE pens_tables (pen_id INT, table_id INT)""")
+ txn.execute("""CREATE TABLE rubbers (id SERIAL PRIMARY KEY, color VARCHAR(255), table_id INT)""");
+
+ txn.execute("""CREATE TABLE roles (id SERIAL PRIMARY KEY, description VARCHAR(255), serviceclass_id INT)""");
+ txn.execute("""CREATE TABLE serviceclasses (id SERIAL PRIMARY KEY, description VARCHAR(255), superclass_id INT)""");
+ txn.execute("""CREATE TABLE superclasses (id SERIAL PRIMARY KEY, description VARCHAR(255))""");
+
return CONNECTION.runInteraction(runInitTxn)
@@ -41,6 +54,9 @@ def runTearDownDB(txn):
txn.execute("DROP SEQUENCE pictures_id_seq CASCADE")
txn.execute("DROP TABLE pictures")
+ txn.execute("DROP SEQUENCE comments_id_seq CASCADE")
+ txn.execute("DROP TABLE comments")
+
txn.execute("DROP SEQUENCE favorite_colors_id_seq CASCADE")
txn.execute("DROP TABLE favorite_colors")
@@ -66,5 +82,23 @@ def runTearDownDB(txn):
txn.execute("DROP TABLE posts_categories")
+ txn.execute("DROP SEQUENCE pens_id_seq CASCADE")
+ txn.execute("DROP TABLE pens")
+
+ txn.execute("DROP SEQUENCE tables_id_seq CASCADE")
+ txn.execute("DROP TABLE tables")
+
+ txn.execute("DROP SEQUENCE rubbers_id_seq CASCADE")
+ txn.execute("DROP TABLE rubbers")
+
+ txn.execute("DROP TABLE pens_tables")
+
+ txn.execute("DROP SEQUENCE roles_id_seq CASCADE")
+ txn.execute("DROP TABLE roles")
+ txn.execute("DROP SEQUENCE serviceclasses_id_seq CASCADE")
+ txn.execute("DROP TABLE serviceclasses")
+ txn.execute("DROP SEQUENCE superclasses_id_seq CASCADE")
+ txn.execute("DROP TABLE superclasses")
+
return CONNECTION.runInteraction(runTearDownDB)
diff --git a/twistar/tests/sqlite_config.py b/twistar/tests/sqlite_config.py
index 89a07b1..f7dd4ce 100644
--- a/twistar/tests/sqlite_config.py
+++ b/twistar/tests/sqlite_config.py
@@ -1,11 +1,12 @@
-from twisted.enterprise import adbapi
from twisted.internet import defer
from twistar.registry import Registry
+from txconnectionpool.txconnectionpool import TxConnectionPool
+
def initDB(testKlass):
location = testKlass.mktemp()
- Registry.DBPOOL = adbapi.ConnectionPool('sqlite3', location, check_same_thread=False)
+ Registry.DBPOOL = TxConnectionPool('sqlite3', location, check_same_thread=False)
def runInitTxn(txn):
txn.execute("""CREATE TABLE users (id INTEGER PRIMARY KEY AUTOINCREMENT,
first_name TEXT, last_name TEXT, age INTEGER, dob DATE)""")
@@ -13,8 +14,10 @@ def runInitTxn(txn):
color TEXT, user_id INTEGER)""")
txn.execute("""CREATE TABLE pictures (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT,
size INTEGER, user_id INTEGER)""")
+ txn.execute("""CREATE TABLE comments (id INTEGER PRIMARY KEY AUTOINCREMENT, subject TEXT,
+ body TEXT, user_id INTEGER)""")
txn.execute("""CREATE TABLE favorite_colors (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT)""")
- txn.execute("""CREATE TABLE favorite_colors_users (favorite_color_id INTEGER, user_id INTEGER)""")
+ txn.execute("""CREATE TABLE favorite_colors_users (favorite_color_id INTEGER, user_id INTEGER, palette_id INTEGER)""")
txn.execute("""CREATE TABLE coltests (id INTEGER PRIMARY KEY AUTOINCREMENT, `select` TEXT, `where` TEXT)""")
txn.execute("""CREATE TABLE boys (id INTEGER PRIMARY KEY AUTOINCREMENT, `name` TEXT)""")
@@ -26,7 +29,16 @@ def runInitTxn(txn):
txn.execute("""CREATE TABLE categories (id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT)""")
txn.execute("""CREATE TABLE posts_categories (category_id INTEGER, blogpost_id INTEGER)""")
+
+ txn.execute("""CREATE TABLE pens (id INTEGER PRIMARY KEY AUTOINCREMENT, color TEXT UNIQUE, len INTEGER)""");
+ txn.execute("""CREATE TABLE tables (id INTEGER PRIMARY KEY AUTOINCREMENT, color TEXT UNIQUE, weight INTEGER)""");
+ txn.execute("""CREATE TABLE pens_tables (pen_id INT, table_id INT)""")
+ txn.execute("""CREATE TABLE rubbers (id INTEGER PRIMARY KEY AUTOINCREMENT, color TEXT, table_id INTEGER)""")
+ txn.execute("""CREATE TABLE roles (id INTEGER PRIMARY KEY AUTOINCREMENT, serviceclass_id INTEGER, description TEXT)""")
+ txn.execute("""CREATE TABLE serviceclasses (id INTEGER PRIMARY KEY AUTOINCREMENT, superclass_id INTEGER, description TEXT)""")
+ txn.execute("""CREATE TABLE superclasses (id INTEGER PRIMARY KEY AUTOINCREMENT, description TEXT)""")
+
return Registry.DBPOOL.runInteraction(runInitTxn)
diff --git a/twistar/tests/test_dbobject.py b/twistar/tests/test_dbobject.py
index 3e7ebdb..4e56ea1 100644
--- a/twistar/tests/test_dbobject.py
+++ b/twistar/tests/test_dbobject.py
@@ -22,6 +22,15 @@ def tearDown(self):
yield tearDownDB(self)
+ @inlineCallbacks
+ def test_count_save(self):
+ count = yield User.count()
+ self.assertEqual(count, 1)
+ self.user.first_name = "something else"
+ user = yield self.user.save()
+ self.assertEqual(user.first_name, "something else")
+
+
@inlineCallbacks
def test_findBy(self):
r = yield User.findBy(first_name="Non", last_name="Existant")
@@ -126,6 +135,7 @@ def test_delete(self):
self.assertEqual(result, None)
+ @inlineCallbacks
def test_delete_all(self):
users = yield User.all()
ids = [user.id for user in users]
@@ -255,3 +265,17 @@ def test_loadRelations(self):
suball = yield user.loadRelations('pictures')
self.assertTrue(not suball.has_key('avatar'))
self.assertEqual(pictures, suball['pictures'])
+
+
+ @inlineCallbacks
+ def test_loadRelations_plural(self):
+ superclass = yield Superclass().save()
+ serviceclass = yield Serviceclass(superclass_id=superclass.id).save()
+
+ superclass = yield Superclass.find(limit=1)
+ all = yield superclass.loadRelations()
+
+ serviceclass = yield superclass.serviceclass.get()
+ self.assertTrue(all.has_key('serviceclass'))
+ self.assertEqual(serviceclass, all['serviceclass'])
+
diff --git a/twistar/tests/test_relationships.py b/twistar/tests/test_relationships.py
index 2ba9f43..a883e8d 100644
--- a/twistar/tests/test_relationships.py
+++ b/twistar/tests/test_relationships.py
@@ -128,6 +128,19 @@ def test_has_many_count(self):
self.assertEqual(totalnum, 4)
+ @inlineCallbacks
+ def test_has_many_count_nocache(self):
+ # First, count comments
+ totalnum = yield self.user.comments.count()
+ self.assertEqual(totalnum, 0)
+
+ for _ in range(3):
+ pic = yield Comment(user_id=self.user.id).save()
+
+ totalnum = yield self.user.comments.count()
+ self.assertEqual(totalnum, 3)
+
+
@inlineCallbacks
def test_has_many_get_with_args(self):
# First, make a few pics
@@ -141,6 +154,20 @@ def test_has_many_get_with_args(self):
self.assertEqual(pics[0].name,'a pic')
+ @inlineCallbacks
+ def test_has_many_get_same_find_args(self):
+ # First, make a few pics
+ ids = [self.picture.id]
+ for _ in range(3):
+ pic = yield Picture(user_id=self.user.id).save()
+ ids.append(pic.id)
+
+ args = {'orderby': None, 'where': None, 'group': None, 'limit': None}
+ pics = yield self.user.pictures.get(**args)
+ find_pics = yield Picture.find(**args)
+ self.assertEqual(pics, find_pics)
+
+
@inlineCallbacks
def test_has_many_count_with_args(self):
# First, make a few pics
@@ -235,6 +262,23 @@ def test_habtm(self):
self.assertEqual(newcolorids, colorids)
+ @inlineCallbacks
+ def test_habtm_with_joinwhere(self):
+ color = yield FavoriteColor(name="red").save()
+ colors = [self.favcolor, color]
+ colorids = [color.id for color in colors]
+ yield FavoriteColor(name="green").save()
+
+ args = {'user_id': self.user.id, 'favorite_color_id': colors[0].id, 'palette_id': 1}
+ yield self.config.insert('favorite_colors_users', args)
+ args = {'user_id': self.user.id, 'favorite_color_id': colors[1].id, 'palette_id': 2}
+ yield self.config.insert('favorite_colors_users', args)
+
+ newcolors = yield self.user.favorite_colors.get(join_where=['palette_id = ?', 2])
+ newcolorids = [color.id for color in newcolors]
+ self.assertEqual(newcolorids, [colors[1].id])
+
+
@inlineCallbacks
def test_habtm_count(self):
color = yield FavoriteColor(name="red").save()
@@ -266,6 +310,24 @@ def test_habtm_get_with_args(self):
self.assertEqual(newcolor.id, color.id)
+ @inlineCallbacks
+ def test_habtm_get_same_find_args(self):
+ color = yield FavoriteColor(name="red").save()
+ colors = [self.favcolor, color]
+ colorids = [color.id for color in colors]
+
+ args = {'user_id': self.user.id, 'favorite_color_id': colors[0].id}
+ yield self.config.insert('favorite_colors_users', args)
+ args = {'user_id': self.user.id, 'favorite_color_id': colors[1].id}
+ yield self.config.insert('favorite_colors_users', args)
+
+ args = {'orderby': None, 'where': None, 'group': None, 'limit': None}
+
+ newcolor = yield self.user.favorite_colors.get(**args)
+ newcolor_find = yield FavoriteColor.find(**args)
+ self.assertEqual(newcolor, newcolor_find)
+
+
@inlineCallbacks
def test_habtm_count_with_args(self):
color = yield FavoriteColor(name="red").save()
@@ -281,6 +343,21 @@ def test_habtm_count_with_args(self):
self.assertEqual(newcolorsnum, 1)
+ @inlineCallbacks
+ def test_habtm_count_with__no_args(self):
+ color = yield FavoriteColor(name="red").save()
+ colors = [self.favcolor, color]
+ colorids = [color.id for color in colors]
+
+ args = {'user_id': self.user.id, 'favorite_color_id': colors[0].id}
+ yield self.config.insert('favorite_colors_users', args)
+ args = {'user_id': self.user.id, 'favorite_color_id': colors[1].id}
+ yield self.config.insert('favorite_colors_users', args)
+
+ newcolorsnum = yield self.user.favorite_colors.count(where=None)
+ self.assertEqual(newcolorsnum, 2)
+
+
@inlineCallbacks
def test_set_habtm(self):
user = yield User().save()
@@ -329,7 +406,7 @@ def test_clear_jointable_on_delete_habtm_with_custom_args(self):
cat_id = category.id
yield category.delete()
res = yield self.config.select(join_tablename, where=['category_id = ?', cat_id], limit=1)
- self.assertIsNone(res)
+ self.assertTrue(res is None)
@inlineCallbacks
@@ -344,3 +421,11 @@ def test_set_habtm_blank(self):
yield user.favorite_colors.set([])
newcolors = yield user.favorite_colors.get()
self.assertEqual(len(newcolors), 0)
+
+
+ @inlineCallbacks
+ def test_belongs_to_plural(self):
+ serviceclass = yield Serviceclass().save()
+ role = yield Role(description="Anything", serviceclass_id=serviceclass.id).save()
+ tmp = yield role.serviceclass.get()
+ self.assertEqual(serviceclass, tmp)
diff --git a/twistar/tests/test_transactions.py b/twistar/tests/test_transactions.py
new file mode 100644
index 0000000..ac704df
--- /dev/null
+++ b/twistar/tests/test_transactions.py
@@ -0,0 +1,635 @@
+from twisted.trial import unittest
+from twisted.enterprise import adbapi
+from twisted.internet.defer import inlineCallbacks
+from twisted.internet import reactor
+
+from twistar.exceptions import TransactionNotStartedError, DBObjectSaveError, TransactionAlreadyStartedError
+
+from utils import *
+
+class TransactionTest(unittest.TestCase):
+ @inlineCallbacks
+ def setUp(self):
+ yield initDB(self)
+ self.config = Registry.getConfig()
+
+ @inlineCallbacks
+ def tearDown(self):
+ yield tearDownDB(self)
+
+ @inlineCallbacks
+ def test_init_start_transaction(self):
+ pen = Pen()
+ tx = yield pen.startTransaction()
+ yield pen.commit()
+
+ @inlineCallbacks
+ def test_init_with_transaction(self):
+ pen = Pen()
+ transaction = yield pen.startTransaction()
+ pen2 = Pen(transaction = transaction)
+ self.assertEqual(pen._transaction, pen2._transaction)
+ yield pen2.commit()
+
+ @inlineCallbacks
+ def test_init_multiple_startTransaction(self):
+ pen = Pen()
+ txn = yield pen.startTransaction()
+ self.assertRaises(TransactionAlreadyStartedError, pen.startTransaction)
+ yield pen.commit()
+
+ def test_read_not_started_transaction(self):
+ pen = Pen()
+ self.assertRaises(TransactionNotStartedError, pen.transaction)
+
+ @inlineCallbacks
+ def test_inject_transaction(self):
+ pen = Pen()
+ yield pen.startTransaction()
+ another_pen = Pen()
+ another_pen.transaction(pen.transaction())
+ self.assertEqual(pen.transaction(), another_pen.transaction())
+ yield pen.commit()
+
+ @inlineCallbacks
+ def test_init_multiple_transaction(self):
+ pen = Pen()
+ txn = yield pen.startTransaction()
+ self.assertEqual(txn, pen.transaction())
+ yield pen.commit()
+
+ def test_fail_commit(self):
+ pen = Pen(color="red", len=10)
+ self.assertRaises(TransactionNotStartedError, pen.commit)
+
+ @inlineCallbacks
+ def test_save_with_transaction_no_commit(self):
+ pen = Pen(color="red", len=10)
+ yield pen.startTransaction()
+ saved_pen = yield pen.save()
+ self.assertTrue(type(pen.id) == int or type(pen.id) == long)
+
+ new_pen = yield Pen.find(pen.id)
+ self.assertEqual(new_pen, None)
+
+ # cleanup, mostly for pgsql
+ yield pen.rollback()
+
+ @inlineCallbacks
+ def test_find_outside_transaction_commit(self):
+ pen = Pen(color="red", len=10)
+ yield pen.startTransaction()
+ saved_pen = yield pen.save()
+ self.assertTrue(type(pen.id) == int or type(pen.id) == long)
+
+ new_pen = yield Pen.find(pen.id)
+ self.assertEqual(new_pen, None)
+
+ yield pen.commit()
+
+ @inlineCallbacks
+ def test_find_inside_transaction_commit(self):
+ pen = Pen(color="red", len=10)
+ txn = yield pen.startTransaction()
+ saved_pen = yield pen.save()
+ self.assertTrue(type(pen.id) == int or type(pen.id) == long)
+
+ new_pen = yield Pen.find(pen.id, transaction=txn)
+ self.assertEqual(new_pen, pen)
+
+ yield pen.commit()
+
+ @inlineCallbacks
+ def test_save_with_transaction_commit(self):
+ pen = Pen(color="red", len=10)
+ yield pen.startTransaction()
+ yield pen.save()
+ yield pen.commit()
+
+ self.assertTrue(type(pen.id) == int or type(pen.id) == long)
+
+ new_pen = yield Pen.find(pen.id)
+ self.assertEqual(new_pen.id, pen.id)
+
+ @inlineCallbacks
+ def test_multiple_save_with_transaction_commit(self):
+ pen = Pen(color="red", len=10)
+ transaction = yield pen.startTransaction()
+ saved_pen = yield pen.save()
+ self.assertTrue(type(pen.id) == int or type(pen.id) == long)
+
+ table = yield Table(color="blue", transaction=transaction)
+ self.assertEqual(transaction, table._transaction)
+ saved_table = yield table.save()
+ self.assertTrue(type(table.id) == int or type(table.id) == long)
+
+ new_pen = yield Pen.find(pen.id)
+ self.assertEqual(new_pen, None)
+
+ new_table = yield Table.find(table.id)
+ self.assertEqual(new_table, None)
+
+ yield pen.commit()
+
+ new_pen = yield Pen.find(pen.id)
+ self.assertEqual(new_pen.id, pen.id)
+
+ new_table = yield Table.find(table.id)
+ self.assertEqual(new_table.id, table.id )
+
+ @inlineCallbacks
+ def test_multiple_save_with_transaction_commit_using_another_obj(self):
+ pen = Pen(color="red", age=10)
+ transaction = yield pen.startTransaction()
+ saved_pen = yield pen.save()
+ self.assertTrue(type(pen.id) == int or type(pen.id) == long)
+
+ table = yield Table(color="blue", transaction=transaction)
+ self.assertEqual(transaction, table._transaction)
+ saved_table = yield table.save()
+ self.assertTrue(type(table.id) == int or type(table.id) == long)
+
+ new_pen = yield Pen.find(pen.id)
+ self.assertEqual(new_pen, None)
+
+ new_table = yield Table.find(table.id)
+ self.assertEqual(new_table, None)
+
+ yield table.commit()
+
+ new_pen = yield Pen.find(pen.id)
+ self.assertEqual(new_pen.id, pen.id)
+
+ new_table = yield Table.find(table.id)
+ self.assertEqual(new_table.id, table.id )
+
+ @inlineCallbacks
+ def test_already_commited(self):
+ pen = Pen(color="red", len=10)
+ transaction = yield pen.startTransaction()
+ yield pen.save()
+ yield pen.commit()
+
+ self.assertEqual(pen._transaction, None)
+
+ new_pen = yield Pen.find(pen.id)
+ self.assertEqual(new_pen.len, 10)
+
+ pen.color="yellow"
+ self.assertRaises(TransactionNotStartedError, pen.commit)
+
+ @inlineCallbacks
+ def test_rollback(self):
+ pen = Pen(color="red")
+ transaction = yield pen.startTransaction()
+ yield pen.save()
+ yield pen.rollback()
+ self.assertEqual(pen._transaction, None)
+
+ self.assertTrue(type(pen.id) == int or type(pen.id) == long)
+ new_pen = yield Pen.find(pen.id)
+ self.assertEqual(new_pen, None)
+
+ @inlineCallbacks
+ def test_concurrent_insert_1(self):
+ pen = Pen(color="red")
+ another_pen = Pen(color="red")
+
+ transaction = yield pen.startTransaction()
+ yield pen.save()
+ yield self.failUnlessFailure(another_pen.save(), Exception)
+ yield pen.commit()
+ another_pen.color = "blue"
+ yield another_pen.save()
+
+ @inlineCallbacks
+ def test_concurrent_insert_2(self):
+ pen = Pen(color="red", len=10)
+ yield pen.startTransaction()
+ yield pen.save()
+
+ another_pen = Pen(color="red", len=20)
+ yield another_pen.startTransaction()
+ yield self.failUnlessFailure(another_pen.save(), Exception)
+ yield another_pen.rollback()
+
+ self.assertEqual(another_pen.color, pen.color)
+ self.assertNotEqual(another_pen._transaction, pen._transaction)
+
+ yield pen.rollback()
+
+ new_pen = yield Pen.find(pen.id)
+ self.assertEqual(new_pen, None)
+
+ yield another_pen.startTransaction()
+ yield another_pen.save()
+ new_pen = yield Pen.find(another_pen.id)
+ self.assertEqual(new_pen, None)
+
+ yield another_pen.commit()
+ new_pen = yield Pen.find(another_pen.id)
+ self.assertEqual(new_pen.len, 20)
+
+ @inlineCallbacks
+ def test_delete(self):
+ pen = Pen(color="red", len=10)
+ yield pen.startTransaction()
+ yield pen.save()
+ yield pen.commit()
+
+ new_pen = yield Pen.find(pen.id)
+ self.assertEqual(new_pen.id, pen.id)
+
+ yield pen.startTransaction()
+ yield pen.delete()
+ yield pen.commit()
+
+ new_pen = None
+ new_pen = yield Pen.find(pen.id, limit=1)
+ self.assertEqual(new_pen, None)
+
+ @inlineCallbacks
+ def test_delete_with_rollback(self):
+ pen = Pen(color="red", len=10)
+ yield pen.startTransaction()
+ yield pen.save()
+ yield pen.commit()
+
+ new_pen = yield Pen.find(pen.id)
+ self.assertEqual(new_pen.id, pen.id)
+
+ yield pen.startTransaction()
+ yield pen.delete()
+
+ self.assertRaises(DBObjectSaveError, pen.save)
+
+ yield pen.rollback()
+
+ new_pen = None
+ new_pen = yield Pen.find(pen.id, limit=1)
+ self.assertEqual(new_pen.color, pen.color)
+
+ @inlineCallbacks
+ def test_set_hasmany(self):
+ table = yield Table(color="red")
+ transaction = yield table.startTransaction()
+ yield table.save()
+ rubbers = []
+ for _ in range(3):
+ rubber = yield Rubber(color="green", transaction=transaction).save()
+ rubbers.append(rubber)
+ rubberids = [int(rubber.id) for rubber in rubbers]
+ yield table.rubbers.set(rubbers, transaction)
+ yield table.commit()
+ self.assertEqual(table._transaction, None)
+ results = yield table.rubbers.get()
+ resultids = [int(rubber.id) for rubber in results]
+ self.assertEqual(rubberids, resultids)
+
+ @inlineCallbacks
+ def test_set_hasmany_rollback(self):
+ table = yield Table(color="red")
+ transaction = yield table.startTransaction()
+ yield table.save()
+ rubbers = []
+ for _ in range(3):
+ rubber = yield Rubber(color="green", transaction=transaction).save()
+ rubbers.append(rubber)
+ rubberids = [int(rubber.id) for rubber in rubbers]
+ yield table.rubbers.set(rubbers, transaction)
+ yield table.rollback()
+ self.assertEqual(table._transaction, None)
+ results = yield table.rubbers.get()
+ self.assertEqual(results, [])
+
+ @inlineCallbacks
+ def test_set_hasmany_no_commit(self):
+ table = yield Table(color="red")
+ transaction = yield table.startTransaction()
+ yield table.save()
+ rubbers = []
+ for _ in range(3):
+ rubber = yield Rubber(color="green", transaction=transaction).save()
+ rubbers.append(rubber)
+ rubberids = [int(rubber.id) for rubber in rubbers]
+ yield table.rubbers.set(rubbers, transaction)
+ results = yield table.rubbers.get()
+ self.assertEqual(results, [])
+ # Needed for pgsql
+ yield table.rollback()
+
+ @inlineCallbacks
+ def test_count_outside_transaction(self):
+ table = Table(color="blue")
+ transaction = yield table.startTransaction()
+ table = yield table.save()
+
+ cnt = yield Table.count()
+ self.assertEqual(cnt, 0)
+
+ yield table.commit()
+
+ cnt = yield Table.count()
+ self.assertEqual(cnt, 1)
+
+ @inlineCallbacks
+ def test_count_inside_transaction(self):
+ table = Table(color="blue")
+ transaction = yield table.startTransaction()
+ table = yield table.save()
+
+ cnt = yield Table.count(transaction=transaction)
+ self.assertEqual(cnt, 1)
+
+ yield table.rollback()
+
+ cnt = yield Table.count()
+ self.assertEqual(cnt, 0)
+
+ @inlineCallbacks
+ def test_count_habtm(self):
+ try:
+ table = Table(color="blue")
+ transaction = yield table.startTransaction()
+ table = yield table.save()
+
+ pen = yield Pen(color="red", transaction=transaction).save()
+ another_pen = yield Pen(color="green", transaction=transaction).save()
+
+ pensid = [pen.id, another_pen.id]
+ pens = [pen, another_pen]
+ yield table.pens.set(pens, transaction)
+
+ cnt = yield table.pens.count(transaction=transaction)
+ self.assertEqual(cnt, 2)
+
+ cnt = yield table.pens.count()
+ self.assertEqual(cnt, 0)
+
+ yield table.commit()
+
+ cnt = yield table.pens.count()
+ self.assertEqual(cnt, 2)
+ except Exception as e:
+ print e
+
+ @inlineCallbacks
+ def test_set_habtm(self):
+ table = yield Table(color="blue").save()
+ pen = yield Pen(color="red").save()
+ another_pen = yield Pen(color="green").save()
+ pensid = [pen.id, another_pen.id]
+ pens = [pen, another_pen]
+ transaction = yield table.startTransaction()
+ yield table.pens.set(pens, transaction)
+ yield table.commit()
+ newpens = yield table.pens.get()
+ newpensids = [pen.id for pen in newpens]
+ self.assertEqual(newpensids, pensid)
+
+ @inlineCallbacks
+ def test_set_habtm_no_commit(self):
+ table = yield Table(color="blue").save()
+ pen = yield Pen(color="red").save()
+ another_pen = yield Pen(color="green").save()
+ pensid = [pen.id, another_pen.id]
+ pens = [pen, another_pen]
+ transaction = yield table.startTransaction()
+ yield table.pens.set(pens, transaction)
+ newpens = yield table.pens.get()
+ self.assertEqual(newpens, [])
+ #Need for pgsql
+ yield table.rollback()
+
+ @inlineCallbacks
+ def test_set_habtm_rollback(self):
+ table = yield Table(color="blue").save()
+ pen = yield Pen(color="red").save()
+ another_pen = yield Pen(color="green").save()
+ pens = [pen, another_pen]
+ transaction = yield table.startTransaction()
+ yield table.pens.set(pens, transaction)
+ yield table.rollback()
+ self.assertEqual(table._transaction, None)
+ newpens = yield table.pens.get()
+ self.assertEqual(newpens, [])
+
+ @inlineCallbacks
+ def test_get_hasmany_outside_transaction(self):
+ table = yield Table(color="red")
+ transaction = yield table.startTransaction()
+ yield table.save()
+ rubbers = []
+ for _ in range(3):
+ rubber = yield Rubber(color="green", transaction=transaction).save()
+ rubbers.append(rubber)
+ rubberids = [int(rubber.id) for rubber in rubbers]
+ yield table.rubbers.set(rubbers, transaction)
+
+ table_rubbers = yield table.rubbers.get()
+ self.assertEqual(table_rubbers, [])
+
+ yield table.rollback()
+
+ @inlineCallbacks
+ def test_get_hasmany_inside_transaction(self):
+ table = yield Table(color="red")
+ transaction = yield table.startTransaction()
+ yield table.save()
+ rubbers = []
+ for _ in range(3):
+ rubber = yield Rubber(color="green", transaction=transaction).save()
+ rubbers.append(rubber)
+ rubberids = [int(rubber.id) for rubber in rubbers]
+ yield table.rubbers.set(rubbers, transaction)
+
+ table_rubbers = yield table.rubbers.get(transaction=transaction)
+ self.assertNotEqual(table_rubbers, [])
+ self.assertEqual(len(table_rubbers), 3)
+
+ yield table.rollback()
+
+ @inlineCallbacks
+ def test_has_one_inside_transaction(self):
+ user = User()
+ txn = yield user.startTransaction()
+ yield user.save()
+ avatar=Avatar()
+ avatar.transaction(txn)
+ yield avatar.save()
+ yield user.avatar.set(avatar, transaction=txn)
+ new_avatar = yield user.avatar.get(transaction=txn)
+ self.assertEqual(avatar, new_avatar)
+ yield user.commit()
+ new_avatar = yield user.avatar.get()
+ self.assertEqual(avatar, new_avatar)
+
+ @inlineCallbacks
+ def test_has_one_outside_transaction(self):
+ user = User()
+ txn = yield user.startTransaction()
+ yield user.save()
+ avatar=Avatar()
+ avatar.transaction(txn)
+ yield avatar.save()
+ yield user.avatar.set(avatar, transaction=txn)
+ new_avatar = yield user.avatar.get()
+ self.assertNotEqual(avatar, new_avatar)
+ self.assertEqual(new_avatar, None)
+ yield user.commit()
+ new_avatar = yield user.avatar.get()
+ self.assertEqual(avatar, new_avatar)
+
+ @inlineCallbacks
+ def test_belongsTo_inside_transaction(self):
+ pic=Picture()
+ txn = yield pic.startTransaction()
+ user = User()
+ user.transaction(txn)
+ yield user.save()
+ yield pic.save()
+ yield pic.user.set(user, transaction=txn)
+ new_user = yield pic.user.get(transaction=txn)
+ yield user.commit()
+ self.assertEqual(user, new_user)
+ new_user = yield pic.user.get()
+ self.assertEqual(user, new_user)
+
+ @inlineCallbacks
+ def test_belongsTo_outside_transaction(self):
+ pic=Picture()
+ txn = yield pic.startTransaction()
+ user = User()
+ user.transaction(txn)
+ yield user.save()
+ yield pic.save()
+ yield pic.user.set(user, transaction=txn)
+ new_user = yield User.find(id=1)
+ self.assertNotEqual(user, new_user)
+ self.assertEqual(new_user, None)
+ yield user.commit()
+ new_user = yield User.find(id=1)
+ self.assertEqual(user, new_user)
+
+ @inlineCallbacks
+ def test_findBy(self):
+ r = yield User.findBy(first_name="Some", last_name="User", age=20)
+ self.assertEqual(r, [])
+
+ user = User(first_name="Some", last_name="User", age=20)
+ txn = yield user.startTransaction()
+ yield user.save()
+
+ r = yield User.findBy(first_name="Some", last_name="User", age=20, transaction=txn)
+ self.assertEqual(r[0], user)
+
+ yield user.commit()
+
+ r = yield User.findBy(first_name="Some", last_name="User", age=20)
+ self.assertEqual(r[0], user)
+
+ @inlineCallbacks
+ def test_findBy_rollback(self):
+ r = yield User.findBy(first_name="Some", last_name="User", age=20)
+ self.assertEqual(r, [])
+
+ user = User(first_name="Some", last_name="User", age=20)
+ txn = yield user.startTransaction()
+ yield user.save()
+
+ r = yield User.findBy(first_name="Some", last_name="User", age=20, transaction=txn)
+ self.assertEqual(r[0], user)
+
+ yield user.rollback()
+
+ r = yield User.findBy(first_name="Some", last_name="User", age=20)
+ self.assertEqual(r, [])
+
+ @inlineCallbacks
+ def test_findOrCreate(self):
+ user = User(first_name="First", last_name="Last", age=10)
+ txn = yield user.startTransaction()
+ yield user.save()
+
+ # make sure we didn't create a new user
+ r = yield User.findOrCreate(first_name="First", transaction=txn)
+ self.assertEqual(r.id, user.id)
+
+ # make sure we do create a new user
+ r = yield User.findOrCreate(first_name="First", last_name="Non", transaction=txn)
+ txn_id = r.id
+ self.assertTrue(r.id != user.id)
+
+ yield user.commit()
+
+ # make sure we do create a new user
+ r = yield User.findOrCreate(first_name="First", last_name="Non")
+ self.assertTrue(r.id != user.id)
+ self.assertTrue(r.id == txn_id)
+
+ @inlineCallbacks
+ def test_findOrCreate_rollback(self):
+ user = User(first_name="First", last_name="Last", age=10)
+ txn = yield user.startTransaction()
+ yield user.save()
+
+ # make sure we didn't create a new user
+ r = yield User.findOrCreate(first_name="First", transaction=txn)
+ self.assertEqual(r.id, user.id)
+
+ # make sure we do create a new user
+ r = yield User.findOrCreate(first_name="First", last_name="Non", transaction=txn)
+ txn_id = r.id
+ self.assertTrue(r.id != user.id)
+
+ yield user.rollback()
+
+ # make sure we do create a new user
+ r = yield User.findOrCreate(first_name="First", last_name="Non")
+ self.assertTrue(r.id != txn_id)
+
+ cnt = yield User.count()
+ self.assertEqual(cnt, 1)
+
+ @inlineCallbacks
+ def test_transacted_operation_after_commit_raises(self):
+ pen = Pen(color="red", len=10)
+ transaction = yield pen.startTransaction()
+ yield pen.save()
+ yield pen.commit()
+
+ new_pen = Pen(color="yellow", len=10)
+ new_pen.transaction(transaction)
+
+ yield self.assertFailure(new_pen.save(), TransactionNotStartedError)
+
+ @inlineCallbacks
+ def test_loadRelations_inside_transactions(self):
+ user = User(first_name="First", last_name="Last", age=10)
+ txn = yield user.startTransaction()
+ yield user.save()
+
+ picture = Picture(name="a pic", size=10, user_id=user.id, transaction=txn)
+ yield picture.save()
+
+ relations = yield user.loadRelations(transaction=txn)
+ user_pictures = yield user.pictures.get(transaction=txn)
+
+ self.assertEqual(user_pictures, relations['pictures'])
+
+ yield user.rollback()
+
+ @inlineCallbacks
+ def test_loadRelations_outside_transactions(self):
+ user = User(first_name="First", last_name="Last", age=10)
+ txn = yield user.startTransaction()
+ yield user.save()
+
+ picture = Picture(name="a pic", size=10, user_id=user.id, transaction=txn)
+ yield picture.save()
+
+ relations = yield user.loadRelations()
+
+ self.assertEqual([], relations['pictures'])
+
+ yield user.rollback()
diff --git a/twistar/tests/utils.py b/twistar/tests/utils.py
index aa2918f..e5bb456 100644
--- a/twistar/tests/utils.py
+++ b/twistar/tests/utils.py
@@ -9,13 +9,16 @@
#from postgres_config import initDB, tearDownDB
class User(DBObject):
- HASMANY = ['pictures']
+ HASMANY = ['pictures', 'comments']
HASONE = ['avatar']
HABTM = ['favorite_colors']
class Picture(DBObject):
BELONGSTO = ['user']
+class Comment(DBObject):
+ BELONGSTO = ['user']
+
class Avatar(DBObject):
pass
@@ -43,7 +46,28 @@ class Girl(DBObject):
class Nickname(DBObject):
BELONGSTO = [{'name': 'nicknameable', 'polymorphic': True}]
+class Pen(DBObject):
+ pass
+
+class Table(DBObject):
+ HABTM = ['pens']
+ HASMANY = ['rubbers']
+
+class Rubber(DBObject):
+ pass
+
+class Role(DBObject):
+ BELONGSTO = ['serviceclass']
+
+class Serviceclass(DBObject):
+ HASMANY = ['roles']
+ BELONGSTO = ['superclass']
+
+class Superclass(DBObject):
+ HASONE = ['serviceclass']
-Registry.register(Picture, User, Avatar, FakeObject, FavoriteColor)
+Registry.register(Picture, User, Comment, Avatar, FakeObject, FavoriteColor)
Registry.register(Boy, Girl, Nickname)
Registry.register(Blogpost, Category)
+Registry.register(Pen, Table, Rubber)
+Registry.register(Role, Serviceclass, Superclass)
diff --git a/twistar/validation.py b/twistar/validation.py
index 2558a03..9a0a7c1 100644
--- a/twistar/validation.py
+++ b/twistar/validation.py
@@ -19,7 +19,7 @@ def presenceOf(obj, names, kwargs):
"""
message = kwargs.get('message', "cannot be blank.")
for name in names:
- if getattr(obj, name, "") == "":
+ if getattr(obj, name, "") in ("", None):
obj.errors.add(name, message)
@@ -31,8 +31,8 @@ def lengthOf(obj, names, kwargs):
containing valid values, or a C{length} keyword with the exact length
allowed.
- For those named properties that do not have
- the specified length, an error will be recorded in C{obj.errors}.
+ For those named properties that do not have the specified length
+ (or that are C{None}), an error will be recorded in C{obj.errors}.
@param obj: The object whose properties need to be tested.
@param names: The names of the properties to test.
@@ -50,7 +50,8 @@ def lengthOf(obj, names, kwargs):
else:
message = kwargs.get('message', "must have a length between %s and %s (inclusive)." % minmax)
for name in names:
- if not len(getattr(obj, name, "")) in xr:
+ val = getattr(obj, name, "")
+ if val is None or not len(val) in xr:
obj.errors.add(name, message)
diff --git a/txconnectionpool/__init__.py b/txconnectionpool/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/txconnectionpool/tests/__init__.py b/txconnectionpool/tests/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/txconnectionpool/tests/test_txconnectionpool.py b/txconnectionpool/tests/test_txconnectionpool.py
new file mode 100644
index 0000000..4af6a7f
--- /dev/null
+++ b/txconnectionpool/tests/test_txconnectionpool.py
@@ -0,0 +1,86 @@
+from twisted.internet import defer
+from twisted.trial import unittest
+
+from txconnectionpool.txconnectionpool import TxConnectionPool
+
+
+class FakeTxnError(Exception):
+ pass
+
+
+class FailingToStartConnection(object):
+ def __init__(self, _pool):
+ raise FakeTxnError('fake connection excpected failure')
+
+
+class FakeCursor(object):
+ def close(self):
+ pass
+
+
+class FailingToCommitOrRollbackConnection(object):
+ def __init__(self, _pool):
+ pass
+
+ def cursor(self):
+ return FakeCursor()
+
+ def reconnect(self):
+ pass
+
+ def rollback(self):
+ raise FakeTxnError('fake connection expected failure')
+
+ def commit(self):
+ raise FakeTxnError('fake connection expected failure')
+
+
+class TxConnectionPoolTest(unittest.TestCase):
+ def test_failure_to_start_txn_will_release_thread(self):
+ cp = TxConnectionPool('sqlite3')
+ cp.max = 1
+ cp.connectionFactory = FailingToStartConnection
+
+ d = cp.startTransaction()
+ self.assertFailure(d, FakeTxnError)
+
+ d = cp.startTransaction()
+ self.assertFailure(d, FakeTxnError)
+
+ return d
+
+ @defer.inlineCallbacks
+ def test_failure_to_commit_will_release_thread(self):
+ cp = TxConnectionPool('sqlite3')
+ cp.max = 1
+ cp.connectionFactory = FailingToCommitOrRollbackConnection
+
+ txn = yield cp.startTransaction()
+ d = cp.commitTransaction(txn)
+ self.assertFailure(d, FakeTxnError)
+
+ yield d
+
+ txn = yield cp.startTransaction()
+ d = cp.commitTransaction(txn)
+ self.assertFailure(d, FakeTxnError)
+
+ yield d
+
+ @defer.inlineCallbacks
+ def test_failure_to_rollback_will_release_thread(self):
+ cp = TxConnectionPool('sqlite3')
+ cp.max = 1
+ cp.connectionFactory = FailingToCommitOrRollbackConnection
+
+ txn = yield cp.startTransaction()
+ d = cp.rollbackTransaction(txn)
+ self.assertFailure(d, FakeTxnError)
+
+ yield d
+
+ txn = yield cp.startTransaction()
+ d = cp.rollbackTransaction(txn)
+ self.assertFailure(d, FakeTxnError)
+
+ yield d
diff --git a/txconnectionpool/txconnectionpool.py b/txconnectionpool/txconnectionpool.py
new file mode 100644
index 0000000..be6f12f
--- /dev/null
+++ b/txconnectionpool/txconnectionpool.py
@@ -0,0 +1,187 @@
+from threading import Lock
+
+from twisted.python import log
+from twisted.enterprise.adbapi import ConnectionPool
+from twistar.exceptions import TransactionNotStartedError
+
+from txthreadworker.txthreadworker import TxThreadWorker
+
+class TxConnectionPool(ConnectionPool):
+
+ txWorkers = {}
+ txWorkersLock = None
+
+ def startTransaction(self):
+ """Open a new database Transaction.
+
+ @return: a Deferred which will fire the Transaction or a Failure.
+
+ Since this connection will be in use until the Transaction is
+ completed, the thread that we call the function in gets blocked
+ until then. The Semaphore it is waiting on is stored in
+ the self.transLock dictionary.
+ """
+
+ warning_limit = self.threadpool.max / 10
+ if len(self.threadpool.working) >= self.threadpool.max:
+ log.msg('Warning! asking for a new thread, but all slots are busy')
+ log.msg(self.dumpThreadPoolStats())
+ elif ( len(self.threadpool.working)
+ >= self.threadpool.max - warning_limit ):
+ log.msg('Warning! threadpool is quite full')
+ log.msg(self.dumpThreadPoolStats())
+
+ worker = TxThreadWorker(self.threadpool)
+ worker.start()
+ if not self.txWorkersLock:
+ self.txWorkersLock = Lock()
+
+ def initTx():
+ conn = self.connectionFactory(self)
+ t = self.transactionFactory(self, conn)
+
+ self.txWorkersLock.acquire()
+ self.txWorkers[t] = worker
+ self.txWorkersLock.release()
+
+ return t
+
+ def stopWorkerOnError(err):
+ # stop the thread if initTx fails
+ return self._stopTxWorker(worker).addCallback(lambda _: err)
+
+ d = worker.submit(initTx)
+ return d.addErrback(stopWorkerOnError)
+
+ def runQueryInTransaction(self, trans, *args, **kw):
+ """Execute an SQL query in the specified Transaction and return the result.
+
+ This function is similar to runQuery but uses a previously created
+ Transaction and does not commit or rollback the connection upon
+ completion.
+ """
+ return self._deferToTrans(self._runQueryInTransaction, trans, *args, **kw)
+
+ def executeOperationInTransaction(self, f, trans, *args, **kw):
+ """
+ Execute the specified function into the transaction thread and return result
+ """
+ return self._deferToTrans(f, trans, *args, **kw)
+
+ def runOperationInTransaction(self, trans, *args, **kw):
+ """Execute an SQL query in the specified Transaction and return None.
+
+ This function is similar to runOperation but uses a previously created
+ Transaction and does not commit or rollback the connection upon
+ completion.
+ """
+ return self._deferToTrans(self._runOperationInTransaction, trans, *args, **kw)
+
+ def commitTransaction(self, trans):
+ """Commit the transaction to the database."""
+ d = self._deferToTrans(self._commitTransaction, trans)
+ d.addCallback(self._stopTxWorker)
+ return d
+
+ def rollbackTransaction(self, trans):
+ """Exit the transaction without committing."""
+
+ d = self._deferToTrans(self._rollbackTransaction, trans)
+ d.addCallback(self._stopTxWorker)
+ return d
+
+ def dumpThreadPoolStats(self):
+ return '\t'.join([ 'waiters: %s' % len(self.threadpool.waiters),
+ 'workers: %s' % len(self.threadpool.working),
+ 'total: %s' % len(self.threadpool.threads),
+ ])
+
+ def _stopTxWorker(self, worker):
+ return worker.stop()
+
+ def _runQueryInTransaction(self, trans, *args, **kwargs):
+ try:
+ trans.execute(*args, **kwargs)
+ if(trans.rowcount != 0):
+ result = trans.fetchall()
+ else:
+ result = []
+ return result
+ except:
+ log.msg('Exception in SQL query %s'%args)
+ log.deferr()
+ raise
+
+ def _runOperationInTransaction(self, trans, *args, **kwargs):
+ try:
+ return trans.execute(*args, **kwargs)
+ except:
+ log.msg('Exception in SQL operation %s %s'%(trans, args))
+ log.deferr()
+ raise
+
+ def _commitTransaction(self, trans):
+ if trans._cursor is None:
+ raise TransactionNotStartedError("Cannot call commit without a transaction")
+
+ worker = self._removeWorkerFromDict(trans)
+
+ try:
+ conn = trans._connection
+ trans.close()
+ conn.commit()
+ except Exception as e:
+ log.err("commit error %r", e)
+ self._stopTxWorker(worker)
+ raise e
+
+ return worker
+
+ def _rollbackTransaction(self, trans):
+ if trans._cursor is None:
+ raise TransactionNotStartedError("Cannot call rollback without a transaction")
+
+ worker = self._removeWorkerFromDict(trans)
+
+ try:
+ conn = trans._connection
+ conn.rollback()
+ except Exception, e:
+ log.err("Rollback error %s"%(e))
+ self._stopTxWorker(worker)
+ raise e
+
+ return worker
+
+ def _removeWorkerFromDict(self, trans):
+ self.txWorkersLock.acquire()
+ try:
+ worker = self.txWorkers[trans]
+ del self.txWorkers[trans]
+ except Exception as e:
+ raise e
+ finally:
+ self.txWorkersLock.release()
+
+ return worker
+
+ def _deferToTrans(self, f, trans, *args, **kwargs):
+ """Internal function.
+
+ Push f onto the transaction's work queue.
+ """
+ d = None
+ self.txWorkersLock.acquire()
+
+ try:
+ worker = self.txWorkers.get(trans)
+ if not worker:
+ err = "Cannot execute operation in the given transaction"
+ raise TransactionNotStartedError(err)
+ d = worker.submit(f, trans, *args, **kwargs)
+ except Exception as e:
+ raise e
+ finally:
+ self.txWorkersLock.release()
+
+ return d
diff --git a/txthreadworker/README b/txthreadworker/README
new file mode 100644
index 0000000..7c94be9
--- /dev/null
+++ b/txthreadworker/README
@@ -0,0 +1 @@
+This class has been contributed from Flavio Grossi
diff --git a/txthreadworker/__init__.py b/txthreadworker/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/txthreadworker/txthreadworker.py b/txthreadworker/txthreadworker.py
new file mode 100644
index 0000000..310abc7
--- /dev/null
+++ b/txthreadworker/txthreadworker.py
@@ -0,0 +1,111 @@
+# vim: ts=4:sw=4:nu:fdc=4:nospell:expandtab
+
+from Queue import Queue
+
+from twisted.internet import defer
+from twisted.internet import reactor
+from twisted.python.constants import Names, NamedConstant
+
+
+class ThreadWorkerState(Names):
+ STOPPED = NamedConstant()
+ RUNNING = NamedConstant()
+
+
+class TxThreadWorker(object):
+ """
+ Thread worker which executes all submitted jobs in a single thread.
+
+ All public methods must be called from the reactor's thread.
+ """
+
+ # this class defines a work queue to which thread's job will be submitted.
+ # The only task this thread will do is try to empty the work queue.
+ # It runs until a stop() is called from the reactor thread, in which case it
+ # will stop as soon as all jobs are completed.
+
+ _STOP = object() # dummmy object to represent the stop command (a job that
+ # when enqueued will stop the ThreadWorker)
+
+ def __init__(self, threadpool=None):
+ self.threadpool = threadpool or reactor.getThreadPool()
+
+ self._current_state = ThreadWorkerState.STOPPED
+ self._work_queue = None
+
+ # deferred to be fired when stop() has completed
+ self._stop_finished = None
+
+ def start(self):
+ """
+ Start this ThreadWorker and make it ready to accept work.
+ """
+ if self._current_state != ThreadWorkerState.STOPPED:
+ return
+
+ self._work_queue = Queue()
+ self._current_state = ThreadWorkerState.RUNNING
+
+ self.threadpool.callInThread(self._start)
+
+ def stop(self):
+ """
+ Stop this ThreadWorker.
+
+ @returns: a deferred which will fire when ThreadWorker has been stopped.
+ @rtype: a C{Defer}
+ """
+ if self._current_state != ThreadWorkerState.RUNNING:
+ return defer.succeed(None)
+
+ self._stop_finished = defer.Deferred()
+ self._work_queue.put(self._STOP)
+
+ return self._stop_finished
+
+ def submit(self, job, *args, **kwargs):
+ """
+ Submit a callable to be run in the ThreadWorker.
+
+ @param job: the job to run
+ @type job: a C{callable} object
+
+ @returns: a deferred which will fire with the job result
+ @rtype: a C{Defer}
+ """
+ if self._current_state != ThreadWorkerState.RUNNING:
+ err = 'Cannot submit jobs to a stopped ThreadWorker'
+ return defer.fail(RuntimeError(err))
+ d = defer.Deferred()
+ self._work_queue.put((d, job, args, kwargs))
+ return d
+
+ def _start(self):
+ while True:
+ job = self._work_queue.get()
+ if job is self._STOP:
+ reactor.callFromThread(self._stop)
+ return
+
+ self._completeJob(*job)
+
+ def _stop(self):
+ self._work_queue = None
+ self._current_state = ThreadWorkerState.STOPPED
+ d = self._stop_finished
+ d.callback(None)
+ self._stop_finished = None
+
+ def _completeJob(self, deferred, job, args, kwargs):
+ try:
+ res = job(*args, **kwargs)
+ except Exception as e:
+ reactor.callFromThread(deferred.errback, e)
+ else:
+ reactor.callFromThread(deferred.callback, res)
+
+ def __repr__(self):
+ object_memory_address = hex(id(self))
+ return "<%s ThreadWorker object at %s>" % (self._current_state,
+ object_memory_address)
+