Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
class UUIDTestModel(peewee.Model):
id = peewee.UUIDField(primary_key=True, default=uuid.uuid4)
text = peewee.CharField()
def __str__(self):
return '<%s id=%s> %s' % (self.__class__.__name__, self.id, self.text)
class CompositeTestModel(peewee.Model):
"""A simple "through" table for many-to-many relationship."""
uuid = peewee.ForeignKeyField(UUIDTestModel)
alpha = peewee.ForeignKeyField(TestModelAlpha)
class Meta:
primary_key = peewee.CompositeKey('uuid', 'alpha')
####################
# Base tests class #
####################
class BaseManagerTestCase(unittest.TestCase):
only = None
models = [TestModel, UUIDTestModel, TestModelAlpha,
TestModelBeta, TestModelGamma, CompositeTestModel]
@classmethod
@contextlib.contextmanager
def manager(cls, objects, allow_sync=False):
### PEEWEE FIELDS ###
entity_id = peewee.IntegerField(index=False)
entity_type = peewee.IntegerField(index=False)
name = peewee.TextField(index=True)
relation_counts = postgres_ext.BinaryJSONField(null=True, index=False)
metadata = postgres_ext.BinaryJSONField(null=True, index=False)
entities = postgres_ext.BinaryJSONField(null=True, index=False)
search_content = postgres_ext.TSVectorField(index=True)
### PEEWEE META ###
class Meta:
db_table = 'entities'
primary_key = peewee.CompositeKey('entity_type', 'entity_id')
### PUBLIC METHODS ###
@classmethod
def bootstrap(cls):
cls.drop_table(True)
cls.create_table()
cls.bootstrap_pass_one()
cls.bootstrap_pass_two()
@classmethod
def bootstrap_pass_one(cls):
PostgresModel.bootstrap_pass_one(
cls,
'artist',
id_attr='entity_id',
f_party_id = CharField(max_length=32)
f_id_type = CharField(max_length=16)
f_encrypt_type = CharField(max_length=16)
f_tag = CharField(max_length=16, default=DEFAULT_TAG)
f_namespcae = CharField(max_length=128)
f_version = CharField(max_length=128)
f_rsa_key_n = CharField(max_length=512)
f_rsa_key_d = CharField(max_length=512)
f_rsa_key_e = CharField(max_length=32)
f_create_datetime = DateTimeField(default=datetime.datetime.now)
f_update_datetime = DateTimeField(default=datetime.datetime.now)
f_description = TextField(null=True, default='')
class Meta:
db_table = "t_id_library_cache_info"
primary_key = CompositeKey('f_party_id', 'f_id_type', 'f_encrypt_type', 'f_tag', 'f_namespcae', 'f_version')
link_type = CharField()
streaming_service_type = CharField(default=StreamingServiceType.SPOTIFY)
created_at = DateTimeField()
updated_at = DateTimeField(null=True)
times_sent = IntegerField(default=1)
artist_name = CharField(null=True)
album_name = CharField(null=True)
track_name = CharField(null=True)
genre = CharField(null=True)
user = ForeignKeyField(User, backref='links')
chat = ForeignKeyField(Chat, backref='links')
last_update_user = ForeignKeyField(User, backref='updated_links', null=True)
class Meta:
database = db
primary_key = CompositeKey('url', 'chat')
def apply_update(self, user):
"""
Set the update fields to the current values
"""
self.updated_at = datetime.datetime.now()
self.last_update_user = user
self.times_sent += 1
def __str__(self):
return 'Link: {}'.format(self.url)
from discograph.library.sqlite.SqliteModel import SqliteModel
class SqliteEntity(SqliteModel):
### PEEWEE FIELDS ###
name = peewee.TextField()
entity_id = peewee.IntegerField(null=False)
entity_type = peewee.IntegerField(null=False)
### PEEWEE META ###
class Meta:
db_table = 'entity'
primary_key = peewee.CompositeKey('entity_id', 'entity_type')
### PRIVATE METHODS ###
@classmethod
def _load_from_mongo_class(cls, mongo_class):
import discograph
entity_type = 1
if mongo_class == discograph.Label:
entity_type = 2
query = mongo_class.objects().no_cache().timeout(False)
query = query.only('discogs_id', 'name')
count = query.count()
rows = []
for i, mongo_document in enumerate(query, 1):
if mongo_document.discogs_id and mongo_document.name:
rows.append(dict(
f_play_id = CharField(max_length=100)
f_pid = IntegerField(null=True)
f_play_name = CharField(max_length=500, null=True, default='')
f_play_conf = LongTextField()
f_status = CharField(max_length=50)
f_create_time = BigIntegerField()
f_update_time = BigIntegerField(null=True)
f_start_time = BigIntegerField(null=True)
f_end_time = BigIntegerField(null=True)
f_hosts = LongTextField(null=True)
f_roles = LongTextField(null=True)
f_elapsed = IntegerField(null=True)
class Meta:
db_table = "t_play"
primary_key = CompositeKey('f_job_id', 'f_play_id')
class Task(DataBaseModel):
f_job_id = CharField(max_length=100)
f_play_id = CharField(max_length=100)
f_task_id = CharField(max_length=100) # NEW
f_role = CharField(max_length=200, null=True)
f_task_name = CharField(max_length=200, null=True) # NEW
f_status = CharField(max_length=50)
f_host = CharField(max_length=100, null=True)
f_create_time = BigIntegerField()
f_update_time = BigIntegerField(null=True)
f_start_time = BigIntegerField(null=True)
f_end_time = BigIntegerField(null=True)
f_elapsed = IntegerField(null=True)
class MovieSample(database.Model):
mid = ForeignKeyField(Movie)
img = CharField()
class Meta:
primary_key = CompositeKey('mid', 'img')
db_table = 'movie_sample'
class MovieCate(database.Model):
mid = ForeignKeyField(Movie)
cate = CharField()
class Meta:
primary_key = CompositeKey('mid', 'cate')
db_table = 'movie_cate'
if __name__ == '__main__':
app.run()