Python psycopg2.extras 模块，DictCursor() 实例源码

def __iter__(self):
        with conn.cursor(cursor_factory=DictCursor) as cur:
            # TODO: save names of table and database
            # to a central location. For now, db=arxive and table=articles
            cur.execute("SELECT * FROM articles;")
            for article in cur:
                abstract = article['abstract'].replace('\n', ' ').strip()
                # train on body, composed of title and abstract
                body = article['title'] + '. '
                body += abstract
                # We want to keep some punctuation, as Word2Vec
                # considers them useful context
                words = re.findall(r"[\w']+|[.,!?;]", body)
                # lowercase. perhaps lemmatize too?
                words = [word.lower() for word in words]
                # document tag. Unique integer 'index' is good.
                # can also add topic tag of form
                # 'topic_{subject_id}' to list
                #tags = [article['index'], article['subject']]
                tags = [article['index']]

                yield TaggedDocument(words, tags)

def test_second_cursor(self):
        """
        Dict's cursor should not interfere with another cursor.
        """
        Person = self.db['person']
        bob = Person(name='Bob').flush()
        aly = Person(name='Aly').flush()
        self.assertDictContains(bob, {'name':'Bob', 'id':1})

        curs2 = self.conn.cursor(cursor_factory=DictCursor)
        persons = Person.get_where()
        self.assertEqual(next(persons), bob)

        curs2.execute('SELECT * FROM person')
        self.assertEqual(next(persons), aly)

        # Using dictorm's cursor will intefere
        persons = Person.get_where()
        self.assertEqual(next(persons), bob)
        persons.curs.execute('SELECT * FROM person')
        self.assertEqual(next(persons), bob)
        self.assertEqual(next(persons), aly)
        self.assertRaises(StopIteration, next, persons)

def getCursor(self,isQuery, withhold=False):
            ''' Return a named cursor. You don't have to close named cursor ''' 
            cursor_name = str(time.time()) 
            conn = self.pool.getconn()
            executionStatus = ''
            try:
                cursor = conn.cursor(cursor_name,cursor_factory=extras.DictCursor,withhold=withhold) if isQuery else conn.cursor()
                yield cursor
            except Exception, e:
                executionStatus = e.pgerror
                _exType, _exVal, exTrace = sys.exc_info()
                print 'Execution Status:',executionStatus
                print 'Stacktrace :',dir(exTrace)
                print 'Query: ',cursor.query
            finally:
                if(executionStatus != ''):
                     conn.rollback()
                else:
                     conn.commit()
                self.pool.putconn(conn)

def get_replication_status(db):
    result = {"result": False, "status": None}
    db_conn = connect('Destination', db_name=db)
    src_db_conn = connect('Source', db_name=db)
    result["result"] = False
    try:
        cur = db_conn.cursor(cursor_factory=extras.DictCursor)
        cur.execute("SELECT status FROM pglogical.show_subscription_status(subscription_name := 'subscription');")
        r = cur.fetchone()
        if r:
            result["result"] = True
            result["status"] = r['status']

    except psycopg2.InternalError:
        result["result"] = False
    except psycopg2.OperationalError:
        result["result"] = False
    except psycopg2.ProgrammingError:
        result["result"] = False

    return result

def get_database_connection(db_name: str=None, user: str=None, host: str=None, password: str=None):
    """Retrieve the database connection.

    This returns a database connection to the database, not just the database
    server.

    :param db_name:     Name of the database to connect to
    :param user:        Database username
    :param host:        Database host url
    :param password:    Database password

    :return:            A psycopg2 `connection`
    """
    if not host:
        host = os.getenv('POSTGRES_HOST', 'localhost')
    if not db_name:
        db_name = os.getenv('POSTGRES_DB', 'anxiety')
    if not user:
        user = os.getenv('POSTGRES_USER', 'postgres')
    if not password:
        password = os.getenv('POSTGRES_PASSWORD', 'p0stgres')

    return connect(database=db_name, user=user, password=password, host=host, cursor_factory=DictCursor)

def connect_to_database(user: str=None, host: str=None, password: str=None):
    """Retrieve a connection to the database server.

    :param user:        Database username
    :param host:        Database host url
    :param password:    Database password

    :return:            A psycopg2 `connection`
    """
    if not host:
        host = os.getenv('POSTGRES_HOST', 'localhost')
    if not user:
        user = os.getenv('POSTGRES_USER', 'postgres')
    if not password:
        password = os.getenv('POSTGRES_PASSWORD', 'p0stgres')

    return connect(user=user, password=password, host=host, cursor_factory=DictCursor)

def get_district_mismatches(self, comparison_view='compare', district_id=182):
        """Get OSM entities which have an fhrs:id for which there is no match
        in the database.

        district_id (integer): Boundary Line district ID
        Returns dict
        """

        dict_cur = self.connection.cursor(cursor_factory=DictCursor)

        sql = ('SELECT osm_name, osm_fhrsid, TRIM(TRAILING ' ' FROM osm_type) as osm_type,\n' +
               'osm_id, CONCAT(substring(osm_type FROM 1 FOR 1), osm_id) AS osm_ident\n' +
               'FROM compare\n' +
               'WHERE status = \'mismatch\' AND osm_district_id = %s')
        values = (district_id,)
        dict_cur.execute(sql, values)

        result = []
        for row in dict_cur.fetchall():
            result.append(row)

        return result

def execute(self, db_name: str, query: str, values: List, _type: str):
        """
        Execute SQL query in connection pool
        """
        warnings.warn("Use single methods!", DeprecationWarning)

        if _type not in ('select', 'insert', 'update', 'delete'):
            raise RuntimeError(
                'Wrong request type {}'.format(_type)
            )
        if not self.dbs[db_name]['master']:
            raise RuntimeError(
                'db {} master is not initialized'.format(db_name)
            )

        pool = self.dbs[db_name]['master']
        if _type == 'select' and 'slave' in self.dbs[db_name]:
            pool = self.dbs[db_name]['slave']

        async with pool.acquire() as conn:
            async with conn.cursor(cursor_factory=DictCursor) as cursor:
                await cursor.execute(query, values)
                if _type == 'select':
                    data = await cursor.fetchall()
                else:
                    data = cursor.rowcount
        return data

def _execute(self, query: str, values: Union[List, Dict], db_name: str = 'default',
                       returning: bool = False):
        pool = self.dbs[db_name]['master']
        if pool is None:
            raise RuntimeError('db {} master is not initialized'.format(db_name))

        async with pool.acquire() as conn:
            async with conn.cursor(cursor_factory=DictCursor) as cursor:
                await cursor.execute(query, values)
                if returning:
                    return await cursor.fetchone()
                else:
                    return cursor.rowcount

def _select(self, query: str, values: Union[List, Dict], db_name: str = 'default'):
        dbs = self.dbs[db_name]
        pool = dbs.get('slave') or dbs.get('master')
        if pool is None:
            raise RuntimeError('db {} master is not initialized'.format(db_name))

        async with pool.acquire() as conn:
            async with conn.cursor(cursor_factory=DictCursor) as cursor:
                await cursor.execute(query, values)
                return await cursor.fetchall()

def _first(self, query: str, values: Union[List, Dict], db_name: str = 'default'):
        dbs = self.dbs[db_name]
        pool = dbs.get('slave') or dbs.get('master')
        if pool is None:
            raise RuntimeError('db {} master is not initialized'.format(db_name))

        async with pool.acquire() as conn:
            async with conn.cursor(cursor_factory=DictCursor) as cursor:
                await cursor.execute(query, values)
                return await cursor.fetchone()

def get_cursor(self):
        """
        Returns a cursor from the provided database connection that DictORM
        objects expect.
        """
        if self.kind == 'sqlite3':
            self.conn.row_factory = sqlite3.Row
            return self.conn.cursor()
        elif self.kind == 'postgresql':
            return self.conn.cursor(cursor_factory=DictCursor)

def get_articles(indices):
    """
    INPUT: list of integers corresponding to
        'index' column values of desired articles in database

    OUTPUT: list of dictionaries, each dictionary
        corresponding to an article
    """
    with conn.cursor(cursor_factory=DictCursor) as cur:
        query = "SELECT * FROM articles \
            WHERE index IN %s \
            ORDER BY last_submitted DESC"
        cur.execute(query, (tuple(indices),))
        articles = cur.fetchall()
        return articles

def get_articles_by_subject(subject):
    """
    INPUT:
        (str): subject name

    OUTPUT: list of dictionaries, each dictionary
        corresponding to an article
    """
    with conn.cursor(cursor_factory=DictCursor) as cur:
        query = "SELECT * FROM articles \
            WHERE subject=%s \
            ORDER BY last_submitted DESC"
        cur.execute(query, (subject,))
        articles = cur.fetchall()
        return articles

def get_article(index):
    """
    INPUT:
        (int): article index

    OUTPUT:
        (dict): dictionary object representing
            article matching the given index
    """
    with conn.cursor(cursor_factory=DictCursor) as cur:
        query = "SELECT * FROM articles WHERE index=%s"
        cur.execute(query, (index, ))
        article = cur.fetchone()
        return article

def fetch(conn, query, params=[]):
    if conn[TYPE] == 'postgresql':
        cur = conn[CONN].cursor(cursor_factory=postgresql_extras.DictCursor)
    if conn[TYPE] == 'mysql':
        cur = conn[CONN].cursor(mysql.cursors.DictCursor)
    cur.execute(query, params)
    while True:
        line = cur.fetchone()
        if not line: break
        yield dict(line)
    cur.close()

def fetchone(conn, query, params=[]):
    if conn[TYPE] == 'postgresql':
        cur = conn[CONN].cursor(cursor_factory=postgresql_extras.DictCursor)
    if conn[TYPE] == 'mysql':
        cur = conn[CONN].cursor(mysql.cursors.DictCursor)
    try:
        cur.execute(query+" LIMIT 1", params)
    except ProgrammingError as e:
        print("incorrectly formulated model definition :\n{0}".format(e))
        sys.exit(1)
    line = cur.fetchone()
    line = dict(line) if line else None
    cur.close()
    return line

def db_getpaste(db, pasteid):
    with db.cursor(cursor_factory=DictCursor) as cur:
        cur.execute(("""SELECT * FROM pastes WHERE pasteid = %s;"""), (pasteid,))
        r = cur.fetchone()
    return r

def getstats(db):
    stats = {}
    with db.cursor(cursor_factory=DictCursor) as cur:
        cur.execute("SELECT * FROM dailystats WHERE date = %s;", (datetime.utcnow().date(),))
        stats['daily'] = cur.fetchone()
        cur.execute("SELECT * FROM stats;")
        totalstats = {}
        for i in cur.fetchall():
            totalstats[i[0]] = i[1]
        stats['total'] = totalstats
        print()
        return stats

def execute_sql(sql):
    # je hebt hier een andere config nodig dan die in de global_test_suite staat!:
    engine = create_engine(test_config['conn_dwh'])
    conn = engine.raw_connection()
    cursor = conn.cursor(cursor_factory=DictCursor)
    cursor.execute(sql)
    result = cursor.fetchall()
    conn.commit()
    cursor.close()

    return result

def execute_sql(sql):
    # conn = psycopg2.connect("""host='localhost' dbname='pyelt_unittests' user='postgres' password='{}'""".format(get_your_password()))
    engine = create_engine(general_config['conn_dwh'])
    conn = engine.raw_connection()
    cursor = conn.cursor(cursor_factory=DictCursor)
    cursor.execute(sql)
    result = cursor.fetchall()
    conn.commit()
    cursor.close()

    return result

def cursor():
    return con.cursor(cursor_factory=DictCursor)

def get_district_postcode_errors(self, comparison_view='compare',
                                     fhrs_table='fhrs_establishments', district_id=182):
        """Get OSM entities which have an fhrs:id that matches an FHRS
        establishment but has no postcode or a mismatching one.

        comparison_view (string): name of comparison database view
        fhrs_table (string): name of FHRS establishments database table
        district_id (integer): Boundary Line district ID
        Returns dict
        """

        dict_cur = self.connection.cursor(cursor_factory=DictCursor)

        sql = ('SELECT osm_name, osm_id, osm_fhrsid, osm_postcode, fhrs_postcode,\n' +
               'TRIM(TRAILING \' \' FROM osm_type) AS osm_type,\n' +
               'CONCAT(substring(osm_type FROM 1 FOR 1), osm_id) AS osm_ident,\n' +
               'CONCAT(\n' +
               'CASE WHEN "AddressLine1" IS NOT NULL THEN\n' +
               '    CONCAT(\'%7Cfixme:addr1=\', "AddressLine1") END,\n' +
               'CASE WHEN "AddressLine2" IS NOT NULL THEN\n' +
               '    CONCAT(\'%7Cfixme:addr2=\', "AddressLine2") END,\n' +
               'CASE WHEN "AddressLine3" IS NOT NULL THEN\n' +
               '    CONCAT(\'%7Cfixme:addr3=\', "AddressLine3") END,\n' +
               'CASE WHEN "AddressLine4" IS NOT NULL THEN\n' +
               '    CONCAT(\'%7Cfixme:addr4=\', "AddressLine4") END,\n' +
               'CASE WHEN "PostCode" IS NOT NULL THEN\n' +
               '    CONCAT(\'%7Caddr:postcode=\', "PostCode") END,\n' +
               '\'%7Csource:addr=FHRS Open Data\') AS add_tags_string\n' +
               'FROM compare\n' +
               'LEFT JOIN ' + fhrs_table + ' ON fhrs_fhrsid = "FHRSID"\n' +
               'WHERE status = \'matched_postcode_error\' AND '
               'osm_district_id = ' + str(district_id))
        dict_cur.execute(sql)

        result = []
        for row in dict_cur.fetchall():
            result.append(row)

        return result

def get_district_duplicates(self, osm_table='osm', fhrs_table='fhrs_establishments',
                                district_id=182):
        """Get OSM entities which have an fhrs:id shared by at least one OSM
        entity within the specified district.

        osm_table (string): name of OSM database table
        fhrs_table (string): name of FHRS establishments database table
        district_id (integer): Boundary Line district ID
        Returns dict
        """

        dict_cur = self.connection.cursor(cursor_factory=DictCursor)

        sql = ('SELECT id, TRIM(TRAILING ' ' FROM type) as type,\n' +
               'CONCAT(substring(type FROM 1 FOR 1), id) AS osm_ident, "fhrs:id",\n' +
               osm_table + '.district_id, name AS osm_name, "BusinessName" AS fhrs_name\n' +
               'FROM ' + osm_table + '\n' +
               'LEFT JOIN ' + fhrs_table + ' ON "fhrs:id" = CAST("FHRSID" AS TEXT)\n' +
               'WHERE "fhrs:id" IN (\n' +
               '    SELECT "fhrs:id" FROM osm\n' +
               '    WHERE district_id = %s\n' +
               '    GROUP BY "fhrs:id" HAVING COUNT("fhrs:id") > 1)\n' +
               'ORDER BY "fhrs:id";')
        values = (district_id,)
        dict_cur.execute(sql, values)

        result = []
        for row in dict_cur.fetchall():
            result.append(row)

        return result

def get_district_distant_matches(self, distant_matches_view='distant_matches',
                                     district_id=182):
        """Get OSM entities that are matched to an FHRS establishment where
        the OSM/FHRS locations are distant.

        distant_matches_view (string): name of distant matches database view
        district_id (integer): Boundary Line district ID
        Returns dict
        """

        dict_cur = self.connection.cursor(cursor_factory=DictCursor)

        sql = ('SELECT osm_id, osm_type,\n' +
               'CONCAT(SUBSTRING(osm_type FROM 1 FOR 1), osm_id) AS osm_ident, fhrs_id,\n' +
               'osm_name, fhrs_name, distance\n' +
               'FROM ' + distant_matches_view + '\n' +
               'WHERE district_id = %s' +
               'ORDER BY distance;')
        values = (district_id,)
        dict_cur.execute(sql, values)

        result = []
        for row in dict_cur.fetchall():
            result.append(row)

        return result

def __init__(self, params, zField='z'):
        psycopg2.extensions.register_type(psycopg2.extensions.UNICODE)
        psycopg2.extensions.register_type(psycopg2.extensions.UNICODEARRAY)        
        connection = psycopg2.connect(**connParams)

        self.geotransform = []        
        self.x = []
        self.y = []
        self.vals = []

        cursor = connection.cursor(cursor_factory=DictCursor)
        cursor.execute(PG_SQL, params)
        xMin, xMax, yMin, yMax = 91, -91, 181, -181
        for record in cursor:
            # print record['txt_geom']
            # lat, lon = [float(x) for x in record['txt_geom'].split(' ')]
            y = record['y']
            x = record['x']

            if yMin > y:
                yMin = y
            if yMax < y:
                yMax = y

            if xMin > x:
                xMin = x
            if xMax < x:
                xMax = x

            self.x.append(x)
            self.y.append(y)
            self.vals.append(record[zField])
        cursor.close()
        connection.close()
        # print xMin, xMax, yMin, yMax
        xSize, ySize = abs(xMax - xMin) / 0.0003, abs(yMin - yMax) / 0.0003
        self.size = xSize, ySize

        self.geotransform = [xMin, (xMax - xMin) / xSize, 0,
                             yMax, 0, (yMin - yMax) / ySize]

        self.proj = SpatialReference()
        self.proj.ImportFromEPSG(4326)
        xSize, ySize = abs(xMax - xMin) / 0.0003, abs(yMin - yMax) / 0.0003
        self.size = xSize, ySize

        self.geotransform = [xMin, (xMax - xMin) / xSize, 0,
                             yMax, 0, (yMin - yMax) / ySize]

def get_gpx(self, geog_col='fhrs_geog', name_col='fhrs_name',
                view_name='compare', district_id_col='fhrs_district_id',
                district_id=182, status=None):
        """Return a GPX representation of waypoints from the database using
        the specified parameters.

        geog_col (string): name of column containing waypoint geography
        name_col (string): name of column containing waypoint name
        view_name (string): name of view which contains the data
        district_id_col (string): name of column containing Boundary Line
            district id
        district_id (integer): Boundary Line district ID
        status (string): status of waypoints to be selected e.g. 'matched'
        Returns string
        """

        # use supplied variables to get waypoints from database
        dict_cur = self.connection.cursor(cursor_factory=DictCursor)

        sql = ("SELECT ST_Y(" + geog_col + "::geometry) as lat, " +
               "ST_X(" + geog_col + "::geometry) as lon,\n" +
               name_col + " as name\n" +
               "FROM " + view_name + "\n" +
               "WHERE " + district_id_col + "=%s")
        if status:
            sql += " AND status=%s"
            values = (district_id, status)
        else:
            values = (district_id,)
        dict_cur.execute(sql, values)

        waypoints = [] # empty list to hold waypoint dicts
        for row in dict_cur.fetchall():
            if row['name']:
                waypoints.append({'lat': str(row['lat']), 'lon': str(row['lon']),
                                  'name': escape(row['name'])})
            else:
                waypoints.append({'lat': str(row['lat']), 'lon': str(row['lon']),
                                  'name': '???'})

        # create GPX file
        output = ('<?xml version="1.0" encoding="UTF-8"?>\n' +
            '<gpx version="1.0" creator="python-fhrs-osm"\n' +
            '    xmlns="http://www.topografix.com/GPX/1/0">\n')
        for waypoint in waypoints:
            output += ('<wpt lat="' + waypoint['lat'] + '" lon="' + waypoint['lon'] + '">\n' +
                '    <name>' + waypoint['name'] + '</name>\n' +
                '</wpt>\n')
        output += '</gpx>'
        return output