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

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_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)