ourdigital/our-claude-skills
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(notion): migrate 31+32 to API 2025-09-03 + add property writes, upsert, anchor-link fix

Browse Source 
Phase 1 — docs alignment:
- Fix path drift in 32 CLAUDE.md (02-notion-writer → 32-notion-writer)
- Align env var to NOTION_API_KEY in 31 docs (NOTION_TOKEN still accepted)
- Document Phase 3 roadmap in 31 (metadata-aware migration, Notion-as-RAG)

Phase 2 — multi-source database support:
- New 32/scripts/_notion_compat.py: client factory, data_source resolution
  with cache, property coercion, find_existing_page (for upsert),
  explain_api_error for friendlier failure messages
- 32 notion_writer.py: drop the 2022-06-28 pin, route schema introspection
  through data_sources.retrieve, build data_source_id parent shape, accept
  arbitrary properties via --properties JSON-or-file flag, add --upsert-by
  for idempotent re-runs
- 32 markdown parser: anchor-link fix — Notion's URL validator rejects
  fragment URLs and relative paths; fragments now render as bold to preserve
  TOC nav intent, relatives drop the link, absolute URLs preserved
- 31 async_organizer.py + schema_migrator.py: same data_sources migration
  with cached resolution; pages.create now uses data_source_id parent
- 16/16 parser tests pass (was 13; +3 link-handling regression guards)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Andrew Yim
2026-04-27 09:57:20 +09:00
parent 2667304bca
commit 144a17c88d
7 changed files with 678 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
custom-skills/31-notion-organizer/code/CLAUDE.md
View File

@@ -79,11 +79,40 @@ python scripts/async_organizer.py --database [ID] --action restructure
Environment variables:
```bash
NOTION_TOKEN=secret_xxx
# Preferred (matches 32-notion-writer)
NOTION_API_KEY=secret_xxx
# Legacy fallback also accepted
# NOTION_TOKEN=secret_xxx
```
The scripts read `NOTION_TOKEN` first, then fall back to `NOTION_API_KEY`. Use `NOTION_API_KEY` for new setups so the same `.env` works across all Notion skills.
## Notes
- Always use `--dry-run` first for destructive operations
- Large operations (1000+ pages) use async with progress reporting
- Scripts implement automatic rate limiting
## Roadmap (Phase 3)
The current scripts cover schema migration and async bulk ops. Two larger goals are parked for a future iteration:
### Goal 1 — Metadata-aware page move/integration
Search source pages, compare their property metadata against a target database schema, and move/migrate while transforming property values to fit. Beyond the existing `schema_migrator.py` (which expects a hand-written mapping file), this would:
- Auto-suggest property mappings using name + type similarity
- Surface unmappable properties before the move (no silent data loss)
- Support cross-database moves (not just same-schema migrations)
### Goal 2 — Notion as personal RAG source
Treat Notion as a small, personal knowledge base for AI agents:
- Search pages by query across databases, filter by property
- Merge results into a single context (with source citations back to page URLs)
- Summarize / distill via LLM into agent-ready snippets
- Export as JSONL or markdown for fine-tuning or RAG indexing
This dovetails with `90-reference-curator` (which does the same for web sources) — Phase 3 would make Notion a first-class source type for that pipeline.

64
custom-skills/31-notion-organizer/code/scripts/async_organizer.py
View File

@@ -55,6 +55,45 @@ class NotionAsyncOrganizer:
self.semaphore = Semaphore(MAX_CONCURRENT_REQUESTS)
self.dry_run = dry_run
self.stats = {"fetched": 0, "updated": 0, "created": 0, "errors": 0}
self._data_source_cache: dict[str, str] = {}
async def _resolve_data_source_id(self, target_id: str) -> str:
"""
Accept either a database_id or data_source_id; return data_source_id.
Caches results so repeated calls within one organizer run don't re-hit
the API. Errors propagate to the caller for proper handling.
"""
if target_id in self._data_source_cache:
return self._data_source_cache[target_id]
from notion_client.errors import APIErrorCode, APIResponseError
try:
await self._rate_limited_request(
self.client.data_sources.retrieve(data_source_id=target_id)
)
self._data_source_cache[target_id] = target_id
return target_id
except APIResponseError as exc:
if exc.code != APIErrorCode.ObjectNotFound:
raise
db = await self._rate_limited_request(
self.client.databases.retrieve(database_id=target_id)
)
sources = db.get("data_sources") or []
if not sources:
raise ValueError(f"Database {target_id} has no data sources attached")
if len(sources) > 1:
names = ", ".join(s.get("name", s["id"]) for s in sources)
raise ValueError(
f"Database {target_id} has multiple data sources ({names}); "
"pass a specific data_source_id"
)
ds_id = sources[0]["id"]
self._data_source_cache[target_id] = ds_id
return ds_id
@retry(
stop=stop_after_attempt(3),
@@ -67,31 +106,33 @@ class NotionAsyncOrganizer:
await asyncio.sleep(REQUEST_DELAY)
return await coro
async def fetch_database_schema(self, database_id: str) -> dict:
"""Fetch database schema/properties."""
logger.info(f"Fetching database schema: {database_id}")
async def fetch_database_schema(self, target_id: str) -> dict:
"""Fetch full schema (properties) for a database or data source."""
data_source_id = await self._resolve_data_source_id(target_id)
logger.info(f"Fetching schema: {data_source_id}")
response = await self._rate_limited_request(
self.client.databases.retrieve(database_id=database_id)
self.client.data_sources.retrieve(data_source_id=data_source_id)
)
self.stats["fetched"] += 1
return response
async def fetch_all_pages(
self,
database_id: str,
target_id: str,
filter_obj: dict | None = None,
sorts: list | None = None,
) -> list[dict]:
"""Fetch all pages from a database with pagination."""
"""Fetch all pages from a data source with pagination."""
data_source_id = await self._resolve_data_source_id(target_id)
all_pages = []
has_more = True
start_cursor = None
logger.info(f"Fetching pages from database: {database_id}")
logger.info(f"Fetching pages from data source: {data_source_id}")
while has_more:
query_params = {
"database_id": database_id,
"data_source_id": data_source_id,
"page_size": 100,
}
if start_cursor:
@@ -102,7 +143,7 @@ class NotionAsyncOrganizer:
query_params["sorts"] = sorts
response = await self._rate_limited_request(
self.client.databases.query(**query_params)
self.client.data_sources.query(**query_params)
)
all_pages.extend(response["results"])
@@ -287,7 +328,10 @@ async def example_bulk_status_update(
async def main():
"""Main entry point."""
parser = argparse.ArgumentParser(description="Notion Async Organizer")
parser.add_argument("--database-id", "-d", required=True, help="Database ID")
parser.add_argument(
"--database-id", "-d", required=True,
help="Database ID or data source ID (resolved automatically)",
)
parser.add_argument(
"--dry-run", action="store_true", help="Preview changes without executing"
)

69
custom-skills/31-notion-organizer/code/scripts/schema_migrator.py
View File

@@ -53,6 +53,7 @@ class SchemaMigrator:
"pages_skipped": 0,
"errors": 0,
}
self._data_source_cache: dict[str, str] = {}
@retry(
stop=stop_after_attempt(3),
@@ -63,24 +64,59 @@ class SchemaMigrator:
await asyncio.sleep(REQUEST_DELAY)
return await coro
async def get_schema(self, database_id: str) -> dict:
"""Get database schema."""
async def _resolve_data_source_id(self, target_id: str) -> str:
"""Accept database_id or data_source_id; return data_source_id."""
if target_id in self._data_source_cache:
return self._data_source_cache[target_id]
from notion_client.errors import APIErrorCode, APIResponseError
try:
await self._request(
self.client.data_sources.retrieve(data_source_id=target_id)
)
self._data_source_cache[target_id] = target_id
return target_id
except APIResponseError as exc:
if exc.code != APIErrorCode.ObjectNotFound:
raise
db = await self._request(self.client.databases.retrieve(database_id=target_id))
sources = db.get("data_sources") or []
if not sources:
raise ValueError(f"Database {target_id} has no data sources attached")
if len(sources) > 1:
names = ", ".join(s.get("name", s["id"]) for s in sources)
raise ValueError(
f"Database {target_id} has multiple data sources ({names}); "
"pass a specific data_source_id"
)
ds_id = sources[0]["id"]
self._data_source_cache[target_id] = ds_id
return ds_id
async def get_schema(self, target_id: str) -> dict:
"""Get full schema for a database or data source."""
data_source_id = await self._resolve_data_source_id(target_id)
return await self._request(
self.client.databases.retrieve(database_id=database_id)
self.client.data_sources.retrieve(data_source_id=data_source_id)
)
async def fetch_all_pages(self, database_id: str) -> list[dict]:
"""Fetch all pages from source database."""
async def fetch_all_pages(self, target_id: str) -> list[dict]:
"""Fetch all pages from source data source."""
data_source_id = await self._resolve_data_source_id(target_id)
pages = []
has_more = True
cursor = None
while has_more:
params = {"database_id": database_id, "page_size": 100}
params = {"data_source_id": data_source_id, "page_size": 100}
if cursor:
params["start_cursor"] = cursor
response = await self._request(self.client.databases.query(**params))
response = await self._request(
self.client.data_sources.query(**params)
)
pages.extend(response["results"])
has_more = response.get("has_more", False)
cursor = response.get("next_cursor")
@@ -235,7 +271,10 @@ class SchemaMigrator:
result = await self._request(
self.client.pages.create(
parent={"database_id": target_database_id},
parent={
"type": "data_source_id",
"data_source_id": target_database_id,
},
properties=properties,
)
)
@@ -253,22 +292,26 @@ class SchemaMigrator:
target_db: str,
mapping: dict,
) -> list[dict]:
"""Execute full migration."""
"""Execute full migration. Resolves both IDs to data_source_ids once."""
logger.info("Resolving source and target IDs...")
source_ds = await self._resolve_data_source_id(source_db)
target_ds = await self._resolve_data_source_id(target_db)
logger.info("Fetching schemas...")
source_schema = await self.get_schema(source_db)
target_schema = await self.get_schema(target_db)
source_schema = await self.get_schema(source_ds)
target_schema = await self.get_schema(target_ds)
logger.info(f"Source: {len(source_schema['properties'])} properties")
logger.info(f"Target: {len(target_schema['properties'])} properties")
logger.info("Fetching source pages...")
pages = await self.fetch_all_pages(source_db)
pages = await self.fetch_all_pages(source_ds)
logger.info(f"Found {len(pages)} pages to migrate")
results = []
for page in tqdm(pages, desc="Migrating"):
result = await self.migrate_page(
page, target_db, mapping, source_schema, target_schema
page, target_ds, mapping, source_schema, target_schema
)
results.append(result)