Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@cached(wrong_param=1)
async def fn(n):
return n
def test_inheritance(self):
assert isinstance(cached_stampede(), cached)
@cached(ttl=3600, timeout=1200)
async def get_trending_tags(context, start_tag: str = '', limit: int = 250):
"""Get top 250 trending tags among pending posts, with stats."""
limit = valid_limit(limit, ubound=250)
start_tag = valid_tag(start_tag or '', allow_empty=True)
if start_tag:
seek = """
HAVING SUM(payout) <= (
SELECT SUM(payout)
FROM hive_posts_cache
WHERE is_paidout = '0'
AND category = :start_tag)
"""
else:
seek = ''
@cached(ttl=3600, timeout=1200)
async def get_trending_tags(context, start_tag: str = '', limit: int = 250):
"""Get top 250 trending tags among pending posts, with stats."""
limit = valid_limit(limit, ubound=250)
start_tag = valid_tag(start_tag or '', allow_empty=True)
if start_tag:
seek = """
HAVING SUM(payout) <= (
SELECT SUM(payout)
FROM hive_posts_cache
WHERE is_paidout = '0'
AND category = :start_tag)
"""
else:
seek = ''
@cached(ttl=7200)
async def payouts_total():
"""Get total sum of all completed payouts."""
# memoized historical sum. To update:
# SELECT SUM(payout) FROM hive_posts_cache
# WHERE is_paidout = 1 AND payout_at <= precalc_date
precalc_date = '2017-08-30 00:00:00'
precalc_sum = Decimal('19358777.541')
# sum all payouts since `precalc_date`
sql = """
SELECT SUM(payout) FROM hive_posts_cache
WHERE is_paidout = '1' AND payout_at > '%s'
""" % (precalc_date)
return float(precalc_sum + DB.query_one(sql)) #TODO: decimal
@cached(ttl = 30 * 60)
async def getUsersEffectiveRating(*, activeOnly=None):
""" Returns a dictionary mapping user handle to his effective rating for all the users.
"""
ratedList = await cf.user.ratedList(activeOnly=activeOnly)
users_effective_rating_dict = {user.handle: user.effective_rating
for user in ratedList}
return users_effective_rating_dict
@cached(key="my_custom_key", ttl=30, alias="default")
async def expensive_call():
log.info("Expensive has been called")
await asyncio.sleep(3)
# You are storing the whole dict under "my_custom_key"
return {"test": str(uuid.uuid4())}
@cached(key="my_custom_key", serializer=JsonSerializer(), timeout=0)
async def time(self):
return {"time": datetime.now().isoformat()}