core

Fill in a module description here

We begin with some helper functions

---

source

hash_json

 hash_json (data:Any)

Compute a SHA-256 hash of JSON-serializable data.

---

source

extend_path

 extend_path (base:str, key_or_index:Union[str,int])

Append a segment to a JSONPath using bracket notation.

---

source

path_for_key

 path_for_key (key:str)

Get a JSONPath for a given key using bracket notation.

---

source

sha256_hex

 sha256_hex (data:bytes)

Get the hex digest of the SHA-256 hash of the given data

---

source

hex_to_bytes

 hex_to_bytes (h:str)

*Convert a hex string to raw bytes.

Accepts optional ‘0x’ prefix and common separators (space, underscore, hyphen, colon). Raises TypeError for non-str inputs and ValueError for invalid hex or odd length after cleaning.*

---

source

OperationHeader

 OperationHeader (kind:str, task:str, tool:str, output_type:str,
                  event_uuid:Optional[bytes], timestamp:Optional[str],
                  meta:Any)

Metadata about an operation.

---

source

Scrapebook

 Scrapebook ()

An ABC exposing the basic API for Scrapebook

---

source

Recorder

 Recorder (book:Scrapebook, kind:str, task:str, tool:str, output_type:str)

Fixed (kind, task, tool, output_type).

---

source

OperationResult

 OperationResult (op:__main__.Operation, path:str)

Handle: (Operation, JSONPath) into that operation’s decoded results. Supports subscripting to extend the path: op[“a”][“b”][0]

---

source

Operation

 Operation (book:Scrapebook, op_id:bytes)

Lazy header; decoded results/inputs are cached along with their dependency sets.

---

source

Artifact

 Artifact (book:Scrapebook, sha256:bytes)

Lazy-loaded bytes referenced by SHA-256; caches after first load.

---

source

OperationResult.value

 OperationResult.value ()

Evaluate JSONPath on the op’s decoded cached results. - 0 matches -> None - 1 match -> the value - >1 matches-> list of values

---

source

Operation.inputs

 Operation.inputs ()

Get an operation’s inputs

---

source

Operation.results

 Operation.results ()

Get the results of an operation

class ScrapebookDict(Scrapebook):
    def __init__(self):
        self.ops = dict()
        self.arts = dict()
        self.deps = dict()
        self.op_used_by = dict()
        self.produced_by = dict()
        self.consumed_by = dict()
        pass

    # persistence
    def persist_operation(
        self,
        *,
        header: OperationHeader,
        inputs_json: Any,
        results_json: Any,
        deps: Set[bytes],
        input_arts: Set[bytes],
        output_arts: Set[bytes]
    ) -> bytes:
        op_id = header.op_id(inputs_json, results_json)
        self.ops[op_id] = { "header" : header, "input" : inputs_json, "result" : results_json}
        for dep in deps:
            self.op_used_by.setdefault(dep, set()).add(op_id)
            self.deps.setdefault(op_id, set()).add(dep)
        for art in input_arts:
            self.consumed_by.setdefault(art, set()).add(op_id)
        for art in output_arts:
            self.produced_by.setdefault(art, set()).add(op_id)
        return op_id

    def fetch_operation_header(self, op_id: bytes) -> OperationHeader:
        return self.ops[op_id]["header"]

    def fetch_results_json(self, op_id: bytes) -> Any:
        return self.ops[op_id]["result"]
    
    def fetch_inputs_json(self, op_id: bytes) -> Any:
        return self.ops[op_id]["input"]

    def fetch_op_ids(self) -> Iterable[bytes]:
        return self.ops.keys()

    def put_artifact_sha256(self, data: bytes) -> bytes:
        """Store bytes content-addressed by SHA-256 and return the hex hash."""
        hash = hashlib.sha256(data).digest()
        self.arts[hash] = data
        return hash
    
    def fetch_artifact_bytes_sha256(self, sha256: bytes) -> bytes:
        return self.arts[sha256]
    
    def fetch_artifact_produced_by(self, sha256: bytes) -> Iterable[bytes]:
        """
        Given the hash of an artifact, return the set of operation IDs that produce it.
        """
        return self.produced_by.get(sha256, set())

    def fetch_artifact_consumed_by(self, sha256: bytes) -> Iterable[bytes]:
        """
        Given the hash of an artifact, return the set of operation IDs that consume it.
        """
        return self.consumed_by.get(sha256, set())

book = ScrapebookDict()

scrapes = book.obs_recorder(task="scrape_site", tool="demo", output_type="http_response")

op = scrapes.record({"url": "http://example.com"}, {"status": 200, "content": b"hello"})
op.validate()

book.fetch_artifact_produced_by(op.id)

set()

(op, op.kind, op.task, op.tool, op.output_type, op.event_uuid.hex(), op.timestamp, op.meta)

(<Operation ae67f785a7e39b117161a13fdd2d6a5139f3e42d8bea6f9e3b264e853a45d784>,
 'obs',
 'scrape_site',
 'demo',
 'http_response',
 '0321dd21d2eabb8da0ea55383fabc053e4bccf7dc2f87c1de29404f0eee5b466',
 '2025-10-05T02:58:41.209504',
 None)

op.results_json()

{'status': 200,
 'content': {'$artifact': '2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824'}}

op.results()

{'status': 200,
 'content': <Artifact 2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824>}

op["content"].value().bytes()

b'hello'

op["content"]

<OperationResult id=ae67f785a7e39b117161a13fdd2d6a5139f3e42d8bea6f9e3b264e853a45d784 path='$["content"]'>

articles = book.trans_recorder(task="parse_article", tool="demo2", output_type="article")

op2 = articles.record({"content" : op["content"], "mode": "cool"}, "good article")

(
    op2, op2.kind, op2.task, op2.tool, op2.output_type, op2.event_uuid, 
    op2.timestamp, op2.meta
)

(<Operation d37127eba9c227c45f73d26c34d4305f90ceff4a0f253577e809a32bbc243cc8>,
 'trans',
 'parse_article',
 'demo2',
 'article',
 None,
 None,
 None)

op2.value()

'good article'

op2.inputs()

{'content': <OperationResult id=ae67f785a7e39b117161a13fdd2d6a5139f3e42d8bea6f9e3b264e853a45d784 path='$["content"]'>,
 'mode': 'cool'}

[dep for dep in op2.deps()]

[<Operation ae67f785a7e39b117161a13fdd2d6a5139f3e42d8bea6f9e3b264e853a45d784>]

op3 = articles.record(
    {"content" : op["content"], "mode": op2, "status": op["status"]}, 
    "better article"
)

op3

<Operation f60c4785522a91181c5110336408465d2591b9f34a59b231ec2e5deea6b48377>

op3.value()

'better article'

[dep for dep in op3.deps()]

[<Operation ae67f785a7e39b117161a13fdd2d6a5139f3e42d8bea6f9e3b264e853a45d784>,
 <Operation d37127eba9c227c45f73d26c34d4305f90ceff4a0f253577e809a32bbc243cc8>,
 <Operation ae67f785a7e39b117161a13fdd2d6a5139f3e42d8bea6f9e3b264e853a45d784>]

op

<Operation ae67f785a7e39b117161a13fdd2d6a5139f3e42d8bea6f9e3b264e853a45d784>

op.validate()

op.artifacts_produced()

{<Artifact 2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824>}

book.validate()

3