diff --git a/README.md b/README.md
index 9dd48bf..6ea4bec 100644
--- a/README.md
+++ b/README.md
@@ -77,6 +77,15 @@ uv sync
 uv run pytest -m integration
 ```
 
+### Database schema
+
+Constraints and indexes are created idempotently (safe to run repeatedly):
+
+```bash
+cd backend
+uv run python -m visual_graph.graph
+```
+
 ## License
 
 This project is licensed under the MIT License.
diff --git a/backend/src/visual_graph/graph/__main__.py b/backend/src/visual_graph/graph/__main__.py
new file mode 100644
index 0000000..8c52371
--- /dev/null
+++ b/backend/src/visual_graph/graph/__main__.py
@@ -0,0 +1,18 @@
+"""``python -m visual_graph.graph`` applies the database schema."""
+
+from visual_graph.graph.connection import create_driver, verify_connection
+from visual_graph.graph.schema import apply_schema
+
+
+def main() -> None:
+    driver = create_driver()
+    try:
+        verify_connection(driver)
+        apply_schema(driver)
+    finally:
+        driver.close()
+    print("Neo4j schema applied.")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/backend/src/visual_graph/graph/schema.py b/backend/src/visual_graph/graph/schema.py
new file mode 100644
index 0000000..eed1a6f
--- /dev/null
+++ b/backend/src/visual_graph/graph/schema.py
@@ -0,0 +1,41 @@
+"""Idempotent Neo4j schema: constraints and indexes.
+
+Every statement uses ``IF NOT EXISTS`` so ``apply_schema`` can run any number of times.
+"""
+
+from neo4j import Driver
+
+SCHEMA_STATEMENTS: tuple[str, ...] = (
+    "CREATE CONSTRAINT image_id_unique IF NOT EXISTS "
+    "FOR (n:Image) REQUIRE n.id IS UNIQUE",
+    "CREATE CONSTRAINT object_instance_id_unique IF NOT EXISTS "
+    "FOR (n:ObjectInstance) REQUIRE n.id IS UNIQUE",
+    # Class uniqueness strategy: `name` is the natural key; create classes with MERGE.
+    "CREATE CONSTRAINT class_name_unique IF NOT EXISTS "
+    "FOR (n:Class) REQUIRE n.name IS UNIQUE",
+    # Required properties would need Enterprise edition, so they are enforced in code.
+    "CREATE INDEX image_uri IF NOT EXISTS FOR (n:Image) ON (n.uri)",
+    "CREATE INDEX object_instance_confidence IF NOT EXISTS "
+    "FOR (n:ObjectInstance) ON (n.confidence)",
+)
+
+EXPECTED_CONSTRAINTS = frozenset(
+    {"image_id_unique", "object_instance_id_unique", "class_name_unique"}
+)
+EXPECTED_INDEXES = frozenset({"image_uri", "object_instance_confidence"})
+
+
+def apply_schema(driver: Driver, database: str | None = None) -> None:
+    with driver.session(database=database) as session:
+        for statement in SCHEMA_STATEMENTS:
+            session.run(statement).consume()
+        # Wait until indexes are online before returning.
+        session.run("CALL db.awaitIndexes(60)").consume()
+
+
+def existing_schema(driver: Driver, database: str | None = None) -> tuple[set[str], set[str]]:
+    """Return ``(constraint names, index names)`` currently in the database."""
+    with driver.session(database=database) as session:
+        constraints = {r["name"] for r in session.run("SHOW CONSTRAINTS YIELD name")}
+        indexes = {r["name"] for r in session.run("SHOW INDEXES YIELD name")}
+    return constraints, indexes
diff --git a/backend/tests/integration/test_schema.py b/backend/tests/integration/test_schema.py
new file mode 100644
index 0000000..8ab85c7
--- /dev/null
+++ b/backend/tests/integration/test_schema.py
@@ -0,0 +1,62 @@
+from collections.abc import Iterator
+
+import pytest
+from neo4j import Driver
+from neo4j.exceptions import ConstraintError
+
+from visual_graph.graph.connection import create_driver
+from visual_graph.graph.schema import (
+    EXPECTED_CONSTRAINTS,
+    EXPECTED_INDEXES,
+    apply_schema,
+    existing_schema,
+)
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture
+def driver() -> Iterator[Driver]:
+    drv = create_driver()
+    with drv.session() as session:
+        session.run("MATCH (n) DETACH DELETE n").consume()
+    yield drv
+    with drv.session() as session:
+        session.run("MATCH (n) DETACH DELETE n").consume()
+    drv.close()
+
+
+def test_schema_creates_constraints_and_indexes(driver: Driver) -> None:
+    apply_schema(driver)
+    constraints, indexes = existing_schema(driver)
+    assert EXPECTED_CONSTRAINTS <= constraints
+    assert EXPECTED_INDEXES <= indexes
+
+
+def test_schema_is_idempotent(driver: Driver) -> None:
+    apply_schema(driver)
+    before = existing_schema(driver)
+    apply_schema(driver)
+    apply_schema(driver)
+    assert existing_schema(driver) == before
+
+
+@pytest.mark.parametrize(
+    ("label", "prop"),
+    [("Image", "id"), ("ObjectInstance", "id"), ("Class", "name")],
+)
+def test_duplicates_are_rejected(driver: Driver, label: str, prop: str) -> None:
+    apply_schema(driver)
+    with driver.session() as session:
+        session.run(f"CREATE (:{label} {{{prop}: 'dup'}})").consume()
+        with pytest.raises(ConstraintError):
+            session.run(f"CREATE (:{label} {{{prop}: 'dup'}})").consume()
+
+
+def test_merge_does_not_duplicate_classes(driver: Driver) -> None:
+    apply_schema(driver)
+    with driver.session() as session:
+        for _ in range(3):
+            session.run("MERGE (:Class {name: 'person'})").consume()
+        count = session.run("MATCH (c:Class {name: 'person'}) RETURN count(c) AS n").single()
+    assert count is not None and count["n"] == 1