Python psycopg2 模块,__name__() 实例源码
我们从Python开源项目中,提取了以下33个代码示例,用于说明如何使用psycopg2.__name__()。
def notify(self, name, sec=0, payload=None):
"""Send a notification to the database,eventually after some time."""
if payload is None:
payload = ''
else:
payload = ",%r" % payload
script = ("""\
import time
time.sleep(%(sec)s)
import %(module)s as psycopg2
import %(module)s.extensions as ext
conn = psycopg2.connect(%(dsn)r)
conn.set_isolation_level(ext.ISOLATION_LEVEL_AUTOCOMMIT)
print conn.get_backend_pid()
curs = conn.cursor()
curs.execute("NOTIFY " %(name)r %(payload)r)
curs.close()
conn.close()
""" % {
'module': psycopg2.__name__,
'dsn': dsn, 'sec': sec, 'name': name, 'payload': payload})
return Popen([sys.executable, '-c', script_to_py3(script)], stdout=PIPE)
def notify(self,%r" % payload
script = ("""\
import time
time.sleep(%(sec)s)
import %(module)s as psycopg2
import %(module)s.extensions as ext
conn = psycopg2.connect(%(dsn)r)
conn.set_isolation_level(ext.ISOLATION_LEVEL_AUTOCOMMIT)
print conn.get_backend_pid()
curs = conn.cursor()
curs.execute("NOTIFY " %(name)r %(payload)r)
curs.close()
conn.close()
""" % {
'module': psycopg2.__name__,
'dsn': dsn, stdout=PIPE)
def extract_dsn(dsn_dict: Dict[str, str], read_only=False):
"""
Break the connection string into a JDBC URL and connection properties.
This is necessary since a JDBC URL may not contain all the properties needed
to successfully connect,e.g. username,password. These properties must
be passed in separately.
Use the postgresql subprotocol and driver regardless of whether the connection
string's protocol was postgres or redshift.
"""
dsn_properties = dict(dsn_dict) # so as to not mutate the argument
dsn_properties.update({
"ApplicationName": __name__,
"readOnly": "true" if read_only else "false",
"driver": "org.postgresql.Driver"
})
if "port" in dsn_properties:
jdbc_url = "jdbc:postgresql://{host}:{port}/{database}".format(**dsn_properties)
else:
jdbc_url = "jdbc:postgresql://{host}/{database}".format(**dsn_properties)
return jdbc_url, dsn_properties
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)
def connection_pool(max_conn, dsn_dict: Dict[str, application_name=psycopg2.__name__):
"""
Create a connection pool (with up to max_conn connections),where all connections will use the
given connection string.
"""
dsn_values = _dsn_connection_values(dsn_dict, application_name)
return psycopg2.pool.ThreadedConnectionPool(1, max_conn, **dsn_values)