Changeset 852 for pg8000/trunk

Timestamp:

03/14/07 11:04:51 (2 years ago)

Author:

mfenniak

Message:

add date and time data types, and timezonetz read. Add bool write -- somehow this was missed.

Files:

• pg8000/trunk/pg8000.py (modified) (12 diffs)
• pg8000/trunk/pg8000_test.py (modified) (4 diffs)

pg8000/trunk/pg8000.py

@@ -38 +38 @@
-import time
-
+debug_log = file("/Users/mfenniak/SQLAlchemy-0.3.5/pg8000_debug.log", "w")
+
-class Warning(StandardError):
-    pass
@@ -106 +108 @@
-        #  2 -- inside quoted identifier   "..."
-        #  3 -- inside escaped single-quote string, E'...'
+        debug_log.write("convert_paramstyle(%r, %r, %r)\n" % (src_style, query, args))
-        state = 0
-        output_query = ""
@@ -289 +292 @@
-
-        def execute(self, operation, args=()):
+            debug_log.write("execute(%r, %r)\n" % (operation, args))
-            if self.cursor == None:
-                raise InterfaceError("cursor is closed")
@@ -1475 +1479 @@
-        return data == "\x01"
-
+    def boolout(v, **kwargs):
+        if v:
+            return 't'
+        else:
+            return 'f'
+
-    def int2recv(data, **kwargs):
-        return struct.unpack("!h", data)[0]
@@ -1529 +1539 @@
-        return v.isoformat(' ')
-
+    def date_in(data, **kwargs):
+        year = int(data[0:4])
+        month = int(data[5:7])
+        day = int(data[8:10])
+        return datetime.date(year, month, day)
+
+    def date_out(v, **kwargs):
+        return v.isoformat()
+
+    def time_in(data, **kwargs):
+        hour = int(data[0:2])
+        minute = int(data[3:5])
+        sec = decimal.Decimal(data[6:])
+        return datetime.time(hour, minute, int(sec), int((sec - int(sec)) * 1000000))
+
+    def time_out(v, **kwargs):
+        return v.isoformat()
+
-    def numeric_in(data, **kwargs):
-        if data.find(".") == -1:
@@ -1544 +1572 @@
-        return v.encode(client_encoding)
-
-description
+**kwargs
-        year = int(data[0:4])
-        month = int(data[5:7])
@@ -1553 +1581 @@
-        sec = decimal.Decimal(data[17:tz_sep])
-        tz = data[tz_sep:]
-        print repr(data), repr(description)
-        print repr(tz)
-        return datetime.datetime(year, month, day, hour, minute, int(sec), int((sec - int(sec)) * 1000000), Types.FixedOffsetTz(tz))
-
@@ -1571 +1597 @@
-            return datetime.timedelta(0)
-
+        def __eq__(self, other):
+            if not isinstance(other, FixedOffsetTz):
+                return False
+            return self.hrs == other.hrs
+
-    def byteasend(v, **kwargs):
-        return str(v)
@@ -1582 +1613 @@
-
-    py_types = {
+        bool: {"tid": 16, "txt_out": boolout},
-        int: {"tid": 1700, "txt_out": numeric_out},
-        long: {"tid": 1700, "txt_out": numeric_out},
@@ -1590 +1622 @@
-        Bytea: {"tid": 17, "bin_out": byteasend},
-        datetime.datetime: {"tid": 1114, "txt_out": timestamp_out},
+        datetime.date: {"tid": 1082, "txt_out": date_out},
+        datetime.time: {"tid": 1083, "txt_out": time_out},
-        type(None): {"tid": -1},
-    }
@@ -1596 +1630 @@
-        16: {"txt_in": boolin, "bin_in": boolrecv, "prefer": "bin"},
-        17: {"bin_in": bytearecv},
+        19: {"txt_in": varcharin}, # name type
-        20: {"txt_in": int8in, "bin_in": int8recv, "prefer": "bin"},
-        21: {"txt_in": int2in, "bin_in": int2recv, "prefer": "bin"},
@@ -1605 +1640 @@
-        1042: {"txt_in": varcharin}, # CHAR type
-        1043: {"txt_in": varcharin}, # VARCHAR type
-
+
+
+
+
-        1186: {"txt_in": interval_in},
-        1700: {"txt_in": numeric_in},
-    }
-
-
-
-
+
+
+

pg8000/trunk/pg8000_test.py

@@ -8 +8 @@
-import struct
-
-
+joy_
-        "host": "joy",
-        "user": "pg8000-test",
@@ -16 +16 @@
-        "ssl": False,
-        }
+db_local_connect = {
+        "unix_sock": "/tmp/.s.PGSQL.5432",
+        "user": "mfenniak"
+        }
+db_connect = db_local_connect
-db = pg8000.Connection(**db_connect)
-dbapi = pg8000.DBAPI
@@ -264 +269 @@
-# Tests relating to type conversion.
-class TypeTests(unittest.TestCase):
+    def TestTimeRoundtrip(self):
+        db.execute("SELECT $1 as f1", datetime.time(4, 5, 6))
+        retval = tuple(db.iterate_dict())
+        self.assert_(retval == ({"f1": datetime.time(4, 5, 6)},),
+                "retrieved value match failed")
+
+    def TestDateRoundtrip(self):
+        db.execute("SELECT $1 as f1", datetime.date(2001, 2, 3))
+        retval = tuple(db.iterate_dict())
+        self.assert_(retval == ({"f1": datetime.date(2001, 2, 3)},),
+                "retrieved value match failed")
+
+    def TestBoolRoundtrip(self):
+        db.execute("SELECT $1 as f1", True)
+        retval = tuple(db.iterate_dict())
+        self.assert_(retval == ({"f1": True},),
+                "retrieved value match failed")
+
-    def TestNullRoundtrip(self):
-        # We can't just "SELECT $1" and set None as the parameter, since it has
@@ -338 +361 @@
-                "retrieved value match failed")
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-        db.execute("SELECT oid FROM pg_type")
-        retval = tuple(db.iterate_dict())