本文整理汇总了Python中redash.utils.utcnow函数的典型用法代码示例。如果您正苦于以下问题:Python utcnow函数的具体用法?Python utcnow怎么用?Python utcnow使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了utcnow函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_get_latest_returns_the_last_cached_result_for_negative_ttl
def test_get_latest_returns_the_last_cached_result_for_negative_ttl(self):
yesterday = utcnow() + datetime.timedelta(days=-100)
very_old = self.factory.create_query_result(retrieved_at=yesterday)
yesterday = utcnow() + datetime.timedelta(days=-1)
qr = self.factory.create_query_result(retrieved_at=yesterday)
found_query_result = models.QueryResult.get_latest(qr.data_source, qr.query_text, -1)
self.assertEqual(found_query_result.id, qr.id)
开发者ID:doomsplayer,项目名称:redash,代码行数:9,代码来源:test_models.py
示例2: test_schedule_until_before
def test_schedule_until_before(self):
"""
Queries with non-null ``schedule['until']`` are reported by
Query.outdated_queries() before the given time is past.
"""
one_day_from_now = (utcnow() + datetime.timedelta(days=1)).strftime("%Y-%m-%d")
two_hours_ago = utcnow() - datetime.timedelta(hours=2)
query = self.factory.create_query(schedule={'interval':'3600', 'until':one_day_from_now, 'time': None, 'day_of_week':None})
query_result = self.factory.create_query_result(query=query.query_text, retrieved_at=two_hours_ago)
query.latest_query_data = query_result
queries = models.Query.outdated_queries()
self.assertIn(query, queries)
开发者ID:ariarijp,项目名称:redash,代码行数:13,代码来源:test_models.py
示例3: test_get_latest_returns_if_ttl_not_expired
def test_get_latest_returns_if_ttl_not_expired(self):
yesterday = utcnow() - datetime.timedelta(seconds=30)
qr = self.factory.create_query_result(retrieved_at=yesterday)
found_query_result = models.QueryResult.get_latest(qr.data_source, qr.query_text, max_age=120)
self.assertEqual(found_query_result, qr)
开发者ID:doomsplayer,项目名称:redash,代码行数:7,代码来源:test_models.py
示例4: test_get_latest_doesnt_return_if_ttl_expired
def test_get_latest_doesnt_return_if_ttl_expired(self):
yesterday = utcnow() - datetime.timedelta(days=1)
qr = self.factory.create_query_result(retrieved_at=yesterday)
found_query_result = models.QueryResult.get_latest(qr.data_source, qr.query_text, max_age=60)
self.assertIsNone(found_query_result)
开发者ID:doomsplayer,项目名称:redash,代码行数:7,代码来源:test_models.py
示例5: check_alerts_for_query
def check_alerts_for_query(self, query_id):
from redash.wsgi import app
logger.debug("Checking query %d for alerts", query_id)
query = models.Query.get_by_id(query_id)
for alert in query.alerts:
alert.query = query
new_state = alert.evaluate()
passed_rearm_threshold = False
if alert.rearm and alert.last_triggered_at:
passed_rearm_threshold = alert.last_triggered_at + datetime.timedelta(seconds=alert.rearm) < utils.utcnow()
if new_state != alert.state or (alert.state == models.Alert.TRIGGERED_STATE and passed_rearm_threshold ):
logger.info("Alert %d new state: %s", alert.id, new_state)
old_state = alert.state
alert.update_instance(state=new_state, last_triggered_at=utils.utcnow())
if old_state == models.Alert.UNKNOWN_STATE and new_state == models.Alert.OK_STATE:
logger.debug("Skipping notification (previous state was unknown and now it's ok).")
continue
host = base_url(alert.query.org)
for subscription in alert.subscriptions:
try:
subscription.notify(alert, query, subscription.user, new_state, app, host)
except Exception as e:
logger.warn("Exception: {}".format(e))
开发者ID:hudl,项目名称:redash,代码行数:27,代码来源:tasks.py
示例6: test_store_result_does_not_modify_query_update_at
def test_store_result_does_not_modify_query_update_at(self):
original_updated_at = utcnow() - datetime.timedelta(hours=1)
query = self.factory.create_query(updated_at=original_updated_at)
models.QueryResult.store_result(query.org_id, query.data_source, query.query_hash, query.query_text, "", 0, utcnow())
self.assertEqual(original_updated_at, query.updated_at)
开发者ID:getredash,项目名称:redash,代码行数:7,代码来源:test_query_results.py
示例7: test_exact_time_that_needs_reschedule
def test_exact_time_that_needs_reschedule(self):
now = utcnow()
yesterday = now - datetime.timedelta(days=1)
scheduled_datetime = now - datetime.timedelta(hours=3)
scheduled_time = "{:02d}:00".format(scheduled_datetime.hour)
self.assertTrue(models.should_schedule_next(yesterday, now,
scheduled_time, 0))
开发者ID:jonyboy2000,项目名称:redash,代码行数:7,代码来源:test_models.py
示例8: outdated_queries
def outdated_queries(cls):
queries = (
Query.query
.options(joinedload(Query.latest_query_data).load_only('retrieved_at'))
.filter(Query.schedule.isnot(None))
.order_by(Query.id)
)
now = utils.utcnow()
outdated_queries = {}
scheduled_queries_executions.refresh()
for query in queries:
if query.schedule['interval'] is None:
continue
if query.schedule['until'] is not None:
schedule_until = pytz.utc.localize(datetime.datetime.strptime(query.schedule['until'], '%Y-%m-%d'))
if schedule_until <= now:
continue
if query.latest_query_data:
retrieved_at = query.latest_query_data.retrieved_at
else:
retrieved_at = now
retrieved_at = scheduled_queries_executions.get(query.id) or retrieved_at
if should_schedule_next(retrieved_at, now, query.schedule['interval'], query.schedule['time'],
query.schedule['day_of_week'], query.schedule_failures):
key = "{}:{}".format(query.query_hash, query.data_source_id)
outdated_queries[key] = query
return outdated_queries.values()
开发者ID:ariarijp,项目名称:redash,代码行数:35,代码来源:__init__.py
示例9: check_alerts_for_query
def check_alerts_for_query(self, query_id):
from redash.wsgi import app
logger.debug("Checking query %d for alerts", query_id)
query = models.Query.get_by_id(query_id)
for alert in query.alerts:
alert.query = query
new_state = alert.evaluate()
passed_rearm_threshold = False
if alert.rearm and alert.last_triggered_at:
passed_rearm_threshold = alert.last_triggered_at + datetime.timedelta(seconds=alert.rearm) < utils.utcnow()
if new_state != alert.state or (alert.state == models.Alert.TRIGGERED_STATE and passed_rearm_threshold ):
logger.info("Alert %d new state: %s", alert.id, new_state)
old_state = alert.state
alert.update_instance(state=new_state, last_triggered_at=utils.utcnow())
if old_state == models.Alert.UNKNOWN_STATE and new_state == models.Alert.OK_STATE:
logger.debug("Skipping notification (previous state was unknown and now it's ok).")
continue
# message = Message
html = """
Check <a href="{host}/alerts/{alert_id}">alert</a> / check <a href="{host}/queries/{query_id}">query</a>.
""".format(host=base_url(alert.query.org), alert_id=alert.id, query_id=query.id)
notify_mail(alert, html, new_state, app)
if settings.HIPCHAT_API_TOKEN:
notify_hipchat(alert, html, new_state)
if settings.WEBHOOK_ENDPOINT:
notify_webhook(alert, query, html, new_state)
开发者ID:examplegithubaccount,项目名称:redash,代码行数:32,代码来源:tasks.py
示例10: test_exact_time_with_day_change
def test_exact_time_with_day_change(self):
now = utcnow().replace(hour=0, minute=1)
previous = (now - datetime.timedelta(days=2)).replace(hour=23,
minute=59)
schedule = "23:59".format(now.hour + 3)
self.assertTrue(models.should_schedule_next(previous, now, schedule,
0))
开发者ID:jonyboy2000,项目名称:redash,代码行数:7,代码来源:test_models.py
示例11: test_backoff
def test_backoff(self):
now = utcnow()
two_hours_ago = now - datetime.timedelta(hours=2)
self.assertTrue(models.should_schedule_next(two_hours_ago, now, "3600",
5))
self.assertFalse(models.should_schedule_next(two_hours_ago, now,
"3600", 10))
开发者ID:jonyboy2000,项目名称:redash,代码行数:7,代码来源:test_models.py
示例12: test_failure_extends_schedule
def test_failure_extends_schedule(self):
"""
Execution failures recorded for a query result in exponential backoff
for scheduling future execution.
"""
query = self.factory.create_query(schedule="60", schedule_failures=4)
retrieved_at = utcnow() - datetime.timedelta(minutes=16)
query_result = self.factory.create_query_result(
retrieved_at=retrieved_at, query_text=query.query_text,
query_hash=query.query_hash)
query.latest_query_data = query_result
self.assertEqual(list(models.Query.outdated_queries()), [])
query_result.retrieved_at = utcnow() - datetime.timedelta(minutes=17)
self.assertEqual(list(models.Query.outdated_queries()), [query])
开发者ID:jonyboy2000,项目名称:redash,代码行数:16,代码来源:test_models.py
示例13: test_skips_fresh_queries
def test_skips_fresh_queries(self):
half_an_hour_ago = utcnow() - datetime.timedelta(minutes=30)
query = self.factory.create_query(schedule={'interval':'3600', 'time': None, 'until':None, 'day_of_week':None})
query_result = self.factory.create_query_result(query=query.query_text, retrieved_at=half_an_hour_ago)
query.latest_query_data = query_result
queries = models.Query.outdated_queries()
self.assertNotIn(query, queries)
开发者ID:ariarijp,项目名称:redash,代码行数:8,代码来源:test_models.py
示例14: test_save_updates_updated_at_field
def test_save_updates_updated_at_field(self):
# This should be a test of ModelTimestampsMixin, but it's easier to test in context of existing model... :-\
one_day_ago = utcnow().date() - datetime.timedelta(days=1)
q = self.factory.create_query(created_at=one_day_ago, updated_at=one_day_ago)
db.session.flush()
q.name = 'x'
db.session.flush()
self.assertNotEqual(q.updated_at, one_day_ago)
开发者ID:ariarijp,项目名称:redash,代码行数:8,代码来源:test_queries.py
示例15: test_get_latest_returns_the_most_recent_result
def test_get_latest_returns_the_most_recent_result(self):
yesterday = utcnow() - datetime.timedelta(seconds=30)
old_qr = self.factory.create_query_result(retrieved_at=yesterday)
qr = self.factory.create_query_result()
found_query_result = models.QueryResult.get_latest(qr.data_source, qr.query_text, 60)
self.assertEqual(found_query_result.id, qr.id)
开发者ID:doomsplayer,项目名称:redash,代码行数:8,代码来源:test_models.py
示例16: test_exact_time_every_x_days_that_doesnt_need_reschedule
def test_exact_time_every_x_days_that_doesnt_need_reschedule(self):
now = utcnow()
four_days_ago = now - datetime.timedelta(days=2)
three_day_interval = "259200"
scheduled_datetime = now - datetime.timedelta(hours=3)
scheduled_time = "{:02d}:00".format(scheduled_datetime.hour)
self.assertFalse(models.should_schedule_next(four_days_ago, now, three_day_interval,
scheduled_time))
开发者ID:ariarijp,项目名称:redash,代码行数:8,代码来源:test_models.py
示例17: setUp
def setUp(self):
super(TestQueryResultStoreResult, self).setUp()
self.data_source = self.factory.data_source
self.query = "SELECT 1"
self.query_hash = gen_query_hash(self.query)
self.runtime = 123
self.utcnow = utcnow()
self.data = "data"
开发者ID:5t111111,项目名称:redash,代码行数:8,代码来源:test_models.py
示例18: test_outdated_queries_works_with_ttl_based_schedule
def test_outdated_queries_works_with_ttl_based_schedule(self):
two_hours_ago = utcnow() - datetime.timedelta(hours=2)
query = self.factory.create_query(schedule="3600")
query_result = self.factory.create_query_result(query=query.query_text, retrieved_at=two_hours_ago)
query.latest_query_data = query_result
queries = models.Query.outdated_queries()
self.assertIn(query, queries)
开发者ID:jonyboy2000,项目名称:redash,代码行数:8,代码来源:test_models.py
示例19: test_outdated_queries_works_with_ttl_based_schedule
def test_outdated_queries_works_with_ttl_based_schedule(self):
two_hours_ago = utcnow() - datetime.timedelta(hours=2)
query = self.factory.create_query(schedule={'interval':'3600', 'time': None, 'until':None, 'day_of_week':None})
query_result = self.factory.create_query_result(query=query.query_text, retrieved_at=two_hours_ago)
query.latest_query_data = query_result
queries = models.Query.outdated_queries()
self.assertIn(query, queries)
开发者ID:ariarijp,项目名称:redash,代码行数:8,代码来源:test_models.py
示例20: test_returns_only_over_a_week_old_results
def test_returns_only_over_a_week_old_results(self):
two_weeks_ago = utcnow() - datetime.timedelta(days=14)
unused_qr = self.factory.create_query_result(retrieved_at=two_weeks_ago)
db.session.flush()
new_unused_qr = self.factory.create_query_result()
self.assertIn((unused_qr.id,), models.QueryResult.unused())
self.assertNotIn((new_unused_qr.id,), models.QueryResult.unused())
开发者ID:jonyboy2000,项目名称:redash,代码行数:8,代码来源:test_models.py
注:本文中的redash.utils.utcnow函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论