Commit 08970884 authored by Lukáš Lalinský's avatar Lukáš Lalinský

Support for TIME

parent 0bdd853b
......@@ -15,6 +15,7 @@
import logging
import collections
import base64
import datetime
from decimal import Decimal
from phoenixdb.types import Binary
from phoenixdb.errors import OperationalError, NotSupportedError, ProgrammingError
......@@ -28,6 +29,21 @@ ColumnDescription = collections.namedtuple('ColumnDescription', 'name type_code
"""Named tuple for representing results from :attr:`Cursor.description`."""
def time_from_java_sql_time(n):
microsecond = 1000 * (n % 1000)
n /= 1000
second = n % 60
n /= 60
minute = n % 60
n /= 60
hour = n % 24
return datetime.time(hour, minute, second, microsecond)
def time_to_java_sql_time(t):
return ((t.hour * 60 + t.minute) * 60 + t.second) * 1000 + t.microsecond / 1000
class Cursor(object):
"""Database cursor for executing queries and iterating over results.
......@@ -137,6 +153,8 @@ class Cursor(object):
self._column_data_types.append((i, Decimal))
elif column['columnClassName'] == 'java.lang.Float' or column['columnClassName'] == 'java.lang.Double':
self._column_data_types.append((i, float))
elif column['columnClassName'] == 'java.sql.Time':
self._column_data_types.append((i, time_from_java_sql_time))
elif column['type']['name'] == 'BINARY':
self._column_data_types.append((i, base64.b64decode))
for parameter in signature['parameters']:
......@@ -158,6 +176,8 @@ class Cursor(object):
self._parameter_data_types.append(('BOOLEAN', None))
elif parameter['className'] == 'java.lang.String':
self._parameter_data_types.append(('STRING', None))
elif parameter['className'] == 'java.sql.Time':
self._parameter_data_types.append(('JAVA_SQL_TIME', time_to_java_sql_time))
elif parameter['className'] == '[B':
self._parameter_data_types.append(('BYTE_STRING', Binary))
else:
......
import unittest
import datetime
import phoenixdb
from decimal import Decimal
from phoenixdb.tests import DatabaseTestCase
......@@ -113,12 +114,22 @@ class TypesTest(DatabaseTestCase):
self.assertEqual(cursor.description[1].type_code, phoenixdb.BOOLEAN)
self.assertEqual(cursor.fetchall(), [[1, True], [2, False], [3, None], [4, True], [5, False], [6, None]])
@unittest.skip("broken")
def test_time(self):
self.createTable("phoenixdb_test_tbl1", "id integer primary key, val time")
with self.conn.cursor() as cursor:
cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (1, '12')")
cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (1, '1970-01-01 12:01:02')")
cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (2, NULL)")
cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (3, ?)", [phoenixdb.Time(12, 1, 2)])
cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (4, ?)", [datetime.time(12, 1, 2)])
cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (5, ?)", [None])
cursor.execute("SELECT id, val FROM phoenixdb_test_tbl1 ORDER BY id")
self.assertEqual(cursor.fetchall(), [
[1, datetime.time(12, 1, 2)],
[2, None],
[3, datetime.time(12, 1, 2)],
[4, datetime.time(12, 1, 2)],
[5, None],
])
def test_varchar(self):
self.createTable("phoenixdb_test_tbl1", "id integer primary key, val varchar")
......
......@@ -24,17 +24,17 @@ __all__ = [
def Date(year, month, day):
"""Constructs an object holding a date value."""
return datetime.date(year, month, day).isoformat()
return datetime.date(year, month, day)
def Time(hour, minute, second):
"""Constructs an object holding a time value."""
return datetime.time(hour, minute, second).isoformat()
return datetime.time(hour, minute, second)
def Timestamp(year, month, day, hour, minute, second):
"""Constructs an object holding a datetime/timestamp value."""
return datetime.datetime(year, month, day, hour, minute, second).isoformat()
return datetime.datetime(year, month, day, hour, minute, second)
def DateFromTicks(ticks):
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment