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)
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。