def save(self, *args, **kwargs):
# unsure how self.updated_at refers to class variable BaseModel.updated_at
self.updated_at = datetime.datetime.now().strftime("%Y/%m/%d %H:%M:%S")
peewee.Model.save(self)
python类Model()的实例源码
def get_model_class(self):
if self.database is None:
raise RuntimeError('Database must be initialized.')
class BaseModel(Model):
class Meta:
database = self.database
return BaseModel
def Model(self):
if self._app is None:
database = getattr(self, 'database', None)
if database is None:
self.database = Proxy()
if not hasattr(self, '_model_class'):
self._model_class = self.get_model_class()
return self._model_class
def get_model_class(self):
if self.database is None:
raise RuntimeError('Database must be initialized.')
class BaseModel(Model):
class Meta:
database = self.database
return BaseModel
def Model(self):
if self._app is None:
database = getattr(self, 'database', None)
if database is None:
self.database = Proxy()
if not hasattr(self, '_model_class'):
self._model_class = self.get_model_class()
return self._model_class
def select(cls, *args, **kwargs):
"""Support read slaves."""
query = super(Model, cls).select(*args, **kwargs)
query.database = cls._get_read_database()
return query
def raw(cls, *args, **kwargs):
query = super(Model, cls).raw(*args, **kwargs)
if query._sql.lower().startswith('select'):
query.database = cls._get_read_database()
return query
def save(self, force_insert=False, **kwargs):
"""Send signals."""
created = force_insert or not bool(self.pk)
self.pre_save.send(self, created=created)
super(Model, self).save(force_insert=force_insert, **kwargs)
self.post_save.send(self, created=created)
def delete_instance(self, *args, **kwargs):
"""Send signals."""
self.pre_delete.send(self)
super(Model, self).delete_instance(*args, **kwargs)
self.post_delete.send(self)
def init_app(self, app, database=None):
"""Initialize application."""
# Register application
if not app:
raise RuntimeError('Invalid application.')
self.app = app
if not hasattr(app, 'extensions'):
app.extensions = {}
app.extensions['peewee'] = self
app.config.setdefault('PEEWEE_CONNECTION_PARAMS', {})
app.config.setdefault('PEEWEE_DATABASE_URI', 'sqlite:///peewee.sqlite')
app.config.setdefault('PEEWEE_MANUAL', False)
app.config.setdefault('PEEWEE_MIGRATE_DIR', 'migrations')
app.config.setdefault('PEEWEE_MIGRATE_TABLE', 'migratehistory')
app.config.setdefault('PEEWEE_MODELS_CLASS', Model)
app.config.setdefault('PEEWEE_MODELS_IGNORE', [])
app.config.setdefault('PEEWEE_MODELS_MODULE', '')
app.config.setdefault('PEEWEE_READ_SLAVES', '')
app.config.setdefault('PEEWEE_USE_READ_SLAVES', True)
# Initialize database
params = app.config['PEEWEE_CONNECTION_PARAMS']
database = database or app.config.get('PEEWEE_DATABASE_URI')
if not database:
raise RuntimeError('Invalid database.')
database = get_database(database, **params)
slaves = app.config['PEEWEE_READ_SLAVES']
if isinstance(slaves, string_types):
slaves = slaves.split(',')
self.slaves = [get_database(slave, **params) for slave in slaves if slave]
self.database.initialize(database)
if self.database.database == ':memory:':
app.config['PEEWEE_MANUAL'] = True
if not app.config['PEEWEE_MANUAL']:
app.before_request(self.connect)
app.teardown_request(self.close)
def test_create_table(self):
class SomeModel(pw.Model):
some_field = pw.CharField(null=True)
class Meta:
database = self.db
self.evolve_and_check_noop()
SomeModel.create(some_field='woot')
self.assertEqual(SomeModel.select().first().some_field, 'woot')
def test_drop_table(self):
class SomeModel(pw.Model):
some_field = pw.CharField(null=True)
class Meta:
database = self.db
self.evolve_and_check_noop()
SomeModel.create(some_field='woot')
peeweedbevolve.clear()
self.evolve_and_check_noop()
with self.assertRaises(pw.ProgrammingError):
SomeModel.create(some_field='woot2') # fails because table isn't there
def test_create_table_with_fk(self):
class SomeModel(pw.Model):
some_field = pw.CharField(null=True)
class Meta:
database = self.db
class SomeModel2(pw.Model):
some_field2 = pw.CharField(null=True)
some_model = pw.ForeignKeyField(rel_model=SomeModel)
class Meta:
database = self.db
self.evolve_and_check_noop()
sm = SomeModel.create(some_field='woot')
sm2 = SomeModel2.create(some_field2='woot2', some_model=sm)
def test_rename_table_aka_string(self):
self.test_create_table()
peeweedbevolve.clear()
class SomeOtherModel(pw.Model):
some_field = pw.CharField(null=True)
class Meta:
database = self.db
aka = 'somemodel'
self.evolve_and_check_noop()
self.assertEqual(SomeOtherModel.select().first().some_field, 'woot')
def test_rename_table_aka_list(self):
self.test_create_table()
peeweedbevolve.clear()
class SomeOtherModel(pw.Model):
some_field = pw.CharField(null=True)
class Meta:
database = self.db
aka = ['somemodel']
self.evolve_and_check_noop()
self.assertEqual(SomeOtherModel.select().first().some_field, 'woot')
def test_add_column(self):
self.test_create_table()
peeweedbevolve.clear()
class SomeModel(pw.Model):
some_field = pw.CharField(null=True)
another_field = pw.CharField(null=True)
class Meta:
database = self.db
self.evolve_and_check_noop()
self.assertEqual(SomeModel.select().first().another_field, None)
def test_rename_column_aka_string(self):
self.test_create_table()
peeweedbevolve.clear()
class SomeModel(pw.Model):
some_other_field = pw.CharField(null=True, aka='some_field')
class Meta:
database = self.db
self.evolve_and_check_noop()
self.assertEqual(SomeModel.select().first().some_other_field, 'woot')
def test_drop_column(self):
self.test_create_table()
peeweedbevolve.clear()
class SomeModel(pw.Model):
class Meta:
database = self.db
self.evolve_and_check_noop()
self.assertFalse(hasattr(SomeModel.select().first(), 'some_field'))
def test_rename_table_add_column(self):
self.test_create_table()
peeweedbevolve.clear()
class SomeOtherModel(pw.Model):
some_field = pw.CharField(null=True)
another_field = pw.CharField(null=True)
class Meta:
database = self.db
aka = 'somemodel'
self.evolve_and_check_noop()
o = SomeOtherModel.select().first()
self.assertEqual(o.some_field, 'woot')
self.assertEqual(o.another_field, None)
