def test_add_not_null_not_in_version(self):
another_column = CharField(null=True, default='')
migrate(migrator.add_column('food', 'another_column', another_column))
# drop the field from the nested version
migrator.drop_column('foodversion', 'another_column').run()
self.assertTableDoesNotHaveColumn('foodversion', 'another_column')
models = introspector.generate_models()
food = models['food']
self.assertTrue(food.another_column.null)
migrate(migrator.add_not_null('food', 'another_column'))
models = introspector.generate_models()
food = models['food']
self.assertFalse(food.another_column.null)
self.assertTableDoesNotHaveColumn('foodversion', 'another_column')
python类CharField()的实例源码
def test_drop_not_null_not_in_version(self):
another_column = CharField(default='')
migrate(migrator.add_column('food', 'another_column', another_column))
# drop the field from the nested version
migrator.drop_column('foodversion', 'another_column').run()
self.assertTableDoesNotHaveColumn('foodversion', 'another_column')
models = introspector.generate_models()
food = models['food']
self.assertFalse(food.another_column.null)
migrate(migrator.drop_not_null('food', 'another_column'))
models = introspector.generate_models()
food = models['food']
self.assertTrue(food.another_column.null)
self.assertTableDoesNotHaveColumn('foodversion', 'another_column')
def column(self, coltype, name, **kwargs):
"""
Generic method to add a column of any type.
:param coltype: Column type (from FIELD_TO_PEEWEE).
:param name: Name of column.
:param kwargs: Arguments for the given column type.
"""
constraints = kwargs.pop('constraints', [])
new_constraints = []
for const in constraints:
if isinstance(const, str):
const = peewee.SQL(const)
new_constraints.append(const)
kwargs['constraints'] = new_constraints
field_class = FIELD_TO_PEEWEE.get(coltype, peewee.CharField)
field_class(**kwargs).add_to_class(self.model, name)
def test_get_fields(self):
handler = PeeweeHandler(User)
fields_objs, fields_names = handler.get_fields()
self.assertIsInstance(fields_objs, list)
self.assertIsInstance(fields_names, list)
expected_fields = ["id", "name", "username", "password_hash", "email", "visits", "description"]
self.assertCountEqual(fields_names, expected_fields)
for field in fields_objs:
self.assertIsInstance(field, Field)
self.assertIsInstance(fields_objs[fields_names.index("id")], IntegerField)
self.assertIsInstance(fields_objs[fields_names.index("name")], CharField)
self.assertIsInstance(fields_objs[fields_names.index("username")], CharField)
self.assertIsInstance(fields_objs[fields_names.index("password_hash")], BlobField)
self.assertIsInstance(fields_objs[fields_names.index("email")], CharField)
self.assertIsInstance(fields_objs[fields_names.index("visits")], IntegerField)
self.assertIsInstance(fields_objs[fields_names.index("description")], CharField)
def test_add_not_null_constraint_with_records_and_default_which_is_function(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=None)
peeweedbevolve.clear()
def woot():
return 'woot'
class SomeModel(pw.Model):
some_field = pw.CharField(null=False, default=woot)
class Meta:
database = self.db
self.evolve_and_check_noop()
self.assertEqual(SomeModel.select().first().some_field, 'woot')
def test_reorder_multi_index(self):
class SomeModel(pw.Model):
some_field = pw.CharField(null=True)
class Meta:
database = self.db
indexes = (
(('id', 'some_field'), False),
)
self.evolve_and_check_noop()
self.assertEqual(sorted(peeweedbevolve.normalize_indexes(peeweedbevolve.get_indexes_by_table(self.db,'somemodel'))), [(u'somemodel', (u'id',), True), (u'somemodel', (u'id',u'some_field'), False)])
peeweedbevolve.clear()
class SomeModel(pw.Model):
some_field = pw.CharField(null=True)
class Meta:
database = self.db
indexes = (
(('some_field', 'id'), False),
)
self.evolve_and_check_noop()
self.assertEqual(sorted(peeweedbevolve.normalize_indexes(peeweedbevolve.get_indexes_by_table(self.db,'somemodel'))), [(u'somemodel', (u'id',), True), (u'somemodel', (u'some_field',u'id'), False)])
def test_drop_column_default(self):
class SomeModel(pw.Model):
some_field = pw.CharField(null=True, default='woot2')
class Meta:
database = self.db
self.evolve_and_check_noop()
model = SomeModel.create()
self.assertEqual(model.some_field, 'woot2')
peeweedbevolve.clear()
class SomeModel(pw.Model):
some_field = pw.CharField(null=True)
class Meta:
database = self.db
self.evolve_and_check_noop()
model = SomeModel.create()
self.assertEqual(model.some_field, None)
self.assertEqual(SomeModel.get(SomeModel.id==model.id).some_field, None)
def test_dont_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()
class SomeModel(pw.Model):
some_field = pw.CharField(null=True)
class Meta:
database = self.db
evolve = False
self.evolve_and_check_noop()
# doesn't fail because table is still there
SomeModel.create(some_field='woot2')
def test_dont_add_column(self):
class SomeModel(pw.Model):
some_field = pw.CharField(null=True)
class Meta:
database = self.db
self.evolve_and_check_noop()
peeweedbevolve.clear()
class SomeModel(pw.Model):
some_field = pw.CharField(null=True)
some_other_field = pw.CharField(null=False)
class Meta:
database = self.db
evolve = False
self.evolve_and_check_noop()
# should not fail because the not-null column wasn't added
SomeModel.create(some_field='woot')
def test_order_by_inheritance():
class Base(TestModel):
created = DateTimeField()
class Meta:
order_by = ('-created',)
class Foo(Base):
data = CharField()
class Bar(Base):
val = IntegerField()
class Meta:
order_by = ('-val',)
foo_order_by = Foo._meta.order_by[0]
assert isinstance(foo_order_by, Field)
assert foo_order_by.model_class is Foo
assert foo_order_by.name == 'created'
bar_order_by = Bar._meta.order_by[0]
assert isinstance(bar_order_by, Field)
assert bar_order_by.model_class is Bar
assert bar_order_by.name == 'val'
def addSong(self, fid, filename, pid=None,artist=None, name=None):
if pid == '':
pid = 'None'
# query = 'INSERT INTO `tracks` (`id`,`filename`,`playlistID`) VALUES ("{}","{}","{}");'.format(fid, filename, pid)
#
self.CURSOR.execute("""INSERT INTO tracks (id,filename,playlistID,artist,name) VALUES(%s,%s,%s,%s,%s)""", (fid,filename,pid,artist,name))
self.db_conn.commit()
result = self.CURSOR.fetchone()
print("Adding new track to tracks table...")
#print(query)
print(result)
#self.CURSOR.close()
return 1
# db = MySQLDatabase('jonhydb', user='john',passwd='megajonhy')
# class Book(peewee.Model):
# author = peewee.CharField()
# title = peewee.TextField()
# class Meta:
# database = db
# Book.create_table()
# book = Book(author="me", title='Peewee is cool')
# book.save()
# for book in Book.filter(author="me"):
# print book.title
def test_add_column(self):
another_column = CharField(null=True)
migrate(migrator.add_column('food', 'another_column', another_column))
self.assertTableHasColumn('food', 'another_column', CharField)
self.assertTableHasColumn('foodversion', 'another_column', CharField)
def add_column(self, table, name, coltype, **kwargs):
"""
Add the given column to the given table.
:param table: Table name to add column to.
:param name: Name of the column field to add.
:param coltype: Column type (from FIELD_TO_PEEWEE).
:param kwargs: Arguments for the given column type.
:return: None
"""
field_class = FIELD_TO_PEEWEE.get(coltype, peewee.CharField)
self.migrator.add_column(table, name, field_class(**kwargs)).run()
def startup_library_service():
wamp = autobahn_sync.AutobahnSync()
wamp.run()
db = pw.SqliteDatabase('books.db')
class Book(pw.Model):
title = pw.CharField()
author = pw.CharField()
class Meta:
database = db
try:
db.create_table(Book)
except pw.OperationalError:
pass
@wamp.register('com.library.get_book')
def get_book(id):
try:
b = Book.get(id=id)
except pw.DoesNotExist:
return {'_error': "Doesn't exist"}
return {'id': id, 'title': b.title, 'author': b.author}
@wamp.register('com.library.new_book')
def new_book(title, author):
book = Book(title=title, author=author)
book.save()
wamp.session.publish('com.library.book_created', book.id)
return {'id': book.id}
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_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_rename_column_aka_list(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_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)
def test_rename_table_rename_column(self):
self.test_create_table()
peeweedbevolve.clear()
class SomeOtherModel(pw.Model):
some_other_field = pw.CharField(null=True, aka='some_field')
class Meta:
database = self.db
aka = 'somemodel'
self.evolve_and_check_noop()
self.assertEqual(SomeOtherModel.select().first().some_other_field, 'woot')
def test_add_not_null_constraint(self):
self.test_create_table()
peeweedbevolve.clear()
class SomeModel(pw.Model):
some_field = pw.CharField(null=False)
class Meta:
database = self.db
self.evolve_and_check_noop()
self.assertEqual(SomeModel.select().first().some_field, 'woot')
with self.db.atomic() as txn:
self.assertRaises(Exception, lambda: SomeModel.create(some_field=None))
def test_remove_not_null_constraint(self):
self.test_add_not_null_constraint()
peeweedbevolve.clear()
class SomeModel(pw.Model):
some_field = pw.CharField(null=True)
class Meta:
database = self.db
self.evolve_and_check_noop()
SomeModel.create()
def test_rename_column_add_not_null_constraint(self):
self.test_create_table()
peeweedbevolve.clear()
class SomeModel(pw.Model):
some_other_field = pw.CharField(null=False, aka='some_field')
class Meta:
database = self.db
self.evolve_and_check_noop()
self.assertEqual(SomeModel.select().first().some_other_field, 'woot')
with self.db.atomic() as txn:
self.assertRaises(Exception, lambda: SomeModel.create(some_other_field=None))
def test_rename_table_rename_column_add_not_null_constraint(self):
self.test_create_table()
peeweedbevolve.clear()
class SomeOtherModel(pw.Model):
some_other_field = pw.CharField(null=False, aka='some_field')
class Meta:
database = self.db
aka = 'somemodel'
self.evolve_and_check_noop()
self.assertEqual(SomeOtherModel.select().first().some_other_field, 'woot')
with self.db.atomic() as txn:
self.assertRaises(Exception, lambda: SomeOtherModel.create(some_other_field=None))
def test_add_index(self):
self.test_create_table()
peeweedbevolve.clear()
class SomeModel(pw.Model):
some_field = pw.CharField(index=True, null=True)
class Meta:
database = self.db
self.evolve_and_check_noop()
self.assertEqual(sorted(peeweedbevolve.normalize_indexes(peeweedbevolve.get_indexes_by_table(self.db,'somemodel'))), [(u'somemodel', (u'id',), True), (u'somemodel', (u'some_field',), False)])
def test_add_index_table_rename(self):
self.test_create_table()
peeweedbevolve.clear()
class SomeModel2(pw.Model):
some_field = pw.CharField(index=True, null=True)
class Meta:
database = self.db
aka = 'somemodel'
self.evolve_and_check_noop()
self.assertEqual(sorted(peeweedbevolve.normalize_indexes(peeweedbevolve.get_indexes_by_table(self.db,'somemodel2'))), [(u'somemodel2', (u'id',), True), (u'somemodel2', (u'some_field',), False)])
def test_add_index_column_rename(self):
self.test_create_table()
peeweedbevolve.clear()
class SomeModel(pw.Model):
some_field2 = pw.CharField(index=True, null=True, aka='some_field')
class Meta:
database = self.db
self.evolve_and_check_noop()
self.assertEqual(sorted(peeweedbevolve.normalize_indexes(peeweedbevolve.get_indexes_by_table(self.db,'somemodel'))), [(u'somemodel', (u'id',), True), (u'somemodel', (u'some_field2',), False)])