# core
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*
``` python
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())
```
``` python
book = ScrapebookDict()
```
``` python
scrapes = book.obs_recorder(task="scrape_site", tool="demo", output_type="http_response")
```
``` python
op = scrapes.record({"url": "http://example.com"}, {"status": 200, "content": b"hello"})
op.validate()
```
``` python
book.fetch_artifact_produced_by(op.id)
```
set()
``` python
(op, op.kind, op.task, op.tool, op.output_type, op.event_uuid.hex(), op.timestamp, op.meta)
```
(,
'obs',
'scrape_site',
'demo',
'http_response',
'0321dd21d2eabb8da0ea55383fabc053e4bccf7dc2f87c1de29404f0eee5b466',
'2025-10-05T02:58:41.209504',
None)
``` python
op.results_json()
```
{'status': 200,
'content': {'$artifact': '2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824'}}
``` python
op.results()
```
{'status': 200,
'content': }
``` python
op["content"].value().bytes()
```
b'hello'
``` python
op["content"]
```
``` python
articles = book.trans_recorder(task="parse_article", tool="demo2", output_type="article")
```
``` python
op2 = articles.record({"content" : op["content"], "mode": "cool"}, "good article")
```
``` python
(
op2, op2.kind, op2.task, op2.tool, op2.output_type, op2.event_uuid,
op2.timestamp, op2.meta
)
```
(,
'trans',
'parse_article',
'demo2',
'article',
None,
None,
None)
``` python
op2.value()
```
'good article'
``` python
op2.inputs()
```
{'content': ,
'mode': 'cool'}
``` python
[dep for dep in op2.deps()]
```
[]
``` python
op3 = articles.record(
{"content" : op["content"], "mode": op2, "status": op["status"]},
"better article"
)
```
``` python
op3
```
``` python
op3.value()
```
'better article'
``` python
[dep for dep in op3.deps()]
```
[,
,
]
``` python
op
```
``` python
op.validate()
```
``` python
op.artifacts_produced()
```
{}
``` python
book.validate()
```
3