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)
python类IntegerField()的实例源码
def test_change_int_column_to_fk(self):
class Person(pw.Model):
class Meta:
database = self.db
class Car(pw.Model):
owner_id = pw.IntegerField(null=False)
class Meta:
database = self.db
self.evolve_and_check_noop()
person = Person.create()
car = Car.create(owner_id=person.id)
peeweedbevolve.unregister(Car)
class Car(pw.Model):
owner = pw.ForeignKeyField(rel_model=Person, null=False)
class Meta:
database = self.db
self.evolve_and_check_noop()
self.assertEqual(Car.select().first().owner_id, person.id)
self.assertRaises(Exception, lambda: Car.create(owner=-1))
def test_change_fk_column_to_int(self):
class Person(pw.Model):
class Meta:
database = self.db
class Car(pw.Model):
owner = pw.ForeignKeyField(rel_model=Person, null=False)
class Meta:
database = self.db
self.evolve_and_check_noop()
person = Person.create()
car = Car.create(owner=person)
peeweedbevolve.unregister(Car)
class Car(pw.Model):
owner_id = pw.IntegerField(null=False)
class Meta:
database = self.db
self.evolve_and_check_noop()
self.assertEqual(Car.select().first().owner_id, person.id)
Car.create(owner_id=-1) # this should not fail
def test_change_integer_to_fake_fk_column(self):
class Person(pw.Model):
class Meta:
database = self.db
class Car(pw.Model):
owner_id = pw.IntegerField(null=False)
class Meta:
database = self.db
self.evolve_and_check_noop()
car = Car.create(owner_id=-1)
peeweedbevolve.unregister(Car)
class Car(pw.Model):
owner = pw.ForeignKeyField(rel_model=Person, null=False, fake=True)
class Meta:
database = self.db
self.evolve_and_check_noop()
person = Person.create()
car = Car.create(owner=-2)
self.assertEqual(Car.select().count(), 2)
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 foreign_key(self, coltype, name, references, **kwargs):
"""
Add a foreign key to the model.
This has some special cases, which is why it's not handled like all the other column types.
:param name: Name of the foreign key.
:param references: Table name in the format of "table.column" or just
"table" (and id will be default column).
:param kwargs: Additional kwargs to pass to the column instance.
You can also provide "on_delete" and "on_update" to add constraints.
:return: None
"""
try:
rel_table, rel_column = references.split('.', 1)
except ValueError:
rel_table, rel_column = references, 'id'
# Create a dummy model that we can relate this field to.
# Add the foreign key as a local field on the dummy model.
# We only do this so that Peewee can generate the nice foreign key constraint for us.
class DummyRelated(peewee.Model):
class Meta:
primary_key = False
database = peewee.Proxy()
db_table = rel_table
rel_field_class = FIELD_TO_PEEWEE.get(coltype, peewee.IntegerField)
rel_field = rel_field_class()
rel_field.add_to_class(DummyRelated, rel_column)
field = peewee.ForeignKeyField(DummyRelated, db_column=name, to_field=rel_column, **kwargs)
field.add_to_class(self.model, name)
def generate_sqlite_db(variants_to_process, temp_sqlite_db_path, sqlite_db_path):
logging.info("populating sqlite database: " + temp_sqlite_db_path)
if os.path.isfile(temp_sqlite_db_path):
run("rm -f " + temp_sqlite_db_path)
sqlite_db = peewee.SqliteDatabase(temp_sqlite_db_path, autocommit=False)
class t(_SharedVariantPositionFields):
n_expected_samples = peewee.IntegerField(index=True, null=True)
n_available_samples = peewee.IntegerField(index=True, null=True)
class Meta:
database = sqlite_db
indexes = (
(('chrom', 'pos', 'ref', 'alt', 'het_or_hom_or_hemi'), True), # True means unique index
)
t.create_table(fail_silently=True)
# copy the records from the Variant table used by generate_HC_bams.py
sqlite_db.connect()
with sqlite_db.atomic():
for v in variants_to_process: #Variant.select().where(Variant.finished==1).dicts():
#shortcuts.model_to_dict(v)
d = {
'chrom': v.chrom, 'pos': v.pos, 'ref': v.ref, 'alt': v.alt, 'het_or_hom_or_hemi': v.het_or_hom_or_hemi,
'n_expected_samples': v.n_expected_samples,
'n_available_samples': v.n_available_samples,
}
# delete readviz_bam_paths as they're no longer relevant because the data from these is being combined into one bam file
#print("INSERTING " + str(d))
t.insert(**d).execute()
sqlite_db.close()
run("mv %s %s" % (temp_sqlite_db_path, sqlite_db_path))
def main(config):
config = load_config(config)
database.init(**config['processing_database'])
database.connect()
migrator = MySQLMigrator(database)
run_type_key = IntegerField(null=True)
run_type_name = CharField(null=True)
migrate(
migrator.add_column('raw_data_files', 'run_type_key', run_type_key),
migrator.add_column('raw_data_files', 'run_type_name', run_type_name)
)
def main(config):
config = load_config(config)
database.init(**config['processing_database'])
database.connect()
migrator = MySQLMigrator(database)
drs_step = IntegerField(null=True)
roi = IntegerField(null=True)
migrate(
migrator.add_column('raw_data_files', 'roi', roi),
migrator.add_column('drs_files', 'roi', roi),
migrator.add_column('drs_files', 'drs_step', drs_step),
)
def test_column_aliases():
tc = TableCreator('awesome')
tc.bare('col_bare')
tc.biginteger('col_biginteger')
tc.binary('col_binary')
tc.blob('col_blob')
tc.bool('col_bool')
tc.char('col_char')
tc.date('col_date')
tc.datetime('col_datetime')
tc.decimal('col_decimal')
tc.double('col_double')
tc.fixed('col_fixed')
tc.float('col_float')
tc.int('col_int')
tc.integer('col_integer')
tc.smallint('col_smallint')
tc.smallinteger('col_smallinteger')
tc.text('col_text')
tc.time('col_time')
tc.uuid('col_uuid')
assert isinstance(tc.model.col_bare, peewee.BareField)
assert isinstance(tc.model.col_biginteger, peewee.BigIntegerField)
assert isinstance(tc.model.col_binary, peewee.BlobField)
assert isinstance(tc.model.col_blob, peewee.BlobField)
assert isinstance(tc.model.col_bool, peewee.BooleanField)
assert isinstance(tc.model.col_char, peewee.CharField)
assert isinstance(tc.model.col_date, peewee.DateField)
assert isinstance(tc.model.col_datetime, peewee.DateTimeField)
assert isinstance(tc.model.col_decimal, peewee.DecimalField)
assert isinstance(tc.model.col_double, peewee.DoubleField)
assert isinstance(tc.model.col_fixed, peewee.CharField)
assert isinstance(tc.model.col_float, peewee.FloatField)
assert isinstance(tc.model.col_int, peewee.IntegerField)
assert isinstance(tc.model.col_integer, peewee.IntegerField)
assert isinstance(tc.model.col_smallint, peewee.SmallIntegerField)
assert isinstance(tc.model.col_smallinteger, peewee.SmallIntegerField)
assert isinstance(tc.model.col_text, peewee.TextField)
assert isinstance(tc.model.col_time, peewee.TimeField)
assert isinstance(tc.model.col_uuid, peewee.UUIDField)
def test_column():
tc = TableCreator('awesome')
tc.primary_key('id')
tc.column('bare', 'col_bare')
tc.column('biginteger', 'col_biginteger')
tc.column('binary', 'col_binary')
tc.column('blob', 'col_blob')
tc.column('bool', 'col_bool')
tc.column('char', 'col_char')
tc.column('date', 'col_date')
tc.column('datetime', 'col_datetime')
tc.column('decimal', 'col_decimal')
tc.column('double', 'col_double')
tc.column('fixed', 'col_fixed')
tc.column('float', 'col_float')
tc.column('int', 'col_int')
tc.column('integer', 'col_integer')
tc.column('smallint', 'col_smallint')
tc.column('smallinteger', 'col_smallinteger')
tc.column('text', 'col_text')
tc.column('time', 'col_time')
tc.column('uuid', 'col_uuid')
assert isinstance(tc.model.id, peewee.PrimaryKeyField)
assert isinstance(tc.model.col_bare, peewee.BareField)
assert isinstance(tc.model.col_biginteger, peewee.BigIntegerField)
assert isinstance(tc.model.col_binary, peewee.BlobField)
assert isinstance(tc.model.col_blob, peewee.BlobField)
assert isinstance(tc.model.col_bool, peewee.BooleanField)
assert isinstance(tc.model.col_char, peewee.CharField)
assert isinstance(tc.model.col_date, peewee.DateField)
assert isinstance(tc.model.col_datetime, peewee.DateTimeField)
assert isinstance(tc.model.col_decimal, peewee.DecimalField)
assert isinstance(tc.model.col_double, peewee.DoubleField)
assert isinstance(tc.model.col_fixed, peewee.CharField)
assert isinstance(tc.model.col_float, peewee.FloatField)
assert isinstance(tc.model.col_int, peewee.IntegerField)
assert isinstance(tc.model.col_integer, peewee.IntegerField)
assert isinstance(tc.model.col_smallint, peewee.SmallIntegerField)
assert isinstance(tc.model.col_smallinteger, peewee.SmallIntegerField)
assert isinstance(tc.model.col_text, peewee.TextField)
assert isinstance(tc.model.col_time, peewee.TimeField)
assert isinstance(tc.model.col_uuid, peewee.UUIDField)
