Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import time
from models import Journal
from pony.orm import db_session, select
LEVEL_CHOICE = [10, 20, 30, 40, 50]
start = time.time()
count = 0
with db_session():
for _ in range(10):
for level in LEVEL_CHOICE:
res = list(select(j for j in Journal if j.level == level))
count += len(res)
now = time.time()
print(f'Pony ORM, D: Rows/sec: {count / (now - start): 10.2f}')
Returns:
An index updated with new documents.
"""
doc_count = self._whoosh.doc_count()
objs = orm.count(e for e in self._model)
field_names = set(self._schema_attrs.keys())
missings = set(self._whoosh.schema.names())
for f in list(field_names - missings):
self.add_field(f, fields.TEXT(self.kw))
if doc_count == 0 and objs > 0:
writer = self._whoosh.writer()
for obj in orm.select(e for e in self._model):
attrs = {self._primary_key: obj.get_pk()}
for f in self._schema_attrs.keys():
attrs[f] = unicode(getattr(obj, f))
writer.add_document(**attrs)
writer.commit()
def texts():
texts = orm.select(t for t in db.Text)[:]
return jsonify([format(t) for t in texts])
async def fetch_playlists(self):
with db_session:
fetched_ids = set(select(p.id for p in Playlist))
for user in self.USER_LIST:
user_playlists = await self.user_playlists(user)
async for playlist in user_playlists.iterall(ignore_exceptions=True):
if not playlist:
continue
logger.info("Got %s", playlist.name)
if playlist.id not in fetched_ids:
Playlist.from_dict(playlist)
fetched_ids.add(playlist.id)
def blacklisted(user):
"""Checks if given user is blacklisted from the bot.
Params
=======
user: discord.User
Returns
=======
If the user is blacklisted: bool"""
with db_session:
return select(u for u in db.Blacklist if u.user_id == user.id).exists()
def get_thankers(self, after_date=None, before_date=None):
if after_date:
user_ids = orm.select(rv.user_id for rv in collipa.models.Thank if rv.image_id == self.id and rv.created_at > after_date)
elif before_date:
user_ids = orm.select(rv.user_id for rv in collipa.models.Thank if rv.image_id == self.id and rv.created_at < before_date)
else:
user_ids = orm.select(rv.user_id for rv in collipa.models.Thank if rv.image_id == self.id)
users = []
if user_ids:
user_ids = user_ids.order_by(lambda: orm.desc(rv.created_at))
users = orm.select(rv for rv in collipa.models.User if rv.id in user_ids)
return users
def items():
li = []
descriptions = orm.select([item.url, item.description] for item in db.StoredItem)[:]
for url, d in descriptions:
li.append({
'url': url,
'description': d,
})
return jsonify(li)
rv.node_id == self.id and
rv.created_at > ago)
topics = topics.order_by(lambda: orm.desc((rv.collect_count +
rv.thank_count - rv.report_count) * 10 +
(rv.up_count - rv.down_count) * 5 +
rv.reply_count * 3))
mc.set('node_%s_hot_topics' % self.id, list(topics),
60 * 60 * 3)
order_by = 'none'
elif category == 'latest':
topics = orm.select(rv for rv in collipa.models.Topic if rv.node_id == self.id)
elif category == 'desert':
topics = orm.select(rv for rv in collipa.models.Topic if
rv.node_id == self.id and rv.reply_count == 0)
else:
topics = orm.select(rv for rv in collipa.models.Topic if
rv.node_id == self.id and rv.role == category)
order_by = 'last_reply_date'
if order_by == 'last_reply_date':
topics = topics.order_by(lambda: orm.desc(rv.last_reply_date))
elif order_by == 'created_at':
topics = topics.order_by(lambda: orm.desc(rv.created_at))
elif order_by == 'active':
topics = topics.order_by(lambda: orm.desc(rv.active))
elif order_by == 'smart':
topics = topics.order_by(lambda: orm.desc((rv.collect_count +
rv.thank_count -
rv.report_count) * 10 +
(rv.up_count -
rv.down_count) * 5 +
rv.reply_count * 3))
def get_uppers(self, after_date=None, before_date=None):
if after_date:
user_ids = orm.select(rv.user_id for rv in collipa.models.Up if rv.album_id == self.id and rv.created_at > after_date)
elif before_date:
user_ids = orm.select(rv.user_id for rv in collipa.models.Up if rv.album_id == self.id and rv.created_at < before_date)
else:
user_ids = orm.select(rv.user_id for rv in collipa.models.Up if rv.album_id == self.id)
users = []
if user_ids:
user_ids = user_ids.order_by(lambda: orm.desc(rv.created_at))
users = orm.select(rv for rv in collipa.models.User if rv.id in user_ids)
return users
def get_root_user(cls):
return orm.select(x for x in cls if x.name == config.root_user).first()