Python peewee 模块，MySQLDatabase() 实例源码

def __get_migrator(self):
        if isinstance(self.db.engine, peewee.SqliteDatabase) or isinstance(self.db.engine, SqliteExtDatabase):
            return SqliteMigrator(self.db.engine)
        elif isinstance(self.db.engine, peewee.MySQLDatabase):
            return MySQLMigrator(self.db.engine)
        elif isinstance(self.db.engine, peewee.PostgresqlDatabase):
            return PostgresqlMigrator(self.db.engine)
        raise ImproperlyConfigured('Database engine doesn\'t support Migrations!')

def get_db_conn():
    import peewee
    env = os.environ.get('SENTINEL_ENV', 'production')

    # default values should be used unless you need a different config for development
    db_host = sentinel_cfg.get('db_host', '127.0.0.1')
    db_port = sentinel_cfg.get('db_port', None)
    db_name = sentinel_cfg.get('db_name', 'sentinel')
    db_user = sentinel_cfg.get('db_user', 'sentinel')
    db_password = sentinel_cfg.get('db_password', 'sentinel')
    db_charset = sentinel_cfg.get('db_charset', 'utf8mb4')
    db_driver = sentinel_cfg.get('db_driver', 'sqlite')

    if (env == 'test'):
        if db_driver == 'sqlite':
            db_name = sqlite_test_db_name(db_name)
        else:
            db_name = "%s_test" % db_name

    peewee_drivers = {
        'mysql': peewee.MySQLDatabase,
        'postgres': peewee.PostgresqlDatabase,
        'sqlite': peewee.SqliteDatabase,
    }
    driver = peewee_drivers.get(db_driver)

    dbpfn = 'passwd' if db_driver == 'mysql' else 'password'
    db_conn = {
        'host': db_host,
        'user': db_user,
        dbpfn: db_password,
    }
    if db_port:
        db_conn['port'] = int(db_port)

    if driver == peewee.SqliteDatabase:
        db_conn = {}

    db = driver(db_name, **db_conn)

    return db

def get_db_conn():
    import peewee
    env = os.environ.get('SENTINEL_ENV', 'production')

    # default values should be used unless you need a different config for development
    db_host = sentinel_cfg.get('db_host', '127.0.0.1')
    db_port = sentinel_cfg.get('db_port', None)
    db_name = sentinel_cfg.get('db_name', 'sentinel')
    db_user = sentinel_cfg.get('db_user', 'sentinel')
    db_password = sentinel_cfg.get('db_password', 'sentinel')
    db_charset = sentinel_cfg.get('db_charset', 'utf8mb4')
    db_driver = sentinel_cfg.get('db_driver', 'sqlite')

    if (env == 'test'):
        if db_driver == 'sqlite':
            db_name = sqlite_test_db_name(db_name)
        else:
            db_name = "%s_test" % db_name

    peewee_drivers = {
        'mysql': peewee.MySQLDatabase,
        'postgres': peewee.PostgresqlDatabase,
        'sqlite': peewee.SqliteDatabase,
    }
    driver = peewee_drivers.get(db_driver)

    dbpfn = 'passwd' if db_driver == 'mysql' else 'password'
    db_conn = {
        'host': db_host,
        'user': db_user,
        dbpfn: db_password,
    }
    if db_port:
        db_conn['port'] = int(db_port)

    if driver == peewee.SqliteDatabase:
        db_conn = {}

    db = driver(db_name, **db_conn)

    return db

def is_mysql(db):
  return isinstance(db, pw.MySQLDatabase)

def setUp(self):
    os.system('echo "create database peeweedbevolve_test;" | mysql')
    self.db = pw.MySQLDatabase('peeweedbevolve_test')
    self.db.connect()
    peeweedbevolve.clear()

def get_database(db_type):
    """
    Get the database object that should be used.

    Args:
        db_type: string

    Returns:
        peewee database driver.
    """

    db = False
    if db_type == 'sql':
        from peewee import SqliteDatabase

        db = SqliteDatabase(env("DB"))

    elif db_type == 'mysql':
        from peewee import MySQLDatabase

        db = MySQLDatabase(
            env("DB"),
            host=env("DB_HOST"),
            port=int(env("DB_PORT")),
            user=env("DB_USERNAME"),
            passwd=env("DB_PASSWORD")
        )

    elif db_type == 'postgresql':
        from peewee import PostgresqlDatabase

        db = PostgresqlDatabase(
            env('DB'),
            host=env("DB_HOST"),
            port=int(env("DB_PORT")),
            user=env("DB_USERNAME"),
            passwd=env("DB_PASSWORD")
        )

    elif db_type == 'berkeley':
        from playhouse.berkeleydb import BerkeleyDatabase

        db = BerkeleyDatabase(
            env('DB'),
            host=env("DB_HOST"),
            port=int(env("DB_PORT")),
            user=env("DB_USERNAME"),
            passwd=env("DB_PASSWORD")
        )

    return db