More advanced topics ==================== .. sectionauthor:: Daniele Varrazzo .. testsetup:: * import re import select cur.execute("CREATE TABLE atable (apoint point)") conn.commit() def wait(conn): while 1: state = conn.poll() if state == psycopg2.extensions.POLL_OK: break elif state == psycopg2.extensions.POLL_WRITE: select.select([], [conn.fileno()], []) elif state == psycopg2.extensions.POLL_READ: select.select([conn.fileno()], [], []) else: raise psycopg2.OperationalError("poll() returned %s" % state) aconn = psycopg2.connect(database='test', async=1) wait(aconn) acurs = aconn.cursor() .. index:: double: Subclassing; Cursor double: Subclassing; Connection .. _subclassing-connection: .. _subclassing-cursor: Connection and cursor factories ------------------------------- Psycopg exposes two new-style classes that can be sub-classed and expanded to adapt them to the needs of the programmer: `psycopg2.extensions.cursor` and `psycopg2.extensions.connection`. The `connection` class is usually sub-classed only to provide an easy way to create customized cursors but other uses are possible. `cursor` is much more interesting, because it is the class where query building, execution and result type-casting into Python variables happens. .. index:: single: Example; Cursor subclass An example of cursor subclass performing logging is:: import psycopg2 import psycopg2.extensions import logging class LoggingCursor(psycopg2.extensions.cursor): def execute(self, sql, args=None): logger = logging.getLogger('sql_debug') logger.info(self.mogrify(sql, args)) try: psycopg2.extensions.cursor.execute(self, sql, args) except Exception, exc: logger.error("%s: %s" % (exc.__class__.__name__, exc)) raise conn = psycopg2.connect(DSN) cur = conn.cursor(cursor_factory=LoggingCursor) cur.execute("INSERT INTO mytable VALUES (%s, %s, %s);", (10, 20, 30)) .. index:: single: Objects; Creating new adapters single: Adaptation; Creating new adapters single: Data types; Creating new adapters .. _adapting-new-types: Adapting new Python types to SQL syntax --------------------------------------- Any Python class or type can be adapted to an SQL string. Adaptation mechanism is similar to the Object Adaptation proposed in the :pep:`246` and is exposed by the `psycopg2.extensions.adapt()` function. The `~cursor.execute()` method adapts its arguments to the `~psycopg2.extensions.ISQLQuote` protocol. Objects that conform to this protocol expose a `!getquoted()` method returning the SQL representation of the object as a string. The easiest way to adapt an object to an SQL string is to register an adapter function via the `~psycopg2.extensions.register_adapter()` function. The adapter function must take the value to be adapted as argument and return a conform object. A convenient object is the `~psycopg2.extensions.AsIs` wrapper, whose `!getquoted()` result is simply the `!str()`\ ing conversion of the wrapped object. .. index:: single: Example; Types adaptation Example: mapping of a `!Point` class into the |point|_ PostgreSQL geometric type: .. doctest:: >>> from psycopg2.extensions import adapt, register_adapter, AsIs >>> class Point(object): ... def __init__(self, x, y): ... self.x = x ... self.y = y >>> def adapt_point(point): ... return AsIs("'(%s, %s)'" % (adapt(point.x), adapt(point.y))) >>> register_adapter(Point, adapt_point) >>> cur.execute("INSERT INTO atable (apoint) VALUES (%s)", ... (Point(1.23, 4.56),)) .. |point| replace:: :sql:`point` .. _point: http://www.postgresql.org/docs/8.4/static/datatype-geometric.html#AEN6084 The above function call results in the SQL command:: INSERT INTO atable (apoint) VALUES ((1.23, 4.56)); .. index:: Type casting .. _type-casting-from-sql-to-python: Type casting of SQL types into Python objects --------------------------------------------- PostgreSQL objects read from the database can be adapted to Python objects through an user-defined adapting function. An adapter function takes two arguments: the object string representation as returned by PostgreSQL and the cursor currently being read, and should return a new Python object. For example, the following function parses the PostgreSQL :sql:`point` representation into the previously defined `!Point` class: >>> def cast_point(value, cur): ... if value is None: ... return None ... ... # Convert from (f1, f2) syntax using a regular expression. ... m = re.match(r"\(([^)]+),([^)]+)\)", value) ... if m: ... return Point(float(m.group(1)), float(m.group(2))) ... else: ... raise InterfaceError("bad point representation: %r" % value) In order to create a mapping from a PostgreSQL type (either standard or user-defined), its OID must be known. It can be retrieved either by the second column of the `cursor.description`: >>> cur.execute("SELECT NULL::point") >>> point_oid = cur.description[0][1] >>> point_oid 600 or by querying the system catalog for the type name and namespace (the namespace for system objects is :sql:`pg_catalog`): >>> cur.execute(""" ... SELECT pg_type.oid ... FROM pg_type JOIN pg_namespace ... ON typnamespace = pg_namespace.oid ... WHERE typname = %(typename)s ... AND nspname = %(namespace)s""", ... {'typename': 'point', 'namespace': 'pg_catalog'}) >>> point_oid = cur.fetchone()[0] >>> point_oid 600 After you know the object OID, you can create and register the new type: >>> POINT = psycopg2.extensions.new_type((point_oid,), "POINT", cast_point) >>> psycopg2.extensions.register_type(POINT) The `~psycopg2.extensions.new_type()` function binds the object OIDs (more than one can be specified) to the adapter function. `~psycopg2.extensions.register_type()` completes the spell. Conversion is automatically performed when a column whose type is a registered OID is read: >>> cur.execute("SELECT '(10.2,20.3)'::point") >>> point = cur.fetchone()[0] >>> print type(point), point.x, point.y 10.2 20.3 .. index:: pair: Asynchronous; Notifications pair: LISTEN; SQL command pair: NOTIFY; SQL command .. _async-notify: Asynchronous notifications -------------------------- Psycopg allows asynchronous interaction with other database sessions using the facilities offered by PostgreSQL commands |LISTEN|_ and |NOTIFY|_. Please refer to the PostgreSQL documentation for examples of how to use this form of communications. Notifications received are made available in the `connection.notifies` list. Notifications can be sent from Python code simply using a :sql:`NOTIFY` command in an `~cursor.execute()` call. Because of the way sessions interact with notifications (see |NOTIFY|_ documentation), you should keep the connection in :ref:`autocommit ` mode if you wish to receive or send notifications in a timely manner. .. |LISTEN| replace:: :sql:`LISTEN` .. _LISTEN: http://www.postgresql.org/docs/8.4/static/sql-listen.html .. |NOTIFY| replace:: :sql:`NOTIFY` .. _NOTIFY: http://www.postgresql.org/docs/8.4/static/sql-notify.html Notification are received after every query execution. If the user is interested in receiveing notification but not in performing any query, the `~connection.poll()` method can be used to check for notification without wasting resources. A simple application could poll the connection from time to time to check if something new has arrived. A better strategy is to use some I/O completion function such as |select()|_ to sleep until awaken from the kernel when there is some data to read on the connection, thereby using no CPU unless there is something to read:: import select import psycopg2 import psycopg2.extensions conn = psycopg2.connect(DSN) conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT) curs = conn.cursor() curs.execute("LISTEN test;") print "Waiting for 'NOTIFY test'" while 1: if select.select([conn],[],[],5) == ([],[],[]): print "Timeout" else: conn.poll() while conn.notifies: print "Got NOTIFY:", conn.notifies.pop() Running the script and executing the command :sql:`NOTIFY test` in a separate :program:`psql` shell, the output may look similar to:: Waiting for 'NOTIFY test' Timeout Timeout Got NOTIFY: (6535, 'test') Timeout ... .. index:: double: Asynchronous; Connection .. _async-support: Asynchronous support -------------------- .. versionadded:: 2.2.0 Psycopg can issue asynchronous queries to a PostgreSQL database. An asynchronous communication style is established passing the parameter *async*\=1 to the `~psycopg2.connect()` function: the returned connection will work in *asynchronous mode*. In asynchronous mode, a Psycopg connection will rely on the caller to poll the socket file descriptor, checking if it is ready to accept data or if a query result has been transferred and is ready to be read on the client. The caller can use the method `~connection.fileno()` to get the connection file descriptor and `~connection.poll()` to make communication proceed according to the current connection state. The following is an example loop using methods `!fileno()` and `!poll()` together with the Python |select()|_ function in order to carry on asynchronous operations with Psycopg:: def wait(conn): while 1: state = conn.poll() if state == psycopg2.extensions.POLL_OK: break elif state == psycopg2.extensions.POLL_WRITE: select.select([], [conn.fileno()], []) elif state == psycopg2.extensions.POLL_READ: select.select([conn.fileno()], [], []) else: raise psycopg2.OperationalError("poll() returned %s" % state) .. |select()| replace:: `!select()` .. _select(): http://docs.python.org/library/select.html#select.select The above loop of course would block an entire application: in a real asynchronous framework, `!select()` would be called on many file descriptors waiting for any of them to be ready. Nonetheless the function can be used to connect to a PostgreSQL server only using nonblocking commands and the connection obtained can be used to perform further nonblocking queries. After `!poll()` has returned `~psycopg2.extensions.POLL_OK`, and thus `!wait()` has returned, the connection can be safely used: >>> aconn = psycopg2.connect(database='test', async=1) >>> wait(aconn) >>> acurs = aconn.cursor() Notice that there are a few other requirements to be met in order to have a completely non-blocking connection attempt: see the libpq documentation for |PQconnectStart|_. .. |PQconnectStart| replace:: `!PQconnectStart()` .. _PQconnectStart: http://www.postgresql.org/docs/8.4/static/libpq-connect.html#AEN33199 The same loop should be also used to perform nonblocking queries: after sending a query via `~cursor.execute()` or `~cursor.callproc()`, call `!poll()` on the connection available from `cursor.connection` until it returns `!POLL_OK`, at which pont the query has been completely sent to the server and, if it produced data, the results have been transferred to the client and available using the regular cursor methods: >>> acurs.execute("SELECT pg_sleep(5); SELECT 42;") >>> wait(acurs.connection) >>> acurs.fetchone()[0] 42 When an asynchronous query is being executed, `connection.isexecuting()` returns `True`. Two cursors can't execute concurrent queries on the same asynchronous connection. There are several limitations in using asynchronous connections: the connection is always in :ref:`autocommit ` mode and it is not possible to change it using `~connection.set_isolation_level()`. So a transaction is not implicitly started at the first query and is not possible to use methods `~connection.commit()` and `~connection.rollback()`: you can manually control transactions using `~cursor.execute()` to send database commands such as :sql:`BEGIN`, :sql:`COMMIT` and :sql:`ROLLBACK`. With asynchronous connections it is also not possible to use `~connection.set_client_encoding()`, `~cursor.executemany()`, :ref:`large objects `, :ref:`named cursors `. :ref:`COPY commands ` are not supported either in asynchronous mode, but this will be probably implemented in a future release. .. index:: single: Greenlet, Coroutine, eventlet, gevent, Wait callback .. _green-support: Support to coroutine libraries ------------------------------ .. versionadded:: 2.2.0 Psycopg can be used together with coroutine_\-based libraries, and participate to cooperative multithread. Coroutine-based libraries (such as Eventlet_ or gevent_) can usually patch the Python standard library in order to enable a coroutine switch in presence of blocking I/O: the process is usually referred as making the system *green*, in reference to greenlet_, the basic Python micro-thread library. Because Psycopg is a C extension module, it is not possible for coroutine libraries to patch it: Psycopg instead enables cooperative multithreading by allowing the registration of a *wait callback* using the `psycopg2.extensions.set_wait_callback()` function. When a wait callback is registered, Psycopg will use `libpq non-blocking calls`__ instead of the regular blocking ones, and will delegate to the callback the responsibility to wait for available data. This way of working is less flexible of complete asynchronous I/O, but has the advantage of maintaining a complete |DBAPI| semantics: from the point of view of the end user, all Psycopg functions and objects will work transparently in the coroutine environment (the calling coroutine will be blocked while other coroutines can be scheduled to run), allowing non modified code and third party libraries (such as SQLAlchemy_) to be used in coroutine-based programs. Notice that, while I/O correctly yields control to other coroutines, each connection has a lock allowing a single cursor at time to communicate with the backend: such lock is not *green*, so blocking against it would block the entire program waiting for data, not the single coroutine. Therefore, programmers are advised to either avoid to share connections between coroutines or to use a library-friendly lock to synchronize shares connections, e.g. for pooling. Coroutine libraries authors should provide a callback implementation (and probably register it) to make Psycopg as green as they want. An example callback (using `!select()` to block) is provided as `psycopg2.extras.wait_select()`: it boils down to something similar to:: def wait_select(conn): while 1: state = conn.poll() if state == extensions.POLL_OK: break elif state == extensions.POLL_READ: select.select([conn.fileno()], [], []) elif state == extensions.POLL_WRITE: select.select([], [conn.fileno()], []) else: raise OperationalError("bad state from poll: %s" % state) .. _coroutine: http://en.wikipedia.org/wiki/Coroutine .. _greenlet: http://pypi.python.org/pypi/greenlet .. _Eventlet: http://eventlet.net/ .. _gevent: http://www.gevent.org/ .. _SQLAlchemy: http://www.sqlalchemy.org/ .. __: http://www.postgresql.org/docs/8.4/static/libpq-async.html .. testcode:: :hide: aconn.close() conn.rollback() cur.execute("DROP TABLE atable") conn.commit() cur.close() conn.close()