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

def init(path):
    """Initialize the database, create missing tables."""
    db.init(path)

    try:
        migrator = playhouse.migrate.SqliteMigrator(db)
        playhouse.migrate.migrate(
            migrator.add_column(
                'ChannelConfig', 'gibber', peewee.BooleanField(null=True)))
    except peewee.OperationalError:
        pass

    db.connect()
    db.create_tables([
        Tell, Message, Quote, Memo,
        Subscriber, Restricted, Alert, ChannelConfig], safe=True)

def _getModel(self):
        class Doc(Model):
            key = CharField(index=True, default="")
            gitHostRefs = CharField(index=True, default="")
            title = CharField(index=True, default="")
            creationTime = TimestampField(index=True, default=j.data.time.epoch)
            modTime = TimestampField(index=True, default=j.data.time.epoch)
            inGithub = BooleanField(index=True, default=False)
            labels = CharField(index=True, default="")
            assignees = CharField(index=True, default="")
            milestone = CharField(index=True, default="")
            priority = CharField(index=True, default="minor")
            type = CharField(index=True, default="unknown")
            state = CharField(index=True, default="new")
            content = TextField(index=False, default="")
            repo = TextField(index=True, default="")
            isClosed = BooleanField(index=True, default=False)

            class Meta:
                database = j.tools.docmanager.indexDB

        return Doc

def test_add_not_null_constraint_with_records_and_false_default(self):
    class SomeModel(pw.Model):
      some_field = pw.BooleanField(null=True)
      class Meta:
        database = self.db
    self.evolve_and_check_noop()
    SomeModel.create(some_field=None)
    peeweedbevolve.clear()
    class SomeModel(pw.Model):
      some_field = pw.BooleanField(null=False, default=False)
      class Meta:
        database = self.db
    self.evolve_and_check_noop()
    self.assertEqual(SomeModel.select().first().some_field, False)

def _get_args(self, args):
        pw_args = []
        for field_name, op, value in args:
            field = self.view.fields[field_name]
            if isinstance(field, peewee.ForeignKeyField):
                tfield = field.to_field
            else:
                tfield = field

            conv_func = None
            # ?????? peewee ????? int/float ???????????
            if isinstance(tfield, peewee.BlobField):
                conv_func = to_bin
            elif isinstance(tfield, peewee.BooleanField):
                conv_func = bool_parse

            if conv_func:
                try:
                    if op == 'in':
                        value = list(map(conv_func, value))
                    else:
                        value = conv_func(value)
                except binascii.Error:
                    self.err = RETCODE.INVALID_HTTP_PARAMS, 'Invalid query value for blob: Odd-length string'
                    return
                except ValueError as e:
                    self.err = RETCODE.INVALID_HTTP_PARAMS, ' '.join(map(str, e.args))

            pw_args.append(getattr(field, _peewee_method_map[op])(value))
        return pw_args

def user_model(database):
    """Fixture that provides a simple User model."""
    class User(Model):
        email = peewee.CharField(max_length=100, null=False, unique=True)
        password = PasswordField(null=False, iterations=4)
        active = peewee.BooleanField(null=False, default=True)

        @classmethod
        def base_query(cls):
            # Only query for active Users
            return super(User, cls).base_query().where(cls.active == True)

    try:
        @pre_save(sender=User)
        def validate_email(sender, instance, created):
            # Ensure that the email is valid.
            assert '@' in instance.email

        @post_save(sender=User)
        def send_welcome_email(sender, instance, created):
            # Send the user an email when they are created initially.
            if created:
                print('Sending welcome email to {}'.format(instance.email))
    except ValueError:
        # This gets hit because you can't connect an event listener more than
        # once and for some reason this fixture is sticking around.
        pass

    User.create_table(True)

    yield User

def add2index(self, **args):
        """
        key = CharField(index=True, default="")
        gitHostRefs = CharField(index=True, default="")
        title = CharField(index=True, default="")
        creationTime = TimestampField(index=True, default=j.data.time.epoch)
        modTime = TimestampField(index=True, default=j.data.time.epoch)
        inGithub = BooleanField(index=True, default=False)
        labels = CharField(index=True, default="")
        assignees = CharField(index=True, default="")
        milestone = CharField(index=True, default="")
        priority = CharField(index=True, default="minor")
        type = CharField(index=True, default="unknown")
        state = CharField(index=True, default="new")
        content = TextField(index=False, default="")
        repo = TextField(index=True, default="")

        @param args is any of the above

        assignees & labels can be given as:
            can be "a,b,c"
            can be "'a','b','c'"
            can be ["a","b","c"]
            can be "a"

        """

        if "gitHostRefs" in args:
            args["gitHostRefs"] = ["%s_%s_%s" % (item["name"], item["id"], item['url']) for item in args["gitHostRefs"]]

        args = self._arraysFromArgsToString(["assignees", "labels", "gitHostRefs"], args)

        # this will try to find the right index obj, if not create
        obj, isnew = self.index.get_or_create(key=args["key"])

        for key, item in args.items():
            if key in obj._data:
                # print("%s:%s" % (key, item))
                obj._data[key] = item

        obj.save()

def qs_filter(cls, qs, flt, value, process_value=True):
        """
        Private method to set WHERE part of query.
        If required, Django-style filter is available via qs.filter()
        and peewee.DQ - this method provides joins.

        Filter relational operators are:
        * NOT - '-', not operator, should be user as prefix
        * < - 'lt', less than
        * > - 'gt', greater than
        * <= - 'lte', less than or equal
        * >= - 'gte', greater than or equal
        * != - 'ne', not equal
        * LIKE - 'like', classic like operator
        * ILIKE - 'ilike', case-insensitive like operator
        * IN - 'in', classic in. Values should be separated by comma
        * ISNULL - 'isnull', operator to know if smth is equal to null. Use -<fieldname>__isnull for IS NOT NULL
        """
        neg = False
        if flt[0] in '-':
            # Register NOT filter clause
            neg = True
            flt = flt[1:]
        fld_name, _, k = flt.rpartition('__')
        if not fld_name:
            # No underscore, simple filter
            fld_name, k = k, ''

        # Get filter
        op = FILTER_MAP.get(k, operator.eq)

        if neg:
            _op = op
            op = lambda f, x: operator.inv(_op(f, x))

        # Get field from model
        # raised AttributeError should be handled on higher level
        fld = getattr(cls.model_cls, fld_name)

        # Additional value processing
        if process_value:
            _v = value.decode()
            if isinstance(fld, peewee.BooleanField) and _v in ('0', 'f'):
                # Assume that '0' and 'f' are FALSE for boolean field
                _v = False
            elif k == 'in':
                # Force set parameter to list
                _v = _v.split(',')
            elif k == 'isnull':
                # ISNULL. Force set parameter to None
                _v = None
        else:
            _v = value

        # Send parameter to ORM
        return qs.where(op(fld, _v))

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)