JSON Folder Sync#

This example syncs JSON content between a local folder and a LabArchives page. Use it when you want a simple batch workflow for moving structured JSON files in or out of a notebook page.

When to Use It#

This is useful for:

  • Backing up structured data from LabArchives to your local machine.

  • Uploading batches of JSON data files to a LabArchives page.

  • Syncing experimental data stored as JSON between local files and LabArchives.

  • Archiving API responses or other structured datasets.

Requirements#

This example assumes the recommended local interactive profile, labapi[dotenv,builtin-auth]. See Installation.

No additional third-party packages are required.

Configuration#

For the local interactive workflow, create a .env file in the repository root:

API_URL="https://api.labarchives.com"
ACCESS_KEYID="your_access_key_id"
ACCESS_PWD="your_password"

You can also provide the same values through shell environment variables. See Your First Entry for both options.

Common Commands#

Upload the sample JSON files included in the repository:

uv run python examples/json_sync/json_sync.py upload examples/json_sync/sample_data "Experiments/2024/Data Analysis" --notebook "My Notebook"

Download JSON entries from a page into a local folder:

uv run python examples/json_sync/json_sync.py download "Experiments/2024/Data Analysis" ./output --notebook "My Notebook"

How It Works#

  • JSON files are uploaded with create_json_entry().

  • Each upload creates a JSON attachment with application/json MIME type and a companion rich-text preview entry.

  • Download mode writes each JSON attachment back to a local .json file.

Notes and Limitations#

  • Invalid JSON files are skipped with an error message.

  • The script creates the output folder if it does not exist during download.

  • The sample upload command uses the checked-in files under examples/json_sync/sample_data.

Ways to Extend It#

  1. Implement diff-based sync so only changed files are uploaded.

  2. Recurse through subdirectories instead of processing a single folder.

  3. Add filename filtering for larger datasets.

  4. Show progress bars with tqdm.

  5. Add retry logic for failed uploads and downloads.

Source Code#

#!/usr/bin/env python3
"""Synchronize JSON files between local disk and a LabArchives page."""

import argparse
import json
import sys
from pathlib import Path

from labapi import (
    AttachmentEntry,
    Client,
    InsertBehavior,
    AbstractTreeContainer,
    NotebookPage,
    TraversalError,
    User,
)


def get_or_create_page(container: AbstractTreeContainer, path: str) -> NotebookPage:
    """Return an existing page at ``path`` or create it with missing parents."""
    try:
        node = container.traverse(path)
    except TraversalError as err:
        if err.available_children is None:
            raise
        return container.create(
            NotebookPage,
            path,
            parents=True,
            if_exists=InsertBehavior.Retain,
        )

    if node.is_dir():
        raise TypeError(f"'{path}' refers to a directory, but a page is required")

    return node.as_page()


def upload_json_folder(
    user: User, notebook_name: str, page_path: str, local_folder: Path
) -> None:
    """Upload all JSON files from a local folder to a LabArchives page."""
    if not local_folder.exists():
        print(f"Error: Local folder '{local_folder}' does not exist")
        sys.exit(1)

    if not local_folder.is_dir():
        print(f"Error: '{local_folder}' is not a directory")
        sys.exit(1)

    # Get the target page
    notebooks = user.notebooks
    try:
        notebook = notebooks[notebook_name]
        print(f"Ensuring page path exists: {page_path}")
        page = get_or_create_page(notebook, page_path)
    except (KeyError, TraversalError, TypeError, ValueError) as e:
        print(f"Error: Could not access or create path '{page_path}': {e}")
        sys.exit(1)

    # Find all JSON files
    json_files = sorted(local_folder.glob("*.json"))

    if not json_files:
        print(f"No JSON files found in '{local_folder}'")
        return

    print(f"Found {len(json_files)} JSON file(s) to upload")

    # Upload each JSON file
    for json_file in json_files:
        print(f"Uploading {json_file.name}...", end=" ")

        try:
            with json_file.open("r", encoding="utf-8") as f:
                data = json.load(f)

            # Create JSON entry
            page.entries.create_json_entry(data)

            print("✓")
        except json.JSONDecodeError as e:
            print(f"✗ (Invalid JSON: {e})")
        except Exception as e:
            print(f"✗ (Error: {e})")

    print(f"\nUpload complete! {len(json_files)} files processed.")


def download_json_entries(
    user: User, notebook_name: str, page_path: str, local_folder: Path
) -> None:
    """Download all JSON entries from a LabArchives page to local files."""
    # Create output folder if it doesn't exist
    local_folder.mkdir(parents=True, exist_ok=True)

    # Get the source page
    notebooks = user.notebooks
    try:
        notebook = notebooks[notebook_name]
    except KeyError as e:
        print(f"Error: Could not find notebook '{notebook_name}': {e}")
        print(f"Available notebooks: {list(notebooks.keys())}")
        sys.exit(1)
    try:
        page = notebook.traverse(page_path).as_page()
    except TraversalError as e:
        print(
            f"Error: Could not find page '{page_path}' in notebook '{notebook_name}': {e}"
        )
        sys.exit(1)
    except TypeError:
        print(f"Error: '{page_path}' refers to a directory, but a page is required")
        sys.exit(1)

    # Find all JSON entries
    entries = page.entries
    json_entries: list[AttachmentEntry] = []

    for entry in entries:
        # JSON entries are stored as attachments. We check both MIME type and filename.
        if isinstance(entry, AttachmentEntry):
            attachment = entry.get_attachment()
            is_json_mime = attachment.mime_type == "application/json"
            is_json_ext = attachment.filename.lower().endswith(".json")

            if is_json_mime or is_json_ext:
                json_entries.append(entry)

    if not json_entries:
        print(f"No JSON entries found on page '{page_path}'")
        return

    print(f"Found {len(json_entries)} JSON entry/entries to download")

    # Download each JSON entry
    for entry in json_entries:
        attachment = entry.get_attachment()
        filename = attachment.filename
        output_path = local_folder / filename

        print(f"Downloading {filename}...", end=" ")

        try:
            # Read JSON data from attachment
            attachment.seek(0)
            data = json.load(attachment)

            # Write to local file
            with output_path.open("w", encoding="utf-8") as f:
                json.dump(data, f, indent=2)

            print("✓")
        except Exception as e:
            print(f"✗ (Error: {e})")

    print(f"\nDownload complete! {len(json_entries)} file(s) saved to '{local_folder}'")


def main() -> None:
    """Run the JSON sync example CLI."""
    parser = argparse.ArgumentParser(
        description="Sync JSON files between local folder and LabArchives page"
    )
    parser.add_argument(
        "action",
        choices=["upload", "download"],
        help="Action to perform: upload to LabArchives or download from LabArchives",
    )
    parser.add_argument(
        "source",
        help="Source: local folder path (upload) or LabArchives page path (download)",
    )
    parser.add_argument(
        "destination",
        help="Destination: LabArchives page path (upload) or local folder path (download)",
    )
    parser.add_argument(
        "--notebook",
        "-n",
        required=True,
        help="Name of the LabArchives notebook to use",
    )

    args = parser.parse_args()

    print("Connecting to LabArchives...")
    try:
        with Client() as client:
            print("Authenticating...")
            user = client.default_authenticate()
            print("✓ Authenticated successfully")

            if args.action == "upload":
                local_folder = Path(args.source)
                page_path = args.destination
                upload_json_folder(user, args.notebook, page_path, local_folder)
            else:  # download
                page_path = args.source
                local_folder = Path(args.destination)
                download_json_entries(user, args.notebook, page_path, local_folder)
    except Exception as e:
        print(f"Authentication error: {e}")
        print("\nMake sure you have a .env file with your credentials:")
        print("  ACCESS_KEYID=your_access_key_id")
        print("  ACCESS_PWD=your_password")
        sys.exit(1)


if __name__ == "__main__":
    main()