psycopg/psycopg2
1
1
Fork 0
mirror of https://github.com/psycopg/psycopg2.git synced 2024-11-11 03:26:37 +03:00
Code Issues Packages Projects Releases Wiki Activity

Explicit cast of the SQL representation of time-related objects.

Browse Source 
Allow the objects to be recognized as the proper type by Postgres in not
strong contexts: problem reported by Peter Eisentraut.

Added tests to check the types are respected in a complete Py -> PG ->
Py roundtrip without context.
This commit is contained in:
Daniele Varrazzo 2010-05-04 01:52:15 +01:00
parent e8c2a14362
commit bf1c76b792
4 changed files with 118 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
ChangeLog
View File

@ -2,6 +2,9 @@
* Added typecasters for arrays of specific MX/Py time-related types.
* psycopg/adapter_[mx]datetime.c: Explicit cast of the SQL representation
of time-related objects.
2010-05-03 Daniele Varrazzo <daniele.varrazzo@gmail.com>
* psycopg/adapter_binary.c: Adapt buffer objects using an explicit cast on

19
psycopg/adapter_datetime.c
View File

@ -57,10 +57,25 @@ static PyObject *
pydatetime_str(pydatetimeObject *self)
{
if (self->type <= PSYCO_DATETIME_TIMESTAMP) {
/* Select the right PG type to cast into. */
char *fmt = NULL;
switch (self->type) {
case PSYCO_DATETIME_TIME:
fmt = "'%s'::time";
break;
case PSYCO_DATETIME_DATE:
fmt = "'%s'::date";
break;
case PSYCO_DATETIME_TIMESTAMP:
fmt = "'%s'::timestamp";
break;
}
PyObject *res = NULL;
PyObject *iso = PyObject_CallMethod(self->wrapped, "isoformat", NULL);
if (iso) {
res = PyString_FromFormat("'%s'", PyString_AsString(iso));
res = PyString_FromFormat(fmt, PyString_AsString(iso));
Py_DECREF(iso);
}
return res;
@ -78,7 +93,7 @@ pydatetime_str(pydatetimeObject *self)
}
buffer[6] = '\0';
return PyString_FromFormat("'%d days %d.%s seconds'",
return PyString_FromFormat("'%d days %d.%s seconds'::interval",
obj->days, obj->seconds, buffer);
}
}

14
psycopg/adapter_mxdatetime.c
View File

@ -56,10 +56,10 @@ mxdatetime_str(mxdatetimeObject *self)
case PSYCO_MXDATETIME_DATE:
dt = (mxDateTimeObject *)self->wrapped;
if (dt->year >= 1)
PyOS_snprintf(buf, sizeof(buf) - 1, "'%04ld-%02d-%02d'",
PyOS_snprintf(buf, sizeof(buf) - 1, "'%04ld-%02d-%02d'::date",
dt->year, (int)dt->month, (int)dt->day);
else
PyOS_snprintf(buf, sizeof(buf) - 1, "'%04ld-%02d-%02d BC'",
PyOS_snprintf(buf, sizeof(buf) - 1, "'%04ld-%02d-%02d BC'::date",
1 - dt->year, (int)dt->month, (int)dt->day);
break;
@ -67,12 +67,12 @@ mxdatetime_str(mxdatetimeObject *self)
dt = (mxDateTimeObject *)self->wrapped;
if (dt->year >= 1)
PyOS_snprintf(buf, sizeof(buf) - 1,
"'%04ld-%02d-%02dT%02d:%02d:%09.6f'",
"'%04ld-%02d-%02dT%02d:%02d:%09.6f'::timestamp",
dt->year, (int)dt->month, (int)dt->day,
(int)dt->hour, (int)dt->minute, dt->second);
else
PyOS_snprintf(buf, sizeof(buf) - 1,
"'%04ld-%02d-%02dT%02d:%02d:%09.6f BC'",
"'%04ld-%02d-%02dT%02d:%02d:%09.6f BC'::timestamp",
1 - dt->year, (int)dt->month, (int)dt->day,
(int)dt->hour, (int)dt->minute, dt->second);
break;
@ -85,16 +85,16 @@ mxdatetime_str(mxdatetimeObject *self)
time */
dtd = (mxDateTimeDeltaObject *)self->wrapped;
if (0 <= dtd->seconds && dtd->seconds < 24*3600) {
PyOS_snprintf(buf, sizeof(buf) - 1, "'%02d:%02d:%09.6f'",
PyOS_snprintf(buf, sizeof(buf) - 1, "'%02d:%02d:%09.6f'::time",
(int)dtd->hour, (int)dtd->minute, dtd->second);
} else {
double ss = dtd->hour*3600.0 + dtd->minute*60.0 + dtd->second;
if (dtd->seconds >= 0)
PyOS_snprintf(buf, sizeof(buf) - 1, "'%ld days %.6f seconds'",
PyOS_snprintf(buf, sizeof(buf) - 1, "'%ld days %.6f seconds'::interval",
dtd->day, ss);
else
PyOS_snprintf(buf, sizeof(buf) - 1, "'-%ld days -%.6f seconds'",
PyOS_snprintf(buf, sizeof(buf) - 1, "'-%ld days -%.6f seconds'::interval",
dtd->day, ss);
}
break;

91
tests/test_dates.py
View File

@ -247,6 +247,47 @@ class DatetimeTests(unittest.TestCase, CommonDatetimeTestsMixin):
self.assertEqual(seconds, -3583504)
self.assertEqual(int(round((value - seconds) * 1000000)), 123456)
def _test_type_roundtrip(self, o1):
o2 = self.execute("select %s;", (o1,))
self.assertEqual(type(o1), type(o2))
def _test_type_roundtrip_array(self, o1):
o1 = [o1]
o2 = self.execute("select %s;", (o1,))
self.assertEqual(type(o1[0]), type(o2[0]))
def test_type_roundtrip_date(self):
from datetime import date
self._test_type_roundtrip(date(2010,05,03))
def test_type_roundtrip_datetime(self):
from datetime import datetime
self._test_type_roundtrip(datetime(2010,05,03,10,20,30))
def test_type_roundtrip_time(self):
from datetime import time
self._test_type_roundtrip(time(10,20,30))
def test_type_roundtrip_interval(self):
from datetime import timedelta
self._test_type_roundtrip(timedelta(seconds=30))
def test_type_roundtrip_date_array(self):
from datetime import date
self._test_type_roundtrip_array(date(2010,05,03))
def test_type_roundtrip_datetime_array(self):
from datetime import datetime
self._test_type_roundtrip_array(datetime(2010,05,03,10,20,30))
def test_type_roundtrip_time_array(self):
from datetime import time
self._test_type_roundtrip_array(time(10,20,30))
def test_type_roundtrip_interval_array(self):
from datetime import timedelta
self._test_type_roundtrip_array(timedelta(seconds=30))
# Only run the datetime tests if psycopg was compiled with support.
if not hasattr(psycopg2._psycopg, 'PYDATETIME'):
@ -264,6 +305,15 @@ class mxDateTimeTests(unittest.TestCase, CommonDatetimeTestsMixin):
self.DATETIME = psycopg2._psycopg.MXDATETIME
self.INTERVAL = psycopg2._psycopg.MXINTERVAL
psycopg2.extensions.register_type(self.DATE, self.conn)
psycopg2.extensions.register_type(self.TIME, self.conn)
psycopg2.extensions.register_type(self.DATETIME, self.conn)
psycopg2.extensions.register_type(self.INTERVAL, self.conn)
psycopg2.extensions.register_type(psycopg2.extensions.MXDATEARRAY, self.conn)
psycopg2.extensions.register_type(psycopg2.extensions.MXTIMEARRAY, self.conn)
psycopg2.extensions.register_type(psycopg2.extensions.MXDATETIMEARRAY, self.conn)
psycopg2.extensions.register_type(psycopg2.extensions.MXINTERVALARRAY, self.conn)
def tearDown(self):
self.conn.close()
@ -370,6 +420,47 @@ class mxDateTimeTests(unittest.TestCase, CommonDatetimeTestsMixin):
self.assertEqual(seconds, -3583504)
self.assertEqual(int(round((value - seconds) * 1000000)), 123456)
def _test_type_roundtrip(self, o1):
o2 = self.execute("select %s;", (o1,))
self.assertEqual(type(o1), type(o2))
def _test_type_roundtrip_array(self, o1):
o1 = [o1]
o2 = self.execute("select %s;", (o1,))
self.assertEqual(type(o1[0]), type(o2[0]))
def test_type_roundtrip_date(self):
from mx.DateTime import Date
self._test_type_roundtrip(Date(2010,05,03))
def test_type_roundtrip_datetime(self):
from mx.DateTime import DateTime
self._test_type_roundtrip(DateTime(2010,05,03,10,20,30))
def test_type_roundtrip_time(self):
from mx.DateTime import Time
self._test_type_roundtrip(Time(10,20,30))
def test_type_roundtrip_interval(self):
from mx.DateTime import DateTimeDeltaFrom
self._test_type_roundtrip(DateTimeDeltaFrom(seconds=30))
def test_type_roundtrip_date_array(self):
from mx.DateTime import Date
self._test_type_roundtrip_array(Date(2010,05,03))
def test_type_roundtrip_datetime_array(self):
from mx.DateTime import DateTime
self._test_type_roundtrip_array(DateTime(2010,05,03,10,20,30))
def test_type_roundtrip_time_array(self):
from mx.DateTime import Time
self._test_type_roundtrip_array(Time(10,20,30))
def test_type_roundtrip_interval_array(self):
from mx.DateTime import DateTimeDeltaFrom
self._test_type_roundtrip_array(DateTimeDeltaFrom(seconds=30))
# Only run the mx.DateTime tests if psycopg was compiled with support.
if not hasattr(psycopg2._psycopg, 'MXDATETIME'):