python类__name__()的实例源码-面圈网

test_notify.py 文件源码项目：psycopg2-for-aws-lambda 作者: iwitaly 项目源码文件源码阅读 25 收藏 0 点赞 0 评论 0

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)

test_notify.py 文件源码项目：psycopg2-for-aws-lambda 作者: iwitaly 项目源码文件源码阅读 27 收藏 0 点赞 0 评论 0

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)

test_notify.py 文件源码项目：ShelbySearch 作者: Agentscreech 项目源码文件源码阅读 24 收藏 0 点赞 0 评论 0

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)

test_notify.py 文件源码项目：userbase-sns-lambda 作者: fartashh 项目源码文件源码阅读 44 收藏 0 点赞 0 评论 0

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)

test_notify.py 文件源码项目：userbase-sns-lambda 作者: fartashh 项目源码文件源码阅读 24 收藏 0 点赞 0 评论 0

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)

test_notify.py 文件源码项目：userbase-sns-lambda 作者: fartashh 项目源码文件源码阅读 21 收藏 0 点赞 0 评论 0

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)

test_notify.py 文件源码项目：userbase-sns-lambda 作者: fartashh 项目源码文件源码阅读 28 收藏 0 点赞 0 评论 0

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)

test_notify.py 文件源码项目：userbase-sns-lambda 作者: fartashh 项目源码文件源码阅读 22 收藏 0 点赞 0 评论 0

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)

test_notify.py 文件源码项目：userbase-sns-lambda 作者: fartashh 项目源码文件源码阅读 25 收藏 0 点赞 0 评论 0

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)

test_notify.py 文件源码项目：userbase-sns-lambda 作者: fartashh 项目源码文件源码阅读 33 收藏 0 点赞 0 评论 0

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)

test_notify.py 文件源码项目：userbase-sns-lambda 作者: fartashh 项目源码文件源码阅读 26 收藏 0 点赞 0 评论 0

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)

test_notify.py 文件源码项目：userbase-sns-lambda 作者: fartashh 项目源码文件源码阅读 25 收藏 0 点赞 0 评论 0

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)

test_notify.py 文件源码项目：Price-Comparator 作者: Thejas-1 项目源码文件源码阅读 25 收藏 0 点赞 0 评论 0

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)

test_notify.py 文件源码项目：nmbs-realtime-feed 作者: datamindedbe 项目源码文件源码阅读 31 收藏 0 点赞 0 评论 0

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)

test_notify.py 文件源码项目：aws-lambda-redshift-copy 作者: christianhxc 项目源码文件源码阅读 24 收藏 0 点赞 0 评论 0

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)

test_notify.py 文件源码项目：PyEloqua-Examples 作者: colemanja91 项目源码文件源码阅读 21 收藏 0 点赞 0 评论 0

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)

db.py 文件源码项目：arthur-redshift-etl 作者: harrystech 项目源码文件源码阅读 26 收藏 0 点赞 0 评论 0

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

test_notify.py 文件源码项目：flask 作者: bobohope 项目源码文件源码阅读 25 收藏 0 点赞 0 评论 0

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)

test_notify.py 文件源码项目：psycopg2-for-aws-lambda 作者: iwitaly 项目源码文件源码阅读 24 收藏 0 点赞 0 评论 0

def test_suite():
    return unittest.TestLoader().loadTestsFromName(__name__)

test_notify.py 文件源码项目：psycopg2-for-aws-lambda 作者: iwitaly 项目源码文件源码阅读 37 收藏 0 点赞 0 评论 0

def test_suite():
    return unittest.TestLoader().loadTestsFromName(__name__)

test_notify.py 文件源码项目：ShelbySearch 作者: Agentscreech 项目源码文件源码阅读 40 收藏 0 点赞 0 评论 0

def test_suite():
    return unittest.TestLoader().loadTestsFromName(__name__)

test_notify.py 文件源码项目：userbase-sns-lambda 作者: fartashh 项目源码文件源码阅读 23 收藏 0 点赞 0 评论 0

def test_suite():
    return unittest.TestLoader().loadTestsFromName(__name__)

test_notify.py 文件源码项目：userbase-sns-lambda 作者: fartashh 项目源码文件源码阅读 23 收藏 0 点赞 0 评论 0

def test_suite():
    return unittest.TestLoader().loadTestsFromName(__name__)

test_notify.py 文件源码项目：userbase-sns-lambda 作者: fartashh 项目源码文件源码阅读 26 收藏 0 点赞 0 评论 0

def test_suite():
    return unittest.TestLoader().loadTestsFromName(__name__)

test_notify.py 文件源码项目：userbase-sns-lambda 作者: fartashh 项目源码文件源码阅读 21 收藏 0 点赞 0 评论 0

def test_suite():
    return unittest.TestLoader().loadTestsFromName(__name__)

test_notify.py 文件源码项目：userbase-sns-lambda 作者: fartashh 项目源码文件源码阅读 22 收藏 0 点赞 0 评论 0

def test_suite():
    return unittest.TestLoader().loadTestsFromName(__name__)

test_notify.py 文件源码项目：userbase-sns-lambda 作者: fartashh 项目源码文件源码阅读 22 收藏 0 点赞 0 评论 0

def test_suite():
    return unittest.TestLoader().loadTestsFromName(__name__)

test_notify.py 文件源码项目：Price-Comparator 作者: Thejas-1 项目源码文件源码阅读 22 收藏 0 点赞 0 评论 0

def test_suite():
    return unittest.TestLoader().loadTestsFromName(__name__)

test_notify.py 文件源码项目：nmbs-realtime-feed 作者: datamindedbe 项目源码文件源码阅读 28 收藏 0 点赞 0 评论 0

def test_suite():
    return unittest.TestLoader().loadTestsFromName(__name__)

test_notify.py 文件源码项目：aws-lambda-redshift-copy 作者: christianhxc 项目源码文件源码阅读 23 收藏 0 点赞 0 评论 0

def test_suite():
    return unittest.TestLoader().loadTestsFromName(__name__)