feat(graph): add bulk graph import

README.md

@@ -501,6 +501,75 @@ db.graph_delete_edge("acme", "competes_with", "globex", namespace="kb", bidirect
 db.graph_delete_node("globex", namespace="kb")
 ```
 
+Load graph records with grouped DynamoDB updates:
+
+```python
+result = db.graph_import(
+    namespace="kb",
+    nodes=[{"id": "acme", "ntype": "company", "props": {"country": "US"}}],
+    edges=[{"src": "acme", "relation": "supplies", "dst": "globex",
+            "weight": 2.5, "props": {"created_at": "2026-10-07T12:00:00Z"}}],
+    batch_size=100,
+)
+print(result.nodes_processed, result.edges_appended)  # 2, 1
+
+db.graph_import_file(namespace="kb", json_file="graph.json")
+db.graph_import_file(namespace="kb", nodes_csv="nodes.csv", edges_csv="edges.csv")
+# Either CSV file can also be supplied alone.
+```
+
+Both client methods default to namespace `"default"`. The storage APIs are
+`GraphStore.import_graph(ns, *, nodes=None, edges=None, batch_size=100)` and
+`GraphStore.import_graph_file(ns, *, json_file=None, nodes_csv=None, edges_csv=None,
+batch_size=100)`. They return `GraphImportResult`: `nodes_processed` counts unique
+IDs, including inferred edge endpoints and already-existing nodes;
+`edges_appended` counts input edge records. Empty record collections return zeros.
+
+- **Python:** `nodes` and `edges` are sequences of dictionaries. Nodes require a
+  non-empty string `id`; optional `ntype` is a string or `None`, and `props` is a
+  dictionary or `None`. Edges require non-empty strings `src`, `relation`, and
+  `dst`; optional `weight` defaults to `1.0`, and omitted/`None` `props` defaults
+  to `{}`. Identifiers are stored unchanged, including spaces.
+- **JSON:** a UTF-8 object containing optional `nodes` and `edges` arrays using
+  those same fields. Missing arrays mean empty collections; `null` arrays are
+  invalid. For example: `{"nodes": [{"id": "acme"}], "edges": [{"src": "acme",
+  "relation": "supplies", "dst": "globex"}]}`.
+- **CSV:** separate UTF-8 files. Node headers: required `id`, optional `ntype,props`.
+  Edge headers: required `src,relation,dst`, optional `weight,props`. A `props`
+  cell must contain a JSON object; use standard CSV quoting, e.g.
+  `acme,company,"{""country"":""US""}"`. Blank optional cells mean omitted values.
+  Supply JSON or CSV files in one call; combining formats is invalid.
+
+Existing edges and document links are preserved. Omitted/`None` node attributes
+preserve existing values; explicit node `props` replaces the entire dictionary,
+so `{}` clears it. For repeated node IDs, later supplied attributes win, while
+omitted values preserve earlier ones. Missing edge endpoints are created without
+overwriting existing endpoint metadata. Duplicate edges remain duplicates, and
+repeating an import appends them again: imports are **not idempotent**.
+
+Finite numeric weights may be negative or greater than one; booleans, NaN,
+infinity, and numbers outside DynamoDB's supported precision/range are rejected.
+Nested floats in properties are copied as `Decimal` values for DynamoDB, caller
+inputs are not mutated, and timestamp strings remain caller-provided strings.
+Unsupported DynamoDB values are rejected before writing.
+
+Every supplied record/file is loaded and validated before any database writes,
+including when both CSV files are supplied. Errors identify a record index or
+filename/CSV row. Complete prevalidation uses memory proportional to the import.
+Each unique node is prepared once, then edges are grouped by source and appended
+in input order within that source. `batch_size` must be a positive integer
+(excluding booleans) and limits edges per append request. Three edges from one
+source to three destinations use five `UpdateItem` requests with the default
+batch size, versus nine via three `add_edge()` calls. This is grouping, not
+compression or a single AWS request; `BatchWriteItem` is not used.
+
+The existing **400 KB DynamoDB item limit** still applies to each node's complete
+adjacency list, even with small chunks. AWS failures can leave partial writes;
+there is no whole-import rollback or exactly-once guarantee. Safe node preparation
+uses the existing retry policy, but the importer does not retry edge appends or
+restart completed chunks. SDK retries can still make ambiguous append failures
+unsafe to replay; inspect the stored graph before deciding how to recover.
+
 ## Ecosystem
 
 dynavec started as a vector database, and it's growing into a full agent stack that runs inside your own cloud account. The plan has four parts:

src/dynavec/__init__.py

@@ -39,6 +39,7 @@
 )
 from .fusion import FitResult, RRFWeightFitter
 from .graph import GraphStore
+from .graph_import import GraphImportResult
 from .hot import HotTier
 from .ingest import (
     CsvSource,
@@ -140,6 +141,7 @@
     "OPQRotation",
     "OptimizedProductQuantizer",
     "GraphStore",
+    "GraphImportResult",
     "BaseCache",
     "SemanticCache",
     "DynamoDBCache",

src/dynavec/client.py

@@ -65,6 +65,7 @@
     NotFoundError,
 )
 from .graph import GraphStore
+from .graph_import import GraphImportResult
 from .hot import HotTier
 from .metadata import build_s3_filter
 from .metrics import normalize_scores as normalize_metric_scores
@@ -1231,6 +1232,35 @@ def list_vectors(
                 )
 
     # -------------------------------------------------------------- graph / ER
+    def graph_import(
+        self,
+        *,
+        nodes: Sequence[dict[str, Any]] | None = None,
+        edges: Sequence[dict[str, Any]] | None = None,
+        namespace: str = "default",
+        batch_size: int = 100,
+    ) -> GraphImportResult:
+        """Import graph records with grouped adjacency updates."""
+        return self.graph.import_graph(namespace, nodes=nodes, edges=edges, batch_size=batch_size)
+
+    def graph_import_file(
+        self,
+        *,
+        json_file: str | Path | None = None,
+        nodes_csv: str | Path | None = None,
+        edges_csv: str | Path | None = None,
+        namespace: str = "default",
+        batch_size: int = 100,
+    ) -> GraphImportResult:
+        """Import graph JSON or node/edge CSV files after complete validation."""
+        return self.graph.import_graph_file(
+            namespace,
+            json_file=json_file,
+            nodes_csv=nodes_csv,
+            edges_csv=edges_csv,
+            batch_size=batch_size,
+        )
+
     def graph_add_node(
         self,
         entity_id: str,

src/dynavec/graph.py

@@ -26,12 +26,16 @@
 
 import re
 from collections import deque
+from collections.abc import Sequence
 from decimal import Decimal
+from pathlib import Path
 from typing import Any, Literal, cast, overload
 
 from .config import DynavecConfig
+from .graph_import import GraphImportResult, _load_files, _prepare_import, _PreparedImport
 from .utils import (
     KEY_SEPARATOR,
+    chunked,
     decode_key_component,
     encode_key_component,
     is_retryable,
@@ -50,6 +54,7 @@ def _edges_changed_or_retryable(exc: Exception) -> bool:
     code = getattr(exc, "response", {}).get("Error", {}).get("Code", "")
     return code == "ConditionalCheckFailedException" or is_retryable(exc)
 
+
 EXPORT_FORMATS = ("mermaid", "dot")
 
 # A bare Mermaid flowchart node id: letters/digits/underscore, not keyword-shaped.
@@ -184,6 +189,61 @@ def _node_pk(ns: str, entity_id: str) -> str:
         )
 
     # --------------------------------------------------------------- mutations
+    def import_graph(
+        self,
+        ns: str,
+        *,
+        nodes: Sequence[dict[str, Any]] | None = None,
+        edges: Sequence[dict[str, Any]] | None = None,
+        batch_size: int = 100,
+    ) -> GraphImportResult:
+        """Prevalidate records, prepare unique nodes, and append grouped edges.
+
+        ``batch_size`` bounds edges per UpdateItem, not item size. Existing
+        metadata follows add_node semantics; duplicate edges remain duplicates.
+        Prevalidation needs memory proportional to the import. AWS failures can
+        leave partial writes; this operation is neither atomic nor idempotent.
+        """
+        return self._write_import(ns, _prepare_import(ns, nodes, edges, batch_size), batch_size)
+
+    def import_graph_file(
+        self,
+        ns: str,
+        *,
+        json_file: str | Path | None = None,
+        nodes_csv: str | Path | None = None,
+        edges_csv: str | Path | None = None,
+        batch_size: int = 100,
+    ) -> GraphImportResult:
+        """Load UTF-8 JSON or separate CSV files; validate both before writing.
+
+        Supply a JSON object with nodes/edges arrays, or either/both CSV files.
+        CSV props cells hold JSON objects. Errors include file/record context.
+        """
+        nodes, edges, node_contexts, edge_contexts = _load_files(json_file, nodes_csv, edges_csv)
+        prepared = _prepare_import(
+            ns, nodes, edges, batch_size, node_contexts=node_contexts, edge_contexts=edge_contexts
+        )
+        return self._write_import(ns, prepared, batch_size)
+
+    def _write_import(
+        self, ns: str, prepared: _PreparedImport, batch_size: int
+    ) -> GraphImportResult:
+        # add_node is safe to retry: it initializes lists only if absent and
+        # assigns the same prevalidated metadata on every attempt.
+        for entity_id, attrs in prepared.nodes.items():
+            self.add_node(ns, entity_id, attrs.get("ntype"), attrs.get("props"))
+        for src, edges in prepared.edges.items():
+            for batch in chunked(edges, batch_size):
+                # No application retry around appends or the multi-write import:
+                # an ambiguous failure may have already appended the chunk.
+                self._table.update_item(
+                    Key={"pk": self._node_pk(ns, src)},
+                    UpdateExpression="SET edges = list_append(if_not_exists(edges, :empty), :e)",
+                    ExpressionAttributeValues={":e": batch, ":empty": []},
+                )
+        return GraphImportResult(len(prepared.nodes), prepared.edge_count)
+
     @retry()
     def add_node(
         self, ns: str, entity_id: str, ntype: str | None = None, props: Props | None = None