Files

tech 1ae84e994a initial: project docs and design spec

2026-04-21 14:51:56 +08:00

26 KiB

Raw Blame History

Order Backup CLI — Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Build a CLI tool that scans all orders from Azure Table, backs them up to per-date CSVs by time range, and deletes backed-up records from production.

Architecture: Three-phase pipeline — scan (Azure → local CSV), backup (local CSV → filtered per-date CSVs), delete (CSV → Azure batch delete). Each phase is a separate CLI command. Config via TOML file.

Tech Stack: Python 3.11+, uv, azure-data-tables, click (CLI), tomllib (stdlib)

File Structure

order-bak/
├── pyproject.toml
├── src/
│   └── order_bak/
│       ├── __init__.py
│       ├── cli.py          # Click CLI entry point (3 commands)
│       ├── config.py       # Load ~/.config/order-bak/config.toml
│       ├── csv_utils.py    # CSV column defs, write/read helpers
│       ├── scan.py         # Scan command logic
│       ├── backup.py       # Backup command logic
│       └── delete.py       # Delete command logic
└── tests/
    ├── __init__.py
    ├── test_config.py
    ├── test_csv_utils.py
    ├── test_backup.py
    └── test_delete.py

Task 1: Project Scaffold

Files:

Create: pyproject.toml
Create: src/order_bak/__init__.py
Create: tests/__init__.py
Step 1: Create pyproject.toml

[project]
name = "order-bak"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = [
    "azure-data-tables>=12.5",
    "click>=8.1",
]

[project.scripts]
order-bak = "order_bak.cli:main"

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.build.targets.wheel]
packages = ["src/order_bak"]

[dependency-groups]
dev = [
    "pytest>=8.0",
]

Step 2: Create package init and test init

src/order_bak/__init__.py — empty file.

tests/__init__.py — empty file.

Step 3: Install dependencies

Run: uv sync --group dev Expected: dependencies installed successfully

Step 4: Verify entry point works

src/order_bak/cli.py:

import click


@click.group()
def main():
    pass


if __name__ == "__main__":
    main()

Run: uv run order-bak Expected: no error, shows help prompt

Step 5: Commit

git add pyproject.toml src/ tests/
git commit -m "feat: project scaffold with uv, click, azure-data-tables"

Task 2: Config Module

Files:

Create: src/order_bak/config.py
Create: tests/test_config.py
Step 1: Write failing test

# tests/test_config.py
import tomllib
from pathlib import Path
from unittest.mock import patch

from order_bak.config import load_config


def test_load_config_defaults(tmp_path):
    config_path = tmp_path / "config.toml"
    config_path.write_text(tomllib.dumps({
        "azure": {"connection_string": "conn_str_val"},
    }))
    cfg = load_config(config_path)
    assert cfg.azure_connection_string == "conn_str_val"
    assert cfg.scan_page_size == 500
    assert cfg.scan_delay_ms == 200
    assert cfg.scan_dir == Path("./scan")
    assert cfg.delete_batch_size == 100
    assert cfg.delete_delay_ms == 300


def test_load_config_custom_values(tmp_path):
    config_path = tmp_path / "config.toml"
    config_path.write_text(tomllib.dumps({
        "azure": {"connection_string": "conn_str_val"},
        "scan": {"page_size": 1000, "delay_ms": 500, "scan_dir": "/data/scan"},
        "delete": {"batch_size": 50, "delay_ms": 100},
    }))
    cfg = load_config(config_path)
    assert cfg.scan_page_size == 1000
    assert cfg.scan_delay_ms == 500
    assert cfg.scan_dir == Path("/data/scan")
    assert cfg.delete_batch_size == 50
    assert cfg.delete_delay_ms == 100


def test_load_config_missing_azure_raises(tmp_path):
    config_path = tmp_path / "config.toml"
    config_path.write_text("[scan]\npage_size = 100\n")
    try:
        load_config(config_path)
        assert False, "should have raised"
    except SystemExit:
        pass

Step 2: Run test to verify it fails

Run: uv run pytest tests/test_config.py -v Expected: FAIL — ModuleNotFoundError: No module named 'order_bak.config'

Step 3: Implement config

# src/order_bak/config.py
import sys
import tomllib
from dataclasses import dataclass
from pathlib import Path


@dataclass
class Config:
    azure_connection_string: str
    scan_page_size: int = 500
    scan_delay_ms: int = 200
    scan_dir: Path = Path("./scan")
    delete_batch_size: int = 100
    delete_delay_ms: int = 300


def load_config(config_path: Path) -> Config:
    if not config_path.exists():
        print(f"Config file not found: {config_path}")
        sys.exit(1)

    with open(config_path, "rb") as f:
        raw = tomllib.load(f)

    azure = raw.get("azure", {})
    connection_string = azure.get("connection_string")
    if not connection_string:
        print("Missing azure.connection_string in config")
        sys.exit(1)

    scan = raw.get("scan", {})
    delete = raw.get("delete", {})

    return Config(
        azure_connection_string=connection_string,
        scan_page_size=scan.get("page_size", 500),
        scan_delay_ms=scan.get("delay_ms", 200),
        scan_dir=Path(scan.get("scan_dir", "./scan")),
        delete_batch_size=delete.get("batch_size", 100),
        delete_delay_ms=delete.get("delay_ms", 300),
    )

Step 4: Run test to verify it passes

Run: uv run pytest tests/test_config.py -v Expected: 3 passed

Step 5: Commit

git add src/order_bak/config.py tests/test_config.py
git commit -m "feat: config module with TOML loading"

Task 3: CSV Utilities

Files:

Create: src/order_bak/csv_utils.py
Create: tests/test_csv_utils.py
Step 1: Write failing test

# tests/test_csv_utils.py
import csv
from datetime import datetime, timezone
from pathlib import Path

from order_bak.csv_utils import (
    CSV_COLUMNS,
    entity_to_row,
    row_to_entity,
    write_csv,
    read_csv,
)


def test_csv_columns_defined():
    assert "PartitionKey" in CSV_COLUMNS
    assert "RowKey" in CSV_COLUMNS
    assert "OrderTime" in CSV_COLUMNS
    assert "state" in CSV_COLUMNS
    assert len(CSV_COLUMNS) == 13


def test_entity_to_row_converts_datetime():
    entity = {
        "PartitionKey": "user1",
        "RowKey": "order1",
        "OrderTime": datetime(2025, 6, 15, 10, 30, 0, tzinfo=timezone.utc),
        "ReceiveTime": datetime(2025, 6, 15, 11, 0, 0, tzinfo=timezone.utc),
        "revenue": 99.5,
        "platformOrder": "plat123",
        "prodCount": 2,
        "pfId": "user1",
        "prodId": "prod1",
        "prodPrice": 49.75,
        "state": 1,
        "ver": 3,
        "customData": '{"key":"val"}',
    }
    row = entity_to_row(entity)
    assert row["OrderTime"] == "2025-06-15T10:30:00+00:00"
    assert row["ReceiveTime"] == "2025-06-15T11:00:00+00:00"
    assert row["revenue"] == "99.5"
    assert row["state"] == "1"


def test_entity_to_row_handles_none():
    entity = {
        "PartitionKey": "user1",
        "RowKey": "order1",
        "OrderTime": datetime(2025, 1, 1, tzinfo=timezone.utc),
        "ReceiveTime": None,
        "revenue": None,
        "platformOrder": None,
        "prodCount": None,
        "pfId": None,
        "prodId": None,
        "prodPrice": None,
        "state": 0,
        "ver": None,
        "customData": None,
    }
    row = entity_to_row(entity)
    assert row["ReceiveTime"] == ""
    assert row["revenue"] == ""


def test_row_to_entity_roundtrip(tmp_path):
    original = {
        "PartitionKey": "user1",
        "RowKey": "order1",
        "OrderTime": "2025-06-15T10:30:00+00:00",
        "ReceiveTime": "",
        "revenue": "99.5",
        "platformOrder": "plat123",
        "prodCount": "2",
        "pfId": "user1",
        "prodId": "prod1",
        "prodPrice": "49.75",
        "state": "1",
        "ver": "3",
        "customData": '{"key":"val"}',
    }
    entity = row_to_entity(original)
    assert entity["PartitionKey"] == "user1"
    assert entity["state"] == 1
    assert entity["revenue"] == 99.5
    assert entity["prodCount"] == 2
    assert entity["ReceiveTime"] is None


def test_write_and_read_csv(tmp_path):
    entities = [
        {
            "PartitionKey": "user1",
            "RowKey": "order1",
            "OrderTime": datetime(2025, 6, 15, 10, 30, tzinfo=timezone.utc),
            "ReceiveTime": None,
            "revenue": 99.5,
            "platformOrder": "p1",
            "prodCount": 2,
            "pfId": "user1",
            "prodId": "prod1",
            "prodPrice": 49.75,
            "state": 1,
            "ver": 3,
            "customData": None,
        },
        {
            "PartitionKey": "user2",
            "RowKey": "order2",
            "OrderTime": datetime(2025, 6, 16, 12, 0, tzinfo=timezone.utc),
            "ReceiveTime": datetime(2025, 6, 16, 13, 0, tzinfo=timezone.utc),
            "revenue": 10.0,
            "platformOrder": "p2",
            "prodCount": 1,
            "pfId": "user2",
            "prodId": "prod2",
            "prodPrice": 10.0,
            "state": 1,
            "ver": 3,
            "customData": '{"note":"test"}',
        },
    ]
    csv_path = tmp_path / "test.csv"
    write_csv(csv_path, entities)

    result = list(read_csv(csv_path))
    assert len(result) == 2
    assert result[0]["PartitionKey"] == "user1"
    assert result[1]["PartitionKey"] == "user2"
    assert result[1]["revenue"] == 10.0
    assert result[0]["ReceiveTime"] is None
    assert result[1]["ReceiveTime"] is not None

Step 2: Run test to verify it fails

Run: uv run pytest tests/test_csv_utils.py -v Expected: FAIL — ModuleNotFoundError: No module named 'order_bak.csv_utils'

Step 3: Implement csv_utils

# src/order_bak/csv_utils.py
import csv
from datetime import datetime
from pathlib import Path
from typing import Any

CSV_COLUMNS = [
    "PartitionKey",
    "RowKey",
    "OrderTime",
    "ReceiveTime",
    "revenue",
    "platformOrder",
    "prodCount",
    "pfId",
    "prodId",
    "prodPrice",
    "state",
    "ver",
    "customData",
]


def _format_value(val: Any) -> str:
    if val is None:
        return ""
    if isinstance(val, datetime):
        return val.isoformat()
    return str(val)


def entity_to_row(entity: dict[str, Any]) -> dict[str, str]:
    return {col: _format_value(entity.get(col)) for col in CSV_COLUMNS}


def _parse_row(raw: dict[str, str]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for col in CSV_COLUMNS:
        val = raw.get(col, "")
        if val == "":
            result[col] = None
            continue
        if col in ("OrderTime", "ReceiveTime"):
            result[col] = datetime.fromisoformat(val)
        elif col in ("revenue", "prodPrice"):
            result[col] = float(val)
        elif col in ("prodCount", "state", "ver"):
            result[col] = int(val)
        else:
            result[col] = val
    return result


def row_to_entity(raw: dict[str, str]) -> dict[str, Any]:
    return _parse_row(raw)


def write_csv(path: Path, entities: list[dict[str, Any]]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=CSV_COLUMNS)
        writer.writeheader()
        for entity in entities:
            writer.writerow(entity_to_row(entity))


def read_csv(path: Path):
    with open(path, newline="") as f:
        reader = csv.DictReader(f)
        for row in reader:
            yield _parse_row(row)

Step 4: Run test to verify it passes

Run: uv run pytest tests/test_csv_utils.py -v Expected: 5 passed

Step 5: Commit

git add src/order_bak/csv_utils.py tests/test_csv_utils.py
git commit -m "feat: CSV utilities for order entity serialization"

Task 4: Scan Command

Files:

Create: src/order_bak/scan.py
Modify: src/order_bak/cli.py
Step 1: Implement scan logic

# src/order_bak/scan.py
import time
from datetime import datetime, timezone
from pathlib import Path

from azure.data.tables import TableClient

from order_bak.csv_utils import CSV_COLUMNS, write_csv


def scan_orders(
    connection_string: str,
    page_size: int,
    delay_ms: int,
    scan_dir: Path,
) -> Path:
    table = TableClient.from_connection_string(connection_string, table_name="order")
    scan_dir.mkdir(parents=True, exist_ok=True)

    now = datetime.now(timezone.utc)
    csv_path = scan_dir / f"scan-{now.strftime('%Y-%m-%d')}.csv"

    paged = table.list_entities(results_per_page=page_size)

    total_scanned = 0
    csv_path.parent.mkdir(parents=True, exist_ok=True)

    with open(csv_path, "w", newline="") as f:
        import csv
        writer = csv.DictWriter(f, fieldnames=CSV_COLUMNS)
        writer.writeheader()

        for page in paged.by_page():
            for entity in page:
                writer.writerow({col: _fmt(entity.get(col)) for col in CSV_COLUMNS})
                total_scanned += 1

            if delay_ms > 0:
                time.sleep(delay_ms / 1000.0)

    print(f"Scan complete: {total_scanned} orders saved to {csv_path}")
    return csv_path


def _fmt(val):
    if val is None:
        return ""
    if isinstance(val, datetime):
        return val.isoformat()
    return str(val)

Step 2: Wire scan command into CLI

# src/order_bak/cli.py
from pathlib import Path

import click

from order_bak.config import load_config


@click.group()
@click.option("--config", "-c", default="~/.config/order-bak/config.toml", type=click.Path())
@click.pass_context
def main(ctx, config):
    ctx.ensure_object(dict)
    ctx.obj["config_path"] = Path(config).expanduser()


@main.command()
@click.pass_context
def scan(ctx):
    cfg = load_config(ctx.obj["config_path"])
    from order_bak.scan import scan_orders
    scan_orders(
        connection_string=cfg.azure_connection_string,
        page_size=cfg.scan_page_size,
        delay_ms=cfg.scan_delay_ms,
        scan_dir=cfg.scan_dir,
    )


if __name__ == "__main__":
    main()

Step 3: Verify CLI help works

Run: uv run order-bak --help Expected: shows scan command

Run: uv run order-bak scan --help Expected: shows scan help

Step 4: Commit

git add src/order_bak/scan.py src/order_bak/cli.py
git commit -m "feat: scan command — full table scan to CSV with throttling"

Task 5: Backup Command

Files:

Create: src/order_bak/backup.py
Create: tests/test_backup.py
Modify: src/order_bak/cli.py
Step 1: Write failing test

# tests/test_backup.py
from datetime import datetime, timezone
from pathlib import Path

from order_bak.csv_utils import write_csv
from order_bak.backup import backup_orders


def _make_entity(pk, rk, order_time, state, receive_time=None):
    return {
        "PartitionKey": pk,
        "RowKey": rk,
        "OrderTime": order_time,
        "ReceiveTime": receive_time,
        "revenue": 10.0,
        "platformOrder": "plat1",
        "prodCount": 1,
        "pfId": pk,
        "prodId": "prod1",
        "prodPrice": 10.0,
        "state": state,
        "ver": 3,
        "customData": None,
    }


def test_backup_filters_by_state_and_date(tmp_path):
    scan_dir = tmp_path / "scan"
    backup_dir = tmp_path / "backups"
    scan_dir.mkdir()
    backup_dir.mkdir()

    entities = [
        _make_entity("u1", "o1", datetime(2025, 1, 1, 10, tzinfo=timezone.utc), 1),
        _make_entity("u2", "o2", datetime(2025, 1, 1, 12, tzinfo=timezone.utc), 0),  # Payed, skip
        _make_entity("u3", "o3", datetime(2025, 1, 2, 8, tzinfo=timezone.utc), 1),
        _make_entity("u4", "o4", datetime(2025, 1, 3, 8, tzinfo=timezone.utc), 1),  # outside range
    ]
    scan_csv = scan_dir / "scan-2025-01-03.csv"
    write_csv(scan_csv, entities)

    result = backup_orders(
        scan_path=scan_csv,
        backup_dir=backup_dir,
        start=datetime(2025, 1, 1, tzinfo=timezone.utc),
        end=datetime(2025, 1, 3, tzinfo=timezone.utc),
    )

    assert result == 2
    jan1 = backup_dir / "2025-01-01.csv"
    jan2 = backup_dir / "2025-01-02.csv"
    assert jan1.exists()
    assert jan2.exists()
    assert not (backup_dir / "2025-01-03.csv").exists()

    from order_bak.csv_utils import read_csv
    jan1_rows = list(read_csv(jan1))
    assert len(jan1_rows) == 1
    assert jan1_rows[0]["RowKey"] == "o1"

Step 2: Run test to verify it fails

Run: uv run pytest tests/test_backup.py -v Expected: FAIL — ModuleNotFoundError: No module named 'order_bak.backup'

Step 3: Implement backup

# src/order_bak/backup.py
import csv
from collections import defaultdict
from datetime import datetime
from pathlib import Path
from typing import Iterator

from order_bak.csv_utils import CSV_COLUMNS, _parse_row


def backup_orders(
    scan_path: Path,
    backup_dir: Path,
    start: datetime,
    end: datetime,
) -> int:
    backup_dir.mkdir(parents=True, exist_ok=True)

    by_date: dict[str, list[dict]] = defaultdict(list)

    with open(scan_path, newline="") as f:
        reader = csv.DictReader(f)
        for raw_row in reader:
            entity = _parse_row(raw_row)
            if entity["state"] != 1:
                continue
            order_time = entity["OrderTime"]
            if order_time is None:
                continue
            if order_time < start or order_time >= end:
                continue
            date_key = order_time.strftime("%Y-%m-%d")
            by_date[date_key].append(raw_row)

    total = 0
    for date_key, rows in sorted(by_date.items()):
        csv_path = backup_dir / f"{date_key}.csv"
        with open(csv_path, "w", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=CSV_COLUMNS)
            writer.writeheader()
            writer.writerows(rows)
        total += len(rows)
        print(f"  {date_key}: {len(rows)} orders")

    print(f"Backup complete: {total} orders across {len(by_date)} days")
    return total

Step 4: Run test to verify it passes

Run: uv run pytest tests/test_backup.py -v Expected: 1 passed

Step 5: Wire backup command into CLI

Add to src/order_bak/cli.py:

@main.command()
@click.option("--start", required=True, type=click.DateTime(formats=["%Y-%m-%d"]), help="Start date (UTC, inclusive)")
@click.option("--end", required=True, type=click.DateTime(formats=["%Y-%m-%d"]), help="End date (UTC, exclusive)")
@click.option("--scan", "scan_path", default=None, type=click.Path(), help="Path to scan CSV (default: latest in scan_dir)")
@click.pass_context
def backup(ctx, start, end, scan_path):
    cfg = load_config(ctx.obj["config_path"])
    from order_bak.backup import backup_orders
    import pathlib

    if scan_path:
        scan_file = pathlib.Path(scan_path)
    else:
        import glob
        scan_files = sorted(cfg.scan_dir.glob("scan-*.csv"))
        if not scan_files:
            click.echo("No scan files found in scan_dir")
            raise SystemExit(1)
        scan_file = scan_files[-1]

    start_utc = start.replace(tzinfo=__import__("datetime").timezone.utc)
    end_utc = end.replace(tzinfo=__import__("datetime").timezone.utc)

    backup_orders(
        scan_path=scan_file,
        backup_dir=pathlib.Path("./backups"),
        start=start_utc,
        end=end_utc,
    )

Step 6: Verify CLI help

Run: uv run order-bak backup --help Expected: shows --start, --end, --scan options

Step 7: Commit

git add src/order_bak/backup.py tests/test_backup.py src/order_bak/cli.py
git commit -m "feat: backup command — filter scan CSV by state and date range"

Task 6: Delete Command

Files:

Create: src/order_bak/delete.py
Create: tests/test_delete.py
Modify: src/order_bak/cli.py
Step 1: Write failing test

# tests/test_delete.py
from collections import defaultdict
from unittest.mock import MagicMock, call

from order_bak.delete import delete_orders


def test_delete_groups_by_partition_and_batches(tmp_path):
    # Create a CSV with entities across different partitions
    from order_bak.csv_utils import write_csv
    from datetime import datetime, timezone

    entities = []
    # 3 entities in partition "user1", 2 in "user2"
    for i in range(3):
        entities.append({
            "PartitionKey": "user1",
            "RowKey": f"order{i}",
            "OrderTime": datetime(2025, 1, 1, tzinfo=timezone.utc),
            "ReceiveTime": None, "revenue": 10.0, "platformOrder": "p",
            "prodCount": 1, "pfId": "user1", "prodId": "prod",
            "prodPrice": 10.0, "state": 1, "ver": 3, "customData": None,
        })
    for i in range(3, 5):
        entities.append({
            "PartitionKey": "user2",
            "RowKey": f"order{i}",
            "OrderTime": datetime(2025, 1, 1, tzinfo=timezone.utc),
            "ReceiveTime": None, "revenue": 10.0, "platformOrder": "p",
            "prodCount": 1, "pfId": "user2", "prodId": "prod",
            "prodPrice": 10.0, "state": 1, "ver": 3, "customData": None,
        })

    csv_path = tmp_path / "2025-01-01.csv"
    write_csv(csv_path, entities)

    mock_table = MagicMock()

    deleted_count = delete_orders(
        table_client=mock_table,
        csv_path=csv_path,
        batch_size=2,
        delay_ms=0,
    )

    assert deleted_count == 5
    # user1 has 3 entities, batch_size=2 → 2 batches (2 + 1)
    # user2 has 2 entities, batch_size=2 → 1 batch
    assert mock_table.submit_transaction.call_count == 3


def test_delete_respects_batch_size(tmp_path):
    from order_bak.csv_utils import write_csv
    from datetime import datetime, timezone

    # 5 entities in same partition, batch_size=2
    entities = [
        {
            "PartitionKey": "user1",
            "RowKey": f"order{i}",
            "OrderTime": datetime(2025, 1, 1, tzinfo=timezone.utc),
            "ReceiveTime": None, "revenue": 10.0, "platformOrder": "p",
            "prodCount": 1, "pfId": "user1", "prodId": "prod",
            "prodPrice": 10.0, "state": 1, "ver": 3, "customData": None,
        }
        for i in range(5)
    ]

    csv_path = tmp_path / "2025-01-01.csv"
    write_csv(csv_path, entities)

    mock_table = MagicMock()

    deleted_count = delete_orders(
        table_client=mock_table,
        csv_path=csv_path,
        batch_size=2,
        delay_ms=0,
    )

    assert deleted_count == 5
    assert mock_table.submit_transaction.call_count == 3  # 2+2+1

Step 2: Run test to verify it fails

Run: uv run pytest tests/test_delete.py -v Expected: FAIL — ModuleNotFoundError: No module named 'order_bak.delete'

Step 3: Implement delete

# src/order_bak/delete.py
import csv
import time
from collections import defaultdict
from itertools import islice
from pathlib import Path


def _chunked(iterable, size):
    it = iter(iterable)
    while chunk := list(islice(it, size)):
        yield chunk


def delete_orders(
    table_client,
    csv_path: Path,
    batch_size: int,
    delay_ms: int,
) -> int:
    by_partition: dict[str, list[dict]] = defaultdict(list)

    with open(csv_path, newline="") as f:
        reader = csv.DictReader(f)
        for row in reader:
            pk = row["PartitionKey"]
            rk = row["RowKey"]
            by_partition[pk].append({"PartitionKey": pk, "RowKey": rk})

    total_deleted = 0
    for pk, entities in by_partition.items():
        for batch in _chunked(entities, batch_size):
            operations = [("delete", e) for e in batch]
            table_client.submit_transaction(operations)
            total_deleted += len(batch)
            if delay_ms > 0:
                time.sleep(delay_ms / 1000.0)

    print(f"Deleted {total_deleted} orders across {len(by_partition)} partitions")
    return total_deleted

Step 4: Run test to verify it passes

Run: uv run pytest tests/test_delete.py -v Expected: 2 passed

Step 5: Wire delete command into CLI

Add to src/order_bak/cli.py:

@main.command()
@click.option("--csv", "csv_path", required=True, type=click.Path(exists=True), help="Path to backup CSV to delete")
@click.pass_context
def delete(ctx, csv_path):
    cfg = load_config(ctx.obj["config_path"])
    from azure.data.tables import TableClient
    from order_bak.delete import delete_orders
    import pathlib

    csv_file = pathlib.Path(csv_path)

    # Count records first
    import csv as csv_mod
    with open(csv_file, newline="") as f:
        count = sum(1 for _ in csv_mod.DictReader(f))

    click.echo(f"Will delete {count} orders from {csv_file}")
    if not click.confirm("Continue?"):
        click.echo("Aborted")
        return

    table = TableClient.from_connection_string(cfg.azure_connection_string, table_name="order")
    delete_orders(
        table_client=table,
        csv_path=csv_file,
        batch_size=cfg.delete_batch_size,
        delay_ms=cfg.delete_delay_ms,
    )

Step 6: Verify CLI help

Run: uv run order-bak delete --help Expected: shows --csv option

Step 7: Commit

git add src/order_bak/delete.py tests/test_delete.py src/order_bak/cli.py
git commit -m "feat: delete command — batch delete from CSV with confirmation"

Task 7: Run All Tests

Step 1: Run full test suite

Run: uv run pytest tests/ -v Expected: all tests pass

Step 2: Run lint check

Run: uv run python -m py_compile src/order_bak/cli.py Expected: no errors

Self-Review Checklist

Spec coverage: scan (Task 4), backup (Task 5), delete (Task 6), config (Task 2), CSV format (Task 3) — all covered
No placeholders: all code blocks contain complete implementations
Type consistency: entity dict keys match across csv_utils, backup, delete — all use CSV_COLUMNS

26 KiB Raw Blame History

Order Backup CLI — Implementation Plan

File Structure

Task 1: Project Scaffold

Task 2: Config Module

Task 3: CSV Utilities

Task 4: Scan Command

Task 5: Backup Command

Task 6: Delete Command

Task 7: Run All Tests

Self-Review Checklist

26 KiB

Raw Blame History