python类DATABASE的实例源码

def __init__(self):
        handlers = [
            URLSpec(r'/', IndexHandler),
            URLSpec(r'/cat/(\d{1,2})', CatHandler, name="cat"),
            URLSpec(r'/detail/(\w+)', DetailHandler),
            URLSpec(r'/login', LoginHandler),
        ]

        tornado.web.Application.__init__(self, handlers, **settings.settings)
        self.db = torndb.Connection(host=settings.DBHOST, database=settings.DATABASE, user=settings.DBUSER, password=settings.DBPWD)

def initialize(self):
        print("Connecting to %s ..." % self.db)
        self.con = sqlite3.connect(DATABASE)

def create_app(loop):
    app = web.Application(loop=loop, debug=settings.DEBUG)
    setup_jinja(app, settings.DEBUG)
    aiohttp_session.setup(app, EncryptedCookieStorage(
        settings.SESSION_SECRET.encode('utf-8'),
        max_age=settings.SESSION_MAX_AGE))
    app.middlewares.append(aiohttp_login.flash.middleware)

    app.router.add_get('/', handlers.index)
    app.router.add_get('/users/', handlers.users, name='users')

    app['db'] = await asyncpg.create_pool(dsn=settings.DATABASE, loop=loop)
    aiohttp_login.setup(app, AsyncpgStorage(app['db']), settings.AUTH)

    return app

def db_connect():
    return create_engine(URL(**settings.DATABASE))

def __init__(self):
        """
        Initializes database connection.
        """
        try:
            self.connection = psycopg2.connect(
                database=DATABASE['database'],
                user=DATABASE['username'],
                password=DATABASE['password'],
                host=DATABASE['host']
            )
            self.cursor = self.connection.cursor()
        except Exception, e:
            raise e

def build(self):

        existed = os.path.exists(DATABASE)

        if not existed:
            db.connect()
            db.create_tables(self.tables)