How to use the psycopg2.extensions function in psycopg2

def test_tpc_commit(self):
        cnn = self.connect()
        xid = cnn.xid(1, "gtrid", "bqual")
        self.assertEqual(cnn.status, ext.STATUS_READY)

        cnn.tpc_begin(xid)
        self.assertEqual(cnn.status, ext.STATUS_BEGIN)

        cur = cnn.cursor()
        cur.execute("insert into test_tpc values ('test_tpc_commit');")
        self.assertEqual(0, self.count_xacts())
        self.assertEqual(0, self.count_test_records())

        cnn.tpc_prepare()
        self.assertEqual(cnn.status, ext.STATUS_PREPARED)
        self.assertEqual(1, self.count_xacts())
        self.assertEqual(0, self.count_test_records())

        cnn.tpc_commit()
        self.assertEqual(cnn.status, ext.STATUS_READY)
        self.assertEqual(0, self.count_xacts())
        self.assertEqual(1, self.count_test_records())

config_manager = ConfigurationManager(
            [required_config],
            app_name='middleware',
            app_description=__doc__,
            values_source_list=[
                {'logger': mock_logging},
                environment,
            ],
            argv_source=[]
        )
        config = config_manager.get_config()
        self.conn = config.database.database_class(
            config.database
        ).connection()
        assert self.conn.get_transaction_status() == \
            psycopg2.extensions.TRANSACTION_STATUS_IDLE

def test_async_cancel(self):
        async_conn = psycopg2.connect(dsn, async_=True)
        self.assertRaises(psycopg2.OperationalError, async_conn.cancel)
        extras.wait_select(async_conn)
        cur = async_conn.cursor()
        cur.execute("select pg_sleep(10)")
        time.sleep(1)
        self.assertTrue(async_conn.isexecuting())
        async_conn.cancel()
        self.assertRaises(psycopg2.extensions.QueryCanceledError,
                          extras.wait_select, async_conn)
        cur.execute("select 1")
        extras.wait_select(async_conn)
        self.assertEqual(cur.fetchall(), [(1, )])

def _logtofile(self, msg, curs):
        msg = self.filter(msg, curs)
        if msg:
            if _sys.version_info[0] >= 3 and isinstance(msg, bytes):
                msg = msg.decode(_ext.encodings[self.encoding], 'replace')
            self._logobj.write(msg + _os.linesep)

if fd in self._futures and not self._futures[fd].done():
                self._futures[fd].set_exception(
                    psycopg2.OperationalError('Connection error (%s)' % error)
                )
        except (psycopg2.Error, psycopg2.Warning) as error:
            if fd in self._futures and not self._futures[fd].done():
                self._futures[fd].set_exception(error)
        else:
            if state == extensions.POLL_OK:
                if fd in self._futures and not self._futures[fd].done():
                    self._futures[fd].set_result(True)
            elif state == extensions.POLL_WRITE:
                self._ioloop.update_handler(fd, ioloop.IOLoop.WRITE)
            elif state == extensions.POLL_READ:
                self._ioloop.update_handler(fd, ioloop.IOLoop.READ)
            elif state == extensions.POLL_ERROR:
                self._ioloop.remove_handler(fd)
                if fd in self._futures and not self._futures[fd].done():
                    self._futures[fd].set_exception(
                        psycopg2.Error('Poll Error'))

import psycopg2
import psycopg2.extensions

class Portal(psycopg2.extensions.cursor):
    def __init__(self, name, curs):
        psycopg2.extensions.cursor.__init__(
            self, curs.connection, '"'+name+'"')
        
CURSOR = psycopg2.extensions.new_type((1790,), "CURSOR", Portal)
psycopg2.extensions.register_type(CURSOR)

conn = psycopg2.connect("dbname=test")

curs = conn.cursor()
curs.execute("SELECT reffunc2()")

portal = curs.fetchone()[0]
print portal.fetchone()
print portal.fetchmany(2)
portal.scroll(0, 'absolute')
print portal.fetchall()

*   the Free Software Foundation; either version 2 of the License, or     *
 *   (at your option) any later version.                                   *
 *                                                                         *
 ***************************************************************************/
"""

from PyQt4.QtCore import *
from PyQt4.QtGui import *

from ..connector import DBConnector
from ..plugin import ConnectionError, DbError, Table

import psycopg2
import psycopg2.extensions
# use unicode!
psycopg2.extensions.register_type(psycopg2.extensions.UNICODE)
psycopg2.extensions.register_type(psycopg2.extensions.UNICODEARRAY)


def classFactory():
	return PostGisDBConnector

class PostGisDBConnector(DBConnector):
	def __init__(self, uri):
		DBConnector.__init__(self, uri)

		self.host = uri.host()
		self.port = uri.port()
		self.dbname = uri.database()
		self.user = uri.username()
		self.passwd = uri.password()

def __init__(self, **keywords):
        if 'pw' in keywords:
            keywords['password'] = keywords.pop('pw')
            
        db_module = import_driver(["psycopg2", "psycopg", "pgdb"], preferred=keywords.pop('driver', None))
        if db_module.__name__ == "psycopg2":
            import psycopg2.extensions
            psycopg2.extensions.register_type(psycopg2.extensions.UNICODE)

        # if db is not provided postgres driver will take it from PGDATABASE environment variable
        if 'db' in keywords:
            keywords['database'] = keywords.pop('db')
        
        self.dbname = "postgres"
        self.paramstyle = db_module.paramstyle
        DB.__init__(self, db_module, keywords)
        self.supports_multiple_insert = True
        self._sequences = None

FROM pg_timezone_names
WHERE
	abbrev = %(tz)s
		AND
	name ~ '^[^/]+/[^/]+$'
		AND
	name !~ '^Etc/'
"""


# globals
_log = logging.getLogger('gm.db_pool')
_log.info('psycopg2 module version: %s' % dbapi.__version__)
_log.info('PostgreSQL via DB-API module "%s": API level %s, thread safety %s, parameter style "%s"' % (dbapi, dbapi.apilevel, dbapi.threadsafety, dbapi.paramstyle))
_log.info('libpq version (compiled in): %s', psycopg2.__libpq_version__)
_log.info('libpq version (loaded now) : %s', psycopg2.extensions.libpq_version())
#if '2.8' in dbapi.__version__:
#	_log.info('psycopg2 v2.8 detected, disabling connection pooling for the time being')
#	_DISABLE_CONNECTION_POOL = True


postgresql_version = None

_timestamp_template = "cast('%s' as timestamp with time zone)"		# MUST NOT be uniocde or else getquoted will not work (true in py3 ?)

_map_psyco_tx_status2str = [
	'TRANSACTION_STATUS_IDLE',
	'TRANSACTION_STATUS_ACTIVE',
	'TRANSACTION_STATUS_INTRANS',
	'TRANSACTION_STATUS_INERROR',
	'TRANSACTION_STATUS_UNKNOWN'
]

#   everyblock is distributed in the hope that it will be useful,
#   but WITHOUT ANY WARRANTY; without even the implied warranty of
#   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#   GNU General Public License for more details.
#
#   You should have received a copy of the GNU General Public License
#   along with everyblock.  If not, see .
#

from django.utils.safestring import SafeUnicode, SafeString
from ebpub.metros.allmetros import METRO_LIST
import psycopg2
import psycopg2.extensions

# This is taken from django/db/backends/postgresql_psycopg2/base.py
psycopg2.extensions.register_type(psycopg2.extensions.UNICODE)
psycopg2.extensions.register_adapter(SafeString, psycopg2.extensions.QuotedString)
psycopg2.extensions.register_adapter(SafeUnicode, psycopg2.extensions.QuotedString)

USERNAME = 'username'
HOST = '192.168.1.100'

def run_query(sql, params):
    """
    Runs the given SQL query against every city database and returns a
    dictionary mapping the short_name to the result.
    """
    result = {}
    for metro in METRO_LIST:
        # The connection parameters are hard-coded rather than using the
        # settings files because Django doesn't yet work with multiple
        # settings files.