Skip to content

Instantly share code, notes, and snippets.

@jamalex
Created January 17, 2019 07:14
Show Gist options
  • Save jamalex/2a17b359f336e1c6a67e24c417ac1711 to your computer and use it in GitHub Desktop.
Save jamalex/2a17b359f336e1c6a67e24c417ac1711 to your computer and use it in GitHub Desktop.
Notion task manager toy example
import datetime
import random
import time
from multiprocessing import Lock
from notion.client import *
from notion.block import *
mutex = Lock()
token_v2 = "<redacted>"
client = NotionClient(token_v2=token_v2, start_monitoring=False)
page = client.get_block("https://www.notion.so/learningequality/Task-Manager-8a948144e0d04a25b30f1846d730f9d9")
log = client.get_block("https://www.notion.so/learningequality/Task-Manager-8a948144e0d04a25b30f1846d730f9d9#a8247c1908ab4292b3de1024f2403ef7")
collection = client.get_block("https://www.notion.so/learningequality/fe3c3a8c6aa24e37a3931b4708c45d71?v=4c79d24b0a4d4155ac52fe964d3d759c").collection
log_lines = []
happy_results = {
"Delete all data": "The data was all deleted.",
"Fold paper airplane": "Your airplane is the BEST!",
"Feed the penguins": "Penguins are suffoncified.",
}
sad_results = {
"Delete all data": ["Insufficient permissions.", "Computer says no.", "I'm sorry, Dave..."],
"Fold paper airplane": ["Spontaneous combustion.", "OWWWW PAPERCUT!!!", "Oops, that was your diploma."],
"Feed the penguins": ["Penguins have a tummyache. :(", "We ran out of food.", "Polar bear invasion!"],
}
def write_to_log(line):
global log_lines
line = "{}\t{}".format(datetime.datetime.now().strftime("%H:%M:%S"), line)
log_lines.append(line)
log_lines = log_lines[-10:]
log.title = "\n".join(log_lines)
def sleep_with_updates(record, start_time, sleep=0.001, interval=0.3):
while sleep > 0:
record.elapsed = int(time.time() - start_time)
time.sleep(sleep if sleep < interval else interval)
sleep -= interval
def run_job(record):
start_time = time.time()
record.result = ""
record.elapsed = 0
record.status = "Starting"
write_to_log("Starting task '{}'...".format(record.task_name))
sleep_with_updates(record, start_time, random.randint(1, 3))
record.status = "Active"
write_to_log("Task '{}' is now running!".format(record.task_name))
sleep_with_updates(record, start_time, random.randint(3, 6))
diceroll = random.randint(0, 6)
sadness = sad_results[record.task_name]
if diceroll >= len(sadness):
record.result = happy_results[record.task_name]
record.status = "Completed"
write_to_log("Completed '{}'. {}".format(record.task_name, record.result))
else:
record.result = sadness[diceroll]
record.status = "Error"
write_to_log("Error running '{}': {}".format(record.task_name, record.result))
def row_callback(record, changes):
with mutex:
start = record.start
record.start = False
if start:
if record.status in [None, "Completed", "Error"]:
run_job(record)
def register_row_callbacks(collection):
for row in collection.get_rows():
row.add_callback(row_callback, callback_id="row_callback")
def collection_callback(record, difference, changes):
register_row_callbacks(record)
collection.add_callback(collection_callback)
register_row_callbacks(collection)
client.start_monitoring()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment