我们从Python开源项目中，提取了以下46个代码示例，用于说明如何使用psycopg2.extensions.STATUS_IN_TRANSACTION。
def _get_json_oids(conn_or_curs, name='json'): # lazy imports from psycopg2.extensions import STATUS_IN_TRANSACTION from psycopg2.extras import _solve_conn_curs conn, curs = _solve_conn_curs(conn_or_curs) # Store the transaction status of the connection to revert it after use conn_status = conn.status # column typarray not available before PG 8.3 typarray = conn.server_version >= 80300 and "typarray" or "NULL" # get the oid for the hstore curs.execute( "SELECT t.oid, %s FROM pg_type t WHERE t.typname = %%s;" % typarray, (name,)) r = curs.fetchone() # revert the status of the connection as before the command if (conn_status != STATUS_IN_TRANSACTION and not conn.autocommit): conn.rollback() if not r: raise conn.ProgrammingError("%s data type not found" % name) return r
def get_oids(self, conn_or_curs): """Return the lists of OID of the hstore and hstore[] types. """ conn, curs = _solve_conn_curs(conn_or_curs) # Store the transaction status of the connection to revert it after use conn_status = conn.status # column typarray not available before PG 8.3 typarray = conn.server_version >= 80300 and "typarray" or "NULL" rv0, rv1 = [], [] # get the oid for the hstore curs.execute("""\ SELECT t.oid, %s FROM pg_type t JOIN pg_namespace ns ON typnamespace = ns.oid WHERE typname = 'hstore'; """ % typarray) for oids in curs: rv0.append(oids[0]) rv1.append(oids[1]) # revert the status of the connection as before the command if (conn_status != _ext.STATUS_IN_TRANSACTION and not conn.autocommit): conn.rollback() return tuple(rv0), tuple(rv1)
def _from_db(self, name, conn_or_curs): """Return a `CompositeCaster` instance for the type *name*. Raise `ProgrammingError` if the type is not found. """ conn, curs = _solve_conn_curs(conn_or_curs) # Store the transaction status of the connection to revert it after use conn_status = conn.status # Use the correct schema if '.' in name: schema, tname = name.split('.', 1) else: tname = name schema = 'public' # column typarray not available before PG 8.3 typarray = conn.server_version >= 80300 and "typarray" or "NULL" # get the type oid and attributes curs.execute("""\ SELECT t.oid, %s, attname, atttypid FROM pg_type t JOIN pg_namespace ns ON typnamespace = ns.oid JOIN pg_attribute a ON attrelid = typrelid WHERE typname = %%s AND nspname = %%s AND attnum > 0 AND NOT attisdropped ORDER BY attnum; """ % typarray, (tname, schema)) recs = curs.fetchall() # revert the status of the connection as before the command if (conn_status != _ext.STATUS_IN_TRANSACTION and not conn.autocommit): conn.rollback() if not recs: raise psycopg2.ProgrammingError( "PostgreSQL type '%s' not found" % name) type_oid = recs[0][0] array_oid = recs[0][1] type_attrs = [(r[2], r[3]) for r in recs] return self(tname, type_oid, type_attrs, array_oid=array_oid, schema=schema)
def _from_db(self, name, conn_or_curs): """Return a `CompositeCaster` instance for the type *name*. Raise `ProgrammingError` if the type is not found. """ conn, curs = _solve_conn_curs(conn_or_curs) # Store the transaction status of the connection to revert it after use conn_status = conn.status # Use the correct schema if '.' in name: schema, tname = name.split('.', 1) else: tname = name schema = 'public' # column typarray not available before PG 8.3 typarray = conn.server_version >= 80300 and "typarray" or "NULL" # get the type oid and attributes curs.execute("""\ SELECT t.oid, %s, attname, atttypid FROM pg_type t JOIN pg_namespace ns ON typnamespace = ns.oid JOIN pg_attribute a ON attrelid = typrelid WHERE typname = %%s AND nspname = %%s AND attnum > 0 AND NOT attisdropped ORDER BY attnum; """ % typarray, (tname, schema)) recs = curs.fetchall() # revert the status of the connection as before the command if (conn_status != _ext.STATUS_IN_TRANSACTION and not conn.autocommit): conn.rollback() if not recs: raise psycopg2.ProgrammingError( "PostgreSQL type '%s' not found" % name) type_oid = recs[0][0] array_oid = recs[0][1] type_attrs = [ (r[2], r[3]) for r in recs ] return self(tname, type_oid, type_attrs, array_oid=array_oid, schema=schema)
