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!')
python类MySQLDatabase()的实例源码
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
