Download notebook · View source · Open in Colab

Ingest Cookbook (CSVParser)

CSV parsers translate rows into commands and blobs for the parallel loader. This example ingests two dishes and their ingredients from the Cookbook dataset. The resulting graph links each dish image to its ingredient entities.

The examples require a configured ApertureDB connection; see client configuration. The setup includes the sample metadata and prepares local input files. It reads Cookbook images from a repository checkout when available, or downloads them from a fixed commit. SHA-256 checks identify the expected image bytes.

Unique record identifiers and a doc_run property keep queries and cleanup within this run. Complete cleanup before restarting.

The repository's create_nested_json.py and convert_ingredients_adb_csv.py scripts support conversion of the full dataset from the live Google Sheet. This example uses the included sample instead.

# Install the required packages
%pip install --quiet "aperturedb==0.5.1"

Note: you may need to restart the kernel to use updated packages.

Prepare Input Data

from pathlib import Path
from types import SimpleNamespace
import csv
import hashlib
import json
import tempfile
import urllib.request
import uuid

# A fixed two-dish metadata sample. Images are pinned to a repository commit
# and verified byte-for-byte, whether read locally or downloaded.
COOKBOOK_REVISION = "51b3c3d9f495f8fd77108a270b88fe8ab8cc0004"
sample = [{'dish_id': 1,
  'image': 'images/001_Large.jpeg',
  'image_sha256': 'b8e9d9ad0dccae2b5218d329d8c30d1b085c2d77431893a434a46d3c1db9eb55',
  'contributor': 'gautam',
  'name': 'rajma chawal',
  'location': 'NJ',
  'cuisine': 'Indian',
  'caption': 'Beans with rice',
  'recipe_url': 'https://www.tarladalal.com/rajma-chawal-punjabi-rajma-chawal-4951r',
  'ingredients': [{'Name': 'rajma'}, {'Name': 'rice'}]},
 {'dish_id': 4,
  'image': 'images/004_Large.jpeg',
  'image_sha256': 'a3b41c925f1d51003ef7dd924999eed0d04b31752a9dacf47065df1de3458a1a',
  'contributor': 'gautam',
  'name': 'Butter chicken',
  'location': 'NJ',
  'cuisine': 'Indian',
  'caption': 'Chicken in Creamy tomato based gravy',
  'recipe_url': 'https://www.indianhealthyrecipes.com/butter-chicken/',
  'ingredients': [{'Name': 'chicken'}, {'Name': 'butter'}, {'Name': 'cream'}]}]

run_id = "Cookbook_" + uuid.uuid4().hex[:12]
temporary = tempfile.TemporaryDirectory(prefix="cookbook-ingestion-")
fixture = SimpleNamespace(
    run_id=run_id,
    ingredient_class="DocumentationIngredient_" + run_id,
    connection_class="DocumentationHasIngredient_" + run_id,
    dishes=sample, directory=Path(temporary.name), close=temporary.cleanup
)
repository = next((p for p in [Path.cwd(), *Path.cwd().parents]
                   if (p / "images/001_Large.jpeg").is_file()), None)
for dish in fixture.dishes:
    relative = dish.pop("image")
    expected_hash = dish.pop("image_sha256")
    if repository is not None:
        image_bytes = (repository / relative).read_bytes()
    else:
        url = f"https://raw.githubusercontent.com/aperture-data/Cookbook/{COOKBOOK_REVISION}/{relative}"
        with urllib.request.urlopen(url, timeout=30) as source:
            image_bytes = source.read()
    assert hashlib.sha256(image_bytes).hexdigest() == expected_hash, relative
    image_path = fixture.directory / Path(relative).name
    image_path.write_bytes(image_bytes)
    dish.update(url=str(image_path), id=f"{run_id}_dish_{dish['dish_id']}", doc_run=run_id)
    for index, ingredient in enumerate(dish["ingredients"]):
        ingredient.update(id=f"{dish['id']}_ingredient_{index}", doc_run=run_id)
(fixture.directory / "dishes.json").write_text(json.dumps(fixture.dishes, indent=2))

# Prepare equivalent CSV inputs. Only the CSVParser lesson consumes them.
def write_csv(name, fields, rows):
    with (fixture.directory / name).open("w", newline="") as output:
        writer = csv.DictWriter(output, fieldnames=fields)
        writer.writeheader()
        writer.writerows(rows)

image_rows = [{**{key: dish[key] for key in
                ("id", "contributor", "name", "location", "cuisine", "caption", "recipe_url", "dish_id", "doc_run")},
               "filename": dish["url"], "constraint_id": dish["id"]} for dish in fixture.dishes]
write_csv("dishes.adb.csv", ["filename", "id", "contributor", "name", "location", "cuisine", "caption", "recipe_url", "dish_id", "doc_run", "constraint_id"], image_rows)
ingredient_rows = [{"EntityClass": fixture.ingredient_class, **ingredient, "constraint_id": ingredient["id"]}
                   for dish in fixture.dishes for ingredient in dish["ingredients"]]
write_csv("ingredients.adb.csv", ["EntityClass", "id", "Name", "doc_run", "constraint_id"], ingredient_rows)
destination = fixture.ingredient_class + "@id"
connections = [{"ConnectionClass": fixture.connection_class, "_Image@id": dish["id"], destination: ingredient["id"],
                "id": ingredient["id"], "doc_run": run_id, "constraint_id": ingredient["id"]}
               for dish in fixture.dishes for ingredient in dish["ingredients"]]
write_csv("dish_ingredients.adb.csv", ["ConnectionClass", "_Image@id", destination, "id", "doc_run", "constraint_id"], connections)
print(f"Prepared {len(fixture.dishes)} dishes and {len(ingredient_rows)} ingredient records.")

Prepared 2 dishes and 5 ingredient records.

CSV input conventions

The prepared files use these headers:

• dishes.adb.csv: filename is a local image path; following columns are image properties. constraint_id makes repeated insertion conditional on the unique image ID.
• ingredients.adb.csv: EntityClass chooses this run's ingredient class; constraint_id tests the unique ingredient ID.
• dish_ingredients.adb.csv: ConnectionClass, _Image@id, and the generated ingredient class followed by @id identify each connection's two endpoints. constraint_id avoids duplicate connections for the same input record.

The CSV headers use property names such as recipe_url instead of the source spreadsheet's Recipe URL header. Unique IDs keep endpoint lookups within this run even when another dataset uses the same dish names.

Ingesting using the CSV Parsers.

The CSV parsers translate each row into ApertureDB commands and input blobs. The loader ingests the dish images and ingredient entities before adding connections between them.

Parser references:

• ImageDataCSV
• EntityDataCSV
• ConnectionDataCSV

from aperturedb.ImageDataCSV import ImageDataCSV
from aperturedb.EntityDataCSV import EntityDataCSV
from aperturedb.ConnectionDataCSV import ConnectionDataCSV
from aperturedb.CommonLibrary import create_connector, execute_query
from tqdm.auto import tqdm

loaders = [
    ImageDataCSV(str(fixture.directory / "dishes.adb.csv")),
    EntityDataCSV(str(fixture.directory / "ingredients.adb.csv")),
    ConnectionDataCSV(str(fixture.directory / "dish_ingredients.adb.csv"))
]
client = create_connector()
ingredient_class = fixture.ingredient_class
connection_class = fixture.connection_class
with tqdm(total=sum(len(loader) for loader in loaders)) as progress:
    for loader in loaders:  # Load endpoints before their connections.
        for query, blobs in loader:
            result, response, output_blobs = execute_query(client, query, blobs)
            assert result == 0, (response, query)
            progress.update(1)

  0%|          | 0/12 [00:00<?, ?it/s]

Load CSVs with the CLI

The same files can be passed to adb ingest from-csv using --ingest-type IMAGE, ENTITY, and CONNECTION, respectively. Use the paths in fixture.directory and load image/entity endpoints before connections. This notebook executes the SDK loop above; the CLI is an alternative, not an additional step needed here.

Verify the persisted graph

The counts show how the two dishes, their ingredients, and the connections between them were stored.

scope = ["$doc_run", "==", fixture.run_id]
query = [
    {"FindImage": {"_ref": 1, "constraints": scope, "blobs": False, "results": {"count": True}}},
    {"FindEntity": {"with_class": ingredient_class, "constraints": scope, "results": {"count": True}}},
    {"FindConnection": {"src": 1, "with_class": connection_class, "results": {"count": True}}}
]
response, _ = client.query(query)
assert isinstance(response, list) and all(next(iter(r.values()))["status"] == 0 for r in response), response
expected = [len(fixture.dishes), sum(len(d["ingredients"]) for d in fixture.dishes)]
assert [response[0]["FindImage"]["count"], response[1]["FindEntity"]["count"]] == expected, response
assert response[2]["FindConnection"]["count"] == expected[1], response
print(f"Verified {expected[0]} images, {expected[1]} ingredients, and {expected[1]} connections.")

Verified 2 images, 5 ingredients, and 5 connections.

Cleanup

Remove this run's objects and temporary input files, then verify zero matches.

# Delete only objects tagged by this run. Incident connections are removed too.
response, _ = client.query([
    {"DeleteImage": {"constraints": scope}},
    {"DeleteEntity": {"with_class": ingredient_class, "constraints": scope}}
])
assert isinstance(response, list) and all(next(iter(r.values()))["status"] == 0 for r in response), response
response, _ = client.query(query)
assert isinstance(response, list) and all(next(iter(r.values()))["status"] == 0 for r in response), response
assert all(next(iter(r.values()))["count"] == 0 for r in response), response
fixture.close()
client.close()
print("Verified cleanup: no fixture images, ingredients, or connections remain.")

Verified cleanup: no fixture images, ingredients, or connections remain.