Timedelta not supported when using pymysql

[mikkiweesenaar]

Environment:

MacOSX El Capitan with Python 3.5.1.
python3 -c "import pony; import pymysql; print(pony.__version__, ' & ', pymysql.__version__);"
0.6.6 & 0.7.6.None

Reproduction scenario:

from datetime import timedelta
import unittest

from pony.orm import *
from pony.orm.tests.testutils import *


#db = Database('sqlite', ':memory:')
db = Database('mysql', host='localhost', port=3306, user='myuser', passwd='mypass', db='mydb')


class TimeDeltaEntities(db.Entity):
    id = PrimaryKey(int)
    t = Required(timedelta)

db.generate_mapping(create_tables=True)

with db_session:
    TimeDeltaEntities(id=1, t=timedelta(minutes=1))
    TimeDeltaEntities(id=2, t=timedelta(hours=1))


class TestTimeDelta(unittest.TestCase):

    @db_session
    def test_1(self):
        timedeltas = TimeDeltaEntities.select()
        self.assertEqual(len(timedeltas), 2)
        t1, t2 = timedeltas
        self.assertTrue(t1.t.seconds, 60 * 1)
        self.assertTrue(t2.t.seconds, 60 * 60 * 1)

Actual behaviour:

======================================================================
ERROR: Failure: CommitException (TypeError: <lambda>() takes exactly 1 argument (2 given))
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/Library/Python/2.7/site-packages/nose/loader.py", line 418, in loadTestsFromName
    addr.filename, addr.module)
  File "/Library/Python/2.7/site-packages/nose/importer.py", line 47, in importFromPath
    return self.importFromDir(dir_path, fqname)
  File "/Library/Python/2.7/site-packages/nose/importer.py", line 94, in importFromDir
    mod = load_module(part_fqname, fh, filename, desc)
  File "/Users/mikkiweesenaar/git/pony/pony/orm/tests/test_timedelta.py", line 20, in <module>
    TimeDeltaEntities(id=2, t=timedelta(hours=1))
  File "/Users/mikkiweesenaar/git/pony/pony/orm/core.py", line 378, in __exit__
    commit()
  File "<auto generated wrapper of commit() function>", line 2, in commit
  File "/Users/mikkiweesenaar/git/pony/pony/utils/utils.py", line 66, in cut_traceback
    return func(*args, **kwargs)
  File "/Users/mikkiweesenaar/git/pony/pony/orm/core.py", line 292, in commit
    transact_reraise(CommitException, exceptions)
  File "/Users/mikkiweesenaar/git/pony/pony/orm/core.py", line 276, in transact_reraise
    reraise(exc_class, new_exc, tb)
  File "/Users/mikkiweesenaar/git/pony/pony/orm/core.py", line 286, in commit
    try: primary_cache.commit()
  File "/Users/mikkiweesenaar/git/pony/pony/orm/core.py", line 1524, in commit
    if cache.modified: cache.flush()
  File "/Users/mikkiweesenaar/git/pony/pony/orm/core.py", line 1585, in flush
    if obj is not None: obj._save_()
  File "/Users/mikkiweesenaar/git/pony/pony/orm/core.py", line 4773, in _save_
    if status == 'created': obj._save_created_()
  File "/Users/mikkiweesenaar/git/pony/pony/orm/core.py", line 4670, in _save_created_
    else: database._exec_sql(sql, arguments, start_transaction=True)
  File "/Users/mikkiweesenaar/git/pony/pony/orm/core.py", line 683, in _exec_sql
    connection = cache.reconnect(e)
  File "/Users/mikkiweesenaar/git/pony/pony/orm/core.py", line 1491, in reconnect
    if not provider.should_reconnect(exc): reraise(*sys.exc_info())
  File "/Users/mikkiweesenaar/git/pony/pony/orm/core.py", line 681, in _exec_sql
    try: new_id = provider.execute(cursor, sql, arguments, returning_id)
  File "<auto generated wrapper of execute() function>", line 2, in execute
  File "/Users/mikkiweesenaar/git/pony/pony/orm/dbapiprovider.py", line 48, in wrap_dbapi_exceptions
    try: return func(provider, *args, **kwargs)
  File "/Users/mikkiweesenaar/git/pony/pony/orm/dbapiprovider.py", line 240, in execute
    else: cursor.execute(sql, arguments)
  File "/Library/Python/2.7/site-packages/pymysql/cursors.py", line 159, in execute
    query = self.mogrify(query, args)
  File "/Library/Python/2.7/site-packages/pymysql/cursors.py", line 138, in mogrify
    query = query % self._escape_args(args, conn)
  File "/Library/Python/2.7/site-packages/pymysql/cursors.py", line 113, in _escape_args
    return tuple(conn.escape(arg) for arg in args)
  File "/Library/Python/2.7/site-packages/pymysql/cursors.py", line 113, in <genexpr>
    return tuple(conn.escape(arg) for arg in args)
  File "/Library/Python/2.7/site-packages/pymysql/connections.py", line 793, in escape
    return escape_item(obj, self.charset, mapping=mapping)
  File "/Library/Python/2.7/site-packages/pymysql/converters.py", line 26, in escape_item
    val = encoder(val, mapping)
CommitException: TypeError: <lambda>() takes exactly 1 argument (2 given)

Expected behaviour:

I would have expected that the behaviour would be the same for both sqlite as mysql - that it would just work. I think it is due to

pony/pony/orm/dbproviders/mysql.py

Line 28 in e66d0fe

mysql_converters.encoders[timedelta] = lambda val: mysql_converters.escape_str(timedelta2str(val))

.
If I comment this line out, the code works fine.

[kozlovsky]

Thanks, should be fixed now

[mikkiweesenaar]

I tried to make tests for it, but apart from really telling the code where my mysql instance is, I wasn't quickly able to create a test.
Can you give me a quick hint with how I can make (some) test(s) for it? :)

-edit- Thanks for the fix!

[abetkin]

Hi @mikkiweesenaar,

The usually convenient way to do the test setup is:

1. Create database as a class fixture
2. Create test data in the test setup
3. Clear all tables in the test teardown

Here is an example of doing this with unittest:

from contextlib import closing
import unittest
from pony.orm import *

# Somewhere in the project
def declare_entities(db):
    class Author(db.Entity):
        name = Required(str)
        books = Set('Book')

    class Book(db.Entity):
        title = Required(str)
        author = Required(Author)


# Then in tests:

class DbFixture(object):
    db_name = 'test_db'
    CONNECTION = {
        'host': "localhost",
        'user': "ponytest",
        'passwd': "ponytest",
    }

    @classmethod
    def setUpClass(cls):
        from pony.orm.dbproviders.mysql import mysql_module
        with closing(mysql_module.connect(**cls.CONNECTION).cursor()) as c:
            try:
                c.execute('use sys')
                c.execute('drop database %s' % cls.db_name)
            except mysql_module.DatabaseError as exc:
                print('Did not drop %s: %s' % (cls.db_name, exc))
            c.execute('create database %s' % cls.db_name)
            c.execute('use %s' % cls.db_name)
        cls.db = Database('mysql', db=cls.db_name, **cls.CONNECTION)
        return cls.db


    @db_session
    def tearDown(self):
        for entity in db.entities.values():
            delete(i for i in entity)


class MyTest(DbFixture, unittest.TestCase):

    @classmethod
    def setUpClass(cls):
        global db
        db = DbFixture.setUpClass()
        declare_entities(db)
        db.generate_mapping(create_tables=True)

    @db_session
    def setUp(self):
        Rowling = db.Author(name='J. K. Rowling')
        db.Book(title='Harry Potter', author=Rowling)


    @db_session
    def test(self):
        self.assertEqual(
            get(b.title for a in db.Author for b in a.books), 'Harry Potter'
        )

    test2 = test # checking that teardown works

Another option is to create new database for every test, which is slower.
In this case you will need only the setUp method.