Browse files

fix time support.

  • Loading branch information...
1 parent 53c1c32 commit 28e058349de80524410566679a37191a91e1b55f @alex alex committed Jul 8, 2011
Showing with 27 additions and 22 deletions.
  1. +12 −1 MySQLdb/converters.py
  2. +15 −21 tests/test_cursor.py
View
13 MySQLdb/converters.py
@@ -1,3 +1,4 @@
+import math
from datetime import datetime, date, time, timedelta
from decimal import Decimal
@@ -62,7 +63,17 @@ def date_decoder(value):
return date(*[int(part) for part in value.split("-")])
def time_decoder(value):
- return timedelta(*[int(part) for part in value.split(":")])
+ # MySQLdb returns a timedelta here, immitate this nonsense.
+ hours, minutes, seconds = value.split(":")
+ td = timedelta(
+ hours = int(hours),
+ minutes = int(minutes),
+ seconds = int(seconds),
+ microseconds = int(math.modf(float(seconds))[0]*1000000),
+ )
+ if hours < 0:
+ td = -td
+ return td
def timestamp_decoder(value):
if " " in value:
View
36 tests/test_cursor.py
@@ -76,14 +76,13 @@ def test_longlong(self, connection):
cur.fetchone()
def test_datetime(self, connection):
- with contextlib.closing(connection.cursor()) as cur:
- cur.execute("SELECT SYSDATE()")
- row = cur.fetchone()
- now = datetime.datetime.now()
- assert row[0].date() == now.date()
- assert row[0].hour == now.hour
- # ~1 in 60 chance of spurious failure, i'll take those odds
- assert row[0].minute == now.minute
+ with self.create_table(connection, "events", dt="TIMESTAMP"):
+ with contextlib.closing(connection.cursor()) as cur:
+ t = datetime.datetime.combine(datetime.datetime.today(), datetime.time(12, 20, 2))
+ cur.execute("INSERT INTO events (dt) VALUES (%s)", (t,))
+ cur.execute("SELECT dt FROM events")
+ r, = cur.fetchall()
+ assert r == (t,)
def test_date(self, connection):
with contextlib.closing(connection.cursor()) as cur:
@@ -92,20 +91,15 @@ def test_date(self, connection):
assert row == (datetime.date.today(),)
def test_time(self, connection):
- with contextlib.closing(connection.cursor()) as cur:
- cur.execute("SELECT CURTIME()")
- row, = cur.fetchall()
- # TODO: more detailed tests, right now we're basically going off
- # what makes SQLAlchemy tests pass
-
- def test_timestamp(self, connection):
- with self.create_table(connection, "events", dt="TIMESTAMP"):
- with contextlib.closing(connection.cursor()) as cur:
- t = datetime.datetime.combine(datetime.datetime.today(), datetime.time(12, 20, 2))
- cur.execute("INSERT INTO events (dt) VALUES (%s)", (t,))
- cur.execute("SELECT dt FROM events")
+ with self.create_table(connection, "events", t="TIME"):
+ with contextlib.closing(connection.cursor()) as cur:
+ t = datetime.time(12, 20, 2)
+ cur.execute("INSERT INTO events (t) VALUES (%s)", (t,))
+ cur.execute("SELECT t FROM events")
r, = cur.fetchall()
- assert r == (t,)
+ # Against all rationality, MySQLdb returns a timedelta here
+ # immitate this idiotic behavior.
+ assert r == (datetime.timedelta(hours=12, minutes=20, seconds=2),)
def test_binary(self, connection):
self.assert_roundtrips(connection, "".join(chr(x) for x in xrange(255)))

0 comments on commit 28e0583

Please sign in to comment.