Merge 5e1d353 into f31b16e

CHANGES.rst

@@ -5,7 +5,8 @@
 3.0a12 (unreleased)
 ===================
 
-- Nothing changed yet.
+- Add the ``gevent psycopg2`` driver to allow using the fast psycopg2
+  driver with gevent.
 
 
 3.0a11 (2019-09-25)

docs/db-specific-options.rst

@@ -28,7 +28,9 @@ PostgreSQL Adapter Options
 ==========================
 
 RelStorage 3.0 requires PostgreSQL 9.6 or above. PostgreSQL 12,
-currently in development, is not supported.
+currently in development, is **not** supported and attempting to use
+it is likely to result in errors locking objects and database
+corruption.
 
 .. tip::
 
@@ -48,7 +50,8 @@ The PostgreSQL adapter accepts:
 
 driver
     All of these drivers use the name of the corresponding PyPI
-    package. The possible options are:
+    package. All drivers support uploading objects using PostgreSQL's
+    fast binary COPY protocol (except where noted). The possible options are:
 
     psycopg2
       A C-based driver that requires the PostgreSQL development
@@ -58,6 +61,16 @@ driver
       without `needing a C compiler
       <http://initd.org/psycopg/docs/install.html#binary-packages>`_.
 
+    gevent psycopg2
+      The same as ``psycopg2``, but made to be gevent compatible
+      through the use of a wait callback. If the system is
+      monkey-patched by gevent, RelStorage will automatically install
+      a wait callback (which is global to all connections created by
+      psycopg2). Otherwise, you must install a compatible wait
+      callback (perhaps using `psycogreen
+      <https://pypi.org/project/psycogreen/>`__). This driver forfeits
+      use of the COPY protocol and use of the C-accelerated BLOBs.
+
     psycopg2cffi
       A C-based driver that requires the PostgreSQL development
       libraries. Optimal on PyPy and almost indistinguishable from

docs/install.rst

@@ -88,11 +88,15 @@ bold** are the recommended adapter installed with the extra.
    |          | 2. MySQL Connector  | 2.  pg8000          |              |
    +----------+---------------------+---------------------+--------------+
 
+gevent
+------
 
 mysqlclient can be used with gevent by explicitly choosing a
-gevent-aware driver. PyMySQL, MySQL Connector/Python (without its C
-extension), and pg8000 are compatible (cooperative) with gevent
-when the system is monkey-patched.
+gevent-aware driver; so can psycopg2 if a wait callback (such as
+``psycogreen``) is installed; RelStorage will install one
+automatically if the system is monkey-patched. PyMySQL, MySQL
+Connector/Python (without its C extension), and pg8000 are compatible
+(cooperative) with gevent when the system is monkey-patched.
 
 For additional details and warnings, see the "driver" section for each database in
 :doc:`db-specific-options`.

setup.py

@@ -212,6 +212,9 @@ def read_file(*path):
             'postgres = relstorage.zodburi_resolver:postgresql_resolver',
             'mysql = relstorage.zodburi_resolver:mysql_resolver',
             'oracle = relstorage.zodburi_resolver:oracle_resolver'
-        ]
+        ],
+        'gevent.plugins.monkey.did_patch_builtins': [
+            'psycopg2 = relstorage.adapters.postgresql.drivers.psycopg2:_gevent_did_patch',
+        ],
     },
 )

src/relstorage/adapters/postgresql/drivers/__init__.py

@@ -38,6 +38,9 @@ class AbstractPostgreSQLDriver(AbstractModuleDriver):
     # "SELECT 1; COMMIT;"
     supports_multiple_statement_execute = True
 
+    # Can we use the COPY command (copy_export)?
+    supports_copy = True
+
     def connect_with_isolation(self, dsn,
                                isolation=None,
                                read_only=False,

src/relstorage/adapters/postgresql/drivers/_lobject.py

@@ -0,0 +1,124 @@
+# -*- coding: utf-8 -*-
+##############################################################################
+#
+# Copyright (c) 2019 Zope Foundation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.1 (ZPL).  A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE.
+#
+##############################################################################
+"""
+psycpg2-like lobject support using the SQL interface,
+for drivers and modes that don't support it natively.
+
+For example, pg8000 provides no LOB support, and psycopg2 when in
+asynchronous mode has no LOB support. This package provides just enough
+functionality for our purposes. Activate it by deriving a class from
+the driver's Connection class and this module's :class:`LobConnectionMixin`.
+"""
+
+from __future__ import absolute_import
+from __future__ import print_function
+from __future__ import division
+
+__all__ = [
+    'LobConnectionMixin',
+]
+
+
+class _WriteBlob(object):
+    closed = False
+
+    def __init__(self, conn, binary):
+        self._binary = binary
+        self._cursor = conn.cursor()
+        self._offset = 0
+        try:
+            self._cursor.execute("SELECT lo_creat(-1)")
+            row = self._cursor.fetchone()
+            self.oid = row[0]
+        except:
+            self._cursor.close()
+            raise
+
+    def close(self):
+        self._cursor.close()
+        self.closed = True
+
+    def write(self, data):
+        self._cursor.execute("SELECT lo_put(%(oid)s, %(off)s, %(data)s)",
+                             {'oid': self.oid, 'off': self._offset, 'data': self._binary(data)})
+        self._offset += len(data)
+        return len(data)
+
+class _UploadBlob(object):
+    closed = False
+    fetch_size = 1024 * 1024 * 9
+
+    def __init__(self, conn, new_file, binary):
+        blob = _WriteBlob(conn, binary)
+        self.oid = blob.oid
+        try:
+            with open(new_file, 'rb') as f:
+                while 1:
+                    data = f.read(self.fetch_size)
+                    if not data:
+                        break
+                    blob.write(data)
+        finally:
+            blob.close()
+
+    def close(self):
+        self.closed = True
+
+class _ReadBlob(object):
+    closed = False
+    fetch_size = 1024 * 1024 * 9
+
+    def __init__(self, conn, oid):
+        self._cursor = conn.cursor()
+        self.oid = oid
+        self.offset = 0
+
+    def export(self, filename):
+        with open(filename, 'wb') as f:
+            while 1:
+                data = self.read(self.fetch_size)
+                if not data:
+                    break
+                f.write(data)
+        self.close()
+
+    def read(self, size):
+        self._cursor.execute("SELECT lo_get(%(oid)s, %(off)s, %(cnt)s)",
+                             {'oid': self.oid, 'off': self.offset, 'cnt': size})
+        row = self._cursor.fetchone()
+        data = row[0]
+        self.offset += len(data)
+        return data
+
+    def close(self):
+        self._cursor.close()
+        self.closed = True
+
+
+
+class LobConnectionMixin(object):
+
+    #: The driver module's ``Binary`` converter object.
+    RSDriverBinary = None
+
+    def lobject(self, oid=0, mode='', new_oid=0, new_file=None):
+        if oid == 0 and new_oid == 0 and mode == 'wb':
+            if new_file:
+                # Upload the whole file right now.
+                return _UploadBlob(self, new_file, self.RSDriverBinary)
+            return _WriteBlob(self, self.RSDriverBinary)
+        if oid != 0 and mode == 'rb':
+            return _ReadBlob(self, oid)
+        raise AssertionError("Unsupported params", dict(locals()))