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)
python类BooleanField()的实例源码
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)
