Examples¶

The book.mark as a decorator¶

import requests
from jotting import book


@book.mark
def get_url(url):
    r = requests.get(url)
    book.write(debug="checking status...")
    r.raise_for_status()
    return r


response = get_url("https://google.com")

Using book as a context manager¶

import requests
from jotting import book

urls = ("https://google.com", "not-here")

responses = []
for u in urls:
    with book("getting %s" % u):
        r = requests.get(u)
        r.raise_for_status()
        responses.append(r)

Changing where you jot down your logs¶

import requests
from jotting import book, to, style

to_print = to.Print(style.Log())
to_file = to.File(path="~/Desktop/logbox.txt")
book.distribute(to_print, to_file)


# we can format the title with
# the inputs of the function
@book.mark("getting {url}")
def get_url(url):
    r = requests.get(url)
    r.raise_for_status()
    return r


response = get_url("https://google.com")

A toy example showing how to link logs across the web with flask¶

from flask import Flask, jsonify, request
from jotting import book
import json

# Server
# ------

app = Flask(__name__)


@app.route("/api/task", methods=["PUT"])
def task():
    data = json.loads(request.data)
    with book('api', data["parent"]):
        book.conclude(status=200)
        return jsonify({"status": 200})


# Client
# ------

with book('put') as b:
    route = '/api/task'
    data = json.dumps({'parent': b.tag})
    app.test_client().put(route, data=data)

Link logs between asynchronous threads¶

import sys
import time
if sys.version_info < (3, 0):
    from Queue import Queue
else:
    from queue import Queue
import threading, requests
from jotting import book, to, read

logbox = "~/Desktop/logbox.txt"
book.distribute(to.File(path=logbox))


def get(queue, url, parent):
    """Get the URL and queue the response."""
    with book("get({url})", parent, url=url):
        try:
            response = requests.get(url)
        except Exception as e:
            queue.put(e)
        else:
            queue.put(response.status_code)


@book.mark
def schedule(function, *args):
    """Create a thread for each mapping of the function to args."""
    q = Queue()
    for x in args:
        # we want to resume the current book
        inputs = (q, x, book.current("tag"))
        threading.Thread(target=function, args=inputs).start()
        book.write(scheduled=x)
    return [q.get(timeout=5) for i in range(len(args))]


urls = ["https://google.com", "https://wikipedia.org"]
responses = schedule(get, *urls)

time.sleep(0.1) # give time for logs to flush

read.Complete(logbox)

Link logs between asynchronous processes¶

import sys
import time
import threading, requests
from jotting import book, to, read
from multiprocessing import Process, Queue

logbox = "~/Desktop/logbox.txt"
book.distribute(to.File(path=logbox))


def get(queue, url, parent):
    """Get the URL and queue the response."""
    book.distribute(to.File(path=logbox))
    with book("get({url})", parent, url=url):
        try:
            response = requests.get(url)
        except Exception as e:
            queue.put(e)
        else:
            queue.put(response.status_code)


@book.mark
def schedule(function, *args):
    """Create a process for each mapping of the function to args."""
    q = Queue()
    for x in args:
        # we want to resume the current book
        inputs = (q, x, book.current("tag"))
        Process(target=function, args=inputs).start()
        book.write(scheduled=x)
    return [q.get(timeout=5) for i in range(len(args))]


urls = ["https://google.com", "https://wikipedia.org"]
responses = schedule(get, *urls)

time.sleep(0.1) # give time for logs to flush

read.Complete(logbox)