Changeset 799

Timestamp:

03/07/07 19:33:41 (2 years ago)

Author:

mfenniak

Message:

working portal based cursor queries

Files:

• pg8000/trunk/pg8000.py (modified) (14 diffs)
• pg8000/trunk/pg8000-test.py (modified) (1 diff)

pg8000/trunk/pg8000.py

@@ -70 +70 @@
-        self.connection = connection
-        if self.connection.iterate_dicts:
-onnection
+ursor
-        else:
-onnection
+ursor
-
-    def __iter__(self):
@@ -83 +83 @@
-        return retval
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+
-    ##
-    # A configuration variable that determines whether iterating over the
@@ -123 +93 @@
-    iterate_dicts = False
-
-
+
+
+
+
+
+
+
-        self._row_desc = None
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
-
-    def _fetch(self):
-
-
-
+
+
+
+
+
+
+
+
+
-        return tuple([Types.py_value(row.fields[i], self._row_desc.fields[i]) for i in range(len(row.fields))])
-
@@ -182 +158 @@
-        return DataIterator(self)
-
+##
+# This class represents a connection to a PostgreSQL database.
+# <p>
+# A single PostgreSQL connection can only perform a single query at a time,
+# which is an important restriction to note.  This limitation can be overcome
+# by retrieving all results immediately after a query, but this approach is not
+# taken by this library.
+# <p>
+# Stability: Added in v1.00, stability guaranteed for v1.xx.
+#
+# @param host   The hostname of the PostgreSQL server to connect with.  Only
+# TCP/IP connections are presently supported, so this parameter is mandatory.
+#
+# @param user   The username to connect to the PostgreSQL server with.  This
+# parameter is mandatory.
+#
+# @param port   The TCP/IP port of the PostgreSQL server instance.  This
+# parameter defaults to 5432, the registered and common port of PostgreSQL
+# TCP/IP servers.
+#
+# @param database   The name of the database instance to connect with.  This
+# parameter is optional, if omitted the PostgreSQL server will assume the
+# database name is the same as the username.
+#
+# @param password   The user password to connect to the server with.  This
+# parameter is optional.  If omitted, and the database server requests password
+# based authentication, the connection will fail.  On the other hand, if this
+# parameter is provided and the database does not request password
+# authentication, then the password will not be used.
+class Connection(object):
+    def __init__(self, host, user, port=5432, database=None, password=None):
+        self._row_desc = None
+        try:
+            self.c = Protocol.Connection(host, port)
+            self.c.connect()
+            self.c.authenticate(user, password=password, database=database)
+        except socket.error, e:
+            raise InterfaceError("communication error", e)
+
-    def begin(self):
-pass
+raise NotSupportedError("uncoded")
-
-    def commit(self):
-pass
+raise NotSupportedError("uncoded")
-
-    def rollback(self):
-pass
+raise NotSupportedError("uncoded")
-
-
@@ -264 +279 @@
-            return val
-
+    class Close(object):
+        def __init__(self, typ, name):
+            if len(typ) != 1:
+                raise InternalError("Close typ must be 1 char")
+            self.typ = typ
+            self.name = name
+
+        def serialize(self):
+            val = self.typ + self.name + "\x00"
+            val = struct.pack("!i", len(val) + 4) + val
+            val = "C" + val
+            return val
+
+    class ClosePortal(Close):
+        def __init__(self, name):
+            Protocol.Close.__init__(self, "P", name)
+
+    class ClosePreparedStatement(Close):
+        def __init__(self, name):
+            Protocol.Close.__init__(self, "S", name)
+
-    class Describe(object):
-        def __init__(self, typ, name):
@@ -289 +325 @@
-            return 'H\x00\x00\x00\x04'
-
+    class Sync(object):
+        def serialize(self):
+            return 'S\x00\x00\x00\x04'
+
-    class PasswordMessage(object):
-        def __init__(self, pwd):
@@ -297 +337 @@
-            val = struct.pack("!i", len(val) + 4) + val
-            val = "p" + val
+            return val
+
+    class Execute(object):
+        def __init__(self, portal, row_count):
+            self.portal = portal
+            self.row_count = row_count
+
+        def serialize(self):
+            val = self.portal + "\x00" + struct.pack("!i", self.row_count)
+            val = struct.pack("!i", len(val) + 4) + val
+            val = "E" + val
-            return val
-
@@ -373 +424 @@
-        def createFromData(data):
-            return Protocol.BindComplete()
+        createFromData = staticmethod(createFromData)
+
+    class CloseComplete(object):
+        def createFromData(data):
+            return Protocol.CloseComplete()
+        createFromData = staticmethod(createFromData)
+
+    class PortalSuspended(object):
+        def createFromData(data):
+            return Protocol.PortalSuspended()
-        createFromData = staticmethod(createFromData)
-
@@ -528 +589 @@
-                    raise InternalError("Unexpected response msg %r" % (msg))
-
+        def fetch_rows(self, portal, row_count):
+            self.verifyState("ready")
+            self._send(Protocol.Execute(portal, row_count))
+            self._send(Protocol.Flush())
+            rows = []
+            end_of_data = False
+            while 1:
+                msg = self._read_message()
+                if isinstance(msg, Protocol.DataRow):
+                    rows.append(msg)
+                elif isinstance(msg, Protocol.PortalSuspended):
+                    # got all the rows we asked for, but not all that exist
+                    break
+                elif isinstance(msg, Protocol.CommandComplete):
+                    self._send(Protocol.ClosePortal(portal))
+                    self._send(Protocol.Sync())
+                    self._waitForReady()
+                    end_of_data = True
+                    break
+                elif isinstance(msg, Protocol.ErrorResponse):
+                    raise msg.createException()
+                else:
+                    raise InternalError("Unexpected response msg %r" % msg)
+            return end_of_data, rows
+
-        def query(self, qs):
-            self.verifyState("ready")
@@ -561 +647 @@
-        "1": ParseComplete,
-        "2": BindComplete,
+        "3": CloseComplete,
+        "s": PortalSuspended,
-        }
-
@@ -591 +679 @@
-        format = description['format']
-        funcs = Types.pg_types.get(type_oid)
-
+s
-            raise NotSupportedError("data response type %r not supported" % (type_oid))
-        func = funcs[format]
@@ -602 +690 @@
-        return data == 't'
-
+    def boolrecv(data, description):
+        return data == "\x01"
+
+    def int2recv(data, description):
+        return struct.unpack("!h", data)[0]
+
+    def int4recv(data, description):
+        return struct.unpack("!i", data)[0]
+
-    def int4in(data, description):
-        return int(data)
@@ -607 +704 @@
-    def int4send(v):
-        return struct.pack("!i", v)
+
+    def timestamp_recv(data, description):
+        val = struct.unpack("!d", data)[0]
+        return datetime.datetime(2000, 1, 1) + datetime.timedelta(seconds = val)
-
-    def timestamp_in(data, description):
@@ -622 +723 @@
-
-    pg_types = {
-
-
-
+
+
+
+
-    }
-

pg8000/trunk/pg8000-test.py

@@ -4 +4 @@
-
-db = pg8000.Connection(host='localhost', user='mfenniak')
-
+
+
-
-print "begin query..."
-
-
+
+
+
+
+
+
+
+
+
+
-    print repr(row)
-print "end query..."