diff --git a/docs/src/content/docs/connectors/notion.mdx b/docs/src/content/docs/connectors/notion.mdx new file mode 100644 index 000000000..558172b1e --- /dev/null +++ b/docs/src/content/docs/connectors/notion.mdx @@ -0,0 +1,481 @@ +--- +title: "*Notion* connector" +toc_max_heading_level: 4 +description: > + Write rows to Notion databases (data sources) as a CocoIndex target. + Upserts and archives are automatic — declare your rows from a Python + class and CocoIndex keeps Notion in sync, including cleanup of rows + that fall out of the declared set. +--- + +The `notion` connector lets you treat a Notion database (called a *data source* in +Notion's 2025-09-03 API) as a first-class CocoIndex target. You declare a Python +class as the row shape, mount the target, and call `declare_row()` — CocoIndex +handles creates, updates, and archives across runs. + +```python +from cocoindex.connectors import notion +``` + +```bash +pip install cocoindex[notion] +``` + +The connector supports two modes: + +- **`managed_by="system"`** (default) — you give the connector a parent (page + or database) and a title; it finds or creates the data source for you, and + PATCH-adds new properties as your row class grows. Destructive changes + (existing column's type changed) are rejected unless you opt in with + `allow_destructive=True`. +- **`managed_by="user"`** — you point it at an existing data source + and the connector keeps individual rows in sync. The declared property + schema is validated against the live data source at mount, and the + connector refuses to write if anything's mismatched. + +## Connection setup + +### NotionClient + +A token-scoped Notion API client with built-in rate limiting (3 concurrent +requests by default, matching Notion's documented sustained limit) and bounded +exponential-backoff retry that honors `Retry-After` on 429s. + +```python +@dataclass +class NotionClient: + token: str + max_concurrency: int = 3 + session: aiohttp.ClientSession | None = None +``` + +**Parameters:** + +- `token` — Internal integration token from + [Notion's integrations page](https://www.notion.so/profile/integrations). +- `max_concurrency` — Cap on concurrent in-flight requests. Default `3`. Lower + if the integration is shared across multiple workloads in the same workspace. +- `session` — Optional `aiohttp.ClientSession` to reuse. If omitted, the + client creates and owns its own session. + +**Use as an async context manager** so the underlying HTTP session is closed +deterministically: + +```python +async with notion.NotionClient(token=os.environ["NOTION_TOKEN"]) as client: + ... +``` + +### Providing the client via ContextKey + +Create a `ContextKey[notion.NotionClient]` and provide it in your lifespan. +The key name is the stable identity CocoIndex uses to track managed rows +across runs. + +```python +import os +from typing import AsyncIterator +import cocoindex as coco +from cocoindex.connectors import notion + +NOTION = coco.ContextKey[notion.NotionClient]("notion_main") + +@coco.lifespan +async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]: + async with notion.NotionClient(token=os.environ["NOTION_TOKEN"]) as client: + builder.provide(NOTION, client) + yield +``` + +## As target + +The connector exposes a two-level target — a database (the data source) at the +top, and pages (rows) underneath. CocoIndex tracks every page it has written +and reconciles on every run: new rows POST, changed rows PATCH, missing rows +get archived (or hard-deleted, or left alone — your choice). + +### Declaring target states + +#### Database (parent state) + +```python +async def mount_database_target( + client: ContextKey[NotionClient], + data_source_id: str | None = None, + schema: DatabaseSchema[RowT] | None = None, + *, + managed_by: Literal["system", "user"] = "system", + parent_page_id: str | None = None, + parent_database_id: str | None = None, + title: str | None = None, + on_delete: OnDelete = OnDelete.ARCHIVE, + allow_destructive: bool = False, +) -> DatabaseTarget[RowT] +``` + +**Parameters:** + +- `client` — A `ContextKey[NotionClient]` identifying the client to use. +- `data_source_id` — Required when `managed_by="user"`. Notion data source ID + (find it via the database URL or `GET /v1/databases/{id}/data_sources`). +- `schema` — A `DatabaseSchema` describing the row class and its mapping onto + Notion properties. +- `managed_by` — `"system"` (default) creates and evolves the data source; + `"user"` reuses an existing one. +- `parent_page_id` / `parent_database_id` — Required in `system` mode, exactly + one of them. The data source is created (or found) under this parent. +- `title` — Required in `system` mode. The Notion display name for the data + source, also used as the stable identifier across runs. +- `on_delete` — Strategy for pages whose source row is no longer declared. +- `allow_destructive` — In `system` mode, opt in to changes that would lose + data (e.g. changing an existing property's type). Default `False` — the + connector refuses such changes at mount. + +**Returns:** A ready-to-use `DatabaseTarget` for declaring rows. + +In `user` mode the connector calls `GET /v1/data_sources/{id}` at mount and +fails loudly on missing properties or type mismatches. In `system` mode it +queries the parent for an existing data source with the matching title — if +none, it creates one via `POST /v1/databases` (page parent) or +`POST /v1/data_sources` (database parent). New properties declared in the +dataclass are added via `PATCH /v1/data_sources/{id}` on subsequent runs. +If `allow_destructive=True`, existing non-title property type mismatches are +also PATCHed to the declared type. Notion may reject unsupported conversions. +The title property is special: Notion requires exactly one title property and +does not allow its type to be added, removed, or changed via the API, so the +connector always raises for title-property mismatches instead of attempting an +automatic destructive change. + +Notion API references checked on 2026-06-01: + +- [Pagination](https://developers.notion.com/reference/intro) — list endpoints + such as block children return at most 100 items per page and continue with + `next_cursor` / `start_cursor`. +- [Rate limits](https://developers.notion.com/reference/request-limits) — + Notion returns `Retry-After` on `429` responses. +- [Update data source properties](https://developers.notion.com/reference/update-data-source-properties) + — data source property types are updated by PATCHing the new type schema; + title property type changes are not allowed. +- [Update a data source](https://developers.notion.com/reference/update-a-data-source) + — Notion describes property type conversion behavior and notes that some + conversions are unsupported. + +Lower-level entry points — `database_target()` returns a `TargetState` for +composition with `coco.mount_target()`, and `declare_database_target()` is the +synchronous declaration variant. Both have the same kwargs as +`mount_database_target`. + +#### Rows (child states) + +Once a `DatabaseTarget` is resolved, declare rows to upsert: + +```python +def DatabaseTarget.declare_row( + self, + *, + row: RowT, +) -> None +``` + +**Parameters:** + +- `row` — A dataclass / NamedTuple / dict / Pydantic model matching the + `DatabaseSchema`. Primary-key fields must be set. + +CocoIndex fingerprints each declared row; unchanged rows are a no-op (no +Notion writes). On rows whose primary key was declared on a prior run but is +absent this run, CocoIndex emits an archive (default) — see +[Delete strategies](#delete-strategies). + +### Database schema + +#### From a Python class + +Bind a Python row class to Notion properties via `Annotated[T, PropType]` +metadata, or via the `property_map` override. + +```python +@classmethod +async def DatabaseSchema.from_class( + cls, + record_type: type[RowT], + primary_key: list[str], + *, + property_map: dict[str, PropType] | None = None, +) -> DatabaseSchema[RowT] +``` + +**Parameters:** + +- `record_type` — The row class (dataclass / NamedTuple / Pydantic model). +- `primary_key` — Field names that uniquely identify a row. Notion has no + native composite-key constraint, so the connector uses `AND`-of-filters to + resolve the row on every cache miss. +- `property_map` — Optional per-field override mapping `field_name → PropType`. + Useful when you don't want to (or can't) put `Annotated[...]` on the field. + +**Example:** + +```python +from typing_extensions import Annotated +from dataclasses import dataclass + +@dataclass(frozen=True) +class Person: + name: Annotated[str, notion.TitleProp("Name")] + email: Annotated[str, notion.EmailProp("Email")] + role: Annotated[str, notion.SelectProp("Role")] + active: Annotated[bool, notion.CheckboxProp("Active")] + +schema = await notion.DatabaseSchema.from_class(Person, primary_key=["name"]) +``` + +Or with explicit `property_map`: + +```python +schema = await notion.DatabaseSchema.from_class( + Person, + primary_key=["name"], + property_map={ + "name": notion.TitleProp("Name"), + "email": notion.EmailProp("Email"), + "role": notion.SelectProp("Role"), + "active": notion.CheckboxProp("Active"), + }, +) +``` + +Both forms validate that at least one field has a binding, that no two fields +map to the same Notion property name, that there's at most one `TitleProp`, and +that every primary-key field has a binding. Fields without a binding are +ignored, which lets the Python class carry transient state that should not be +written to Notion. Typos in `property_map` keys (i.e. keys that don't name a +real field) raise loudly. + +### Property types + +The connector supports the property types you'd reach for first: + +| `PropType` | Notion type | Python type | Usable as PK? | +|-----------|-------------|-------------|---------------| +| `TitleProp` | `title` | `str` | yes | +| `RichTextProp` | `rich_text` | `str` | yes | +| `NumberProp` | `number` | `int` / `float` | yes | +| `UrlProp` | `url` | `str` | yes | +| `EmailProp` | `email` | `str` | yes | +| `SelectProp` | `select` | `str` | yes | +| `MultiSelectProp` | `multi_select` | `list[str]` | no | +| `DateProp` | `date` | `datetime` / `date` / ISO string | yes | +| `CheckboxProp` | `checkbox` | `bool` | yes | +| `RelationProp` | `relation` | page ID or `list[str]` | no | + +Each `PropType` takes the Notion property's display name as its only positional +argument: `notion.SelectProp("Role")` binds to the property literally named +"Role" in your Notion data source. + +`RelationProp` writes Notion relation values as page IDs from the related data +source. It is intentionally page-ID based: it does not resolve related rows by +CocoIndex primary key, coordinate writes across targets, or create related pages +for you. Relation values follow Notion's +[page property value shape](https://developers.notion.com/reference/page-property-values) +(checked 2026-06-01): page references are written as +`{"relation": [{"id": "..."}]}`. The connector does not look up or create +related pages for you. + +For `managed_by="user"`, the relation column must already exist in Notion and +the connector validates that its Notion type is `relation`. For +`managed_by="system"`, pass `target_data_source_id` so CocoIndex can create or +add the relation column. Notion's +[data-source property docs](https://developers.notion.com/reference/property-object) +(checked 2026-06-01) define relation schema with `data_source_id`, and the +related data source must be shared with the integration. + +```python +@dataclass +class TaskRow: + name: Annotated[str, notion.TitleProp("Name")] + project_page_ids: Annotated[ + list[str], + notion.RelationProp("Projects", target_data_source_id=PROJECTS_DS_ID), + ] +``` + +People, files, formula, status, rollup, created_time / last_edited_time, and +"place" properties are not yet supported. + +### Delete strategies + +When a row's primary key was declared on a prior run but isn't declared this +run, the row handler emits one of three actions based on `on_delete`: + +| `OnDelete` | Notion call | When to use | +|---|---|---| +| `ARCHIVE` (default) | `PATCH /v1/pages/{id}` with `{"archived": true}` | Recoverable. Matches what Notion users expect when they hit "Delete" in the UI. | +| `HARD` | `DELETE /v1/blocks/{id}` | Moves the page to Notion's trash (recoverable for 30 days). Use if you don't want archived rows showing up in `archived: true` queries. | +| `IGNORE` | (nothing) | Leave the page alone. CocoIndex drops the tracking record so future runs treat the row as new. Useful for additive-only workflows. | + +```python +target = await notion.mount_database_target( + NOTION, + data_source_id, + schema, + managed_by="user", + on_delete=notion.OnDelete.HARD, +) +``` + +### Page-id persistence + +Notion assigns each page an ID on `POST`, so a fresh process doesn't know +the page IDs of rows it inserted on a previous run. The connector resolves +this lazily: when the row handler needs a page ID and doesn't have one cached, +it issues `POST /v1/data_sources/{id}/query` with a filter on the primary key. +The first hit gets cached for the rest of the run; subsequent actions on the +same row are O(1) lookups. + +Trade-off: heavier per-miss cost than pre-fetching the entire data source up +front, but much cheaper on the common case where CocoIndex's tracking records +short-circuit reconcile and the sink only sees the rows that genuinely +changed. + +## Setup heads-ups + +Documented here so the gotchas we hit while building the connector don't +trip anyone else up. None of these are CocoIndex-specific — they're how Notion +integrations work — but they're easy to miss. + +### 1. Every parent in the path must be shared with the integration + +Sharing a *database* with your integration is not enough on its own. Notion +also checks the parent *page* — if the integration isn't connected to the +page that contains your database, the API returns `object_not_found` with +the misleading message *"Make sure the relevant pages and databases are shared +with your integration"*. + +Fix: open the parent page in Notion, top-right `···` → **Connections** → +**`+ Add connections`** → search your integration → confirm. + +If you don't see this entry, double-check you're looking at the **page**, +not the *database* itself. Linked-view arrows (`↗`) jump to the source where +the Connections menu lives. + +### 2. Internal integrations can't create workspace-level databases + +`POST /v1/databases` with `parent.type = "workspace"` returns +`validation_error` for internal (token-based) integrations. Pass +`parent.page_id` pointing at a page the integration can see. This matters for +`managed_by="system"` (the default), which creates databases — the +user-managed flow sidesteps it since it never creates anything. + +### 3. Sharing is per-integration + +A workspace can have multiple integrations. Sharing a page with the +`cocoindex` integration does not share it with `cocoindex gtm` (or any +other). At mount the connector calls `GET /v1/data_sources/{id}` and fails +loudly if the integration can't see the data source — that's the friendlier +error you'll usually hit first. + +### 4. Notion-Version pinning + +The connector pins `Notion-Version: 2025-09-03`, the first version that +exposes *data sources* as first-class objects distinct from *databases*. +Rolling back to an older API version will break the `data_source_id` routing. + +## Complete example + +```python +import os +import pathlib +from dataclasses import dataclass +from typing import AsyncIterator +from typing_extensions import Annotated + +import cocoindex as coco +from cocoindex.connectors import notion + +NOTION = coco.ContextKey[notion.NotionClient]("notion_main") + + +@dataclass(frozen=True) +class Person: + name: Annotated[str, notion.TitleProp("Name")] + email: Annotated[str, notion.EmailProp("Email")] + role: Annotated[str, notion.SelectProp("Role")] + active: Annotated[bool, notion.CheckboxProp("Active")] + + +PEOPLE: list[Person] = [ + Person(name="Ada Lovelace", email="ada@example.com", + role="Engineer", active=True), + Person(name="Grace Hopper", email="grace@example.com", + role="Engineer", active=True), + Person(name="Alan Turing", email="alan@example.com", + role="Researcher", active=False), +] + + +@coco.lifespan +async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]: + builder.settings.db_path = pathlib.Path("./cocoindex.db") + async with notion.NotionClient(token=os.environ["NOTION_TOKEN"]) as client: + builder.provide(NOTION, client) + yield + + +@coco.fn +async def app_main() -> None: + # managed_by="system" is the default — CocoIndex creates the "People" + # database under the parent page on first run and evolves it as Person grows. + target = await notion.mount_database_target( + NOTION, + schema=await notion.DatabaseSchema.from_class(Person, primary_key=["name"]), + parent_page_id=os.environ["NOTION_PARENT_PAGE"], + title="People", + ) + for person in PEOPLE: + target.declare_row(row=person) + + +app = coco.App( + coco.AppConfig(name="NotionTargetBasics"), + app_main, +) +``` + +Run it: + +```bash +export NOTION_TOKEN=ntn_... +export NOTION_PARENT_PAGE= +cocoindex update main.py:NotionTargetBasics +``` + +On the first run the connector creates a database titled "People" under the +parent page. Edit `PEOPLE` (add, remove, change) and re-run — CocoIndex syncs +the diff, and any row you delete from the list gets archived in Notion on the +next run. + +### User-managed variant + +To point the connector at an existing data source instead of creating one, +pass `managed_by="user"` with its `data_source_id`: + +```python +target = await notion.mount_database_target( + NOTION, + os.environ["NOTION_DATA_SOURCE_ID"], + schema=await notion.DatabaseSchema.from_class(Person, primary_key=["name"]), + managed_by="user", +) +``` + +The declared property schema is validated against the live data source at +mount; the connector refuses to write if anything's mismatched. + +On the first run the connector creates a database titled "People" under the +given page. On later runs, adding a new field with an `Annotated[T, +notion.SomeProp("Foo")]` annotation triggers `PATCH /v1/data_sources/{id}` to +add the new "Foo" property. Existing-property type changes are rejected +unless you pass `allow_destructive=True`. + +Full source: [`examples/notion_target_basics`](https://github.com/cocoindex-io/cocoindex/tree/main/examples/notion_target_basics). diff --git a/docs/src/data/docs-sidebar.ts b/docs/src/data/docs-sidebar.ts index 7e87bc833..b93e563e4 100644 --- a/docs/src/data/docs-sidebar.ts +++ b/docs/src/data/docs-sidebar.ts @@ -66,6 +66,7 @@ export const sidebar: SidebarItem[] = [ { type: 'doc', slug: 'connectors/lancedb', label: 'LanceDB' }, { type: 'doc', slug: 'connectors/localfs', label: 'Local filesystem' }, { type: 'doc', slug: 'connectors/neo4j', label: 'Neo4j' }, + { type: 'doc', slug: 'connectors/notion', label: 'Notion' }, { type: 'doc', slug: 'connectors/oci_object_storage', label: 'OCI Object Storage' }, { type: 'doc', slug: 'connectors/postgres', label: 'Postgres' }, { type: 'doc', slug: 'connectors/qdrant', label: 'Qdrant' }, diff --git a/examples/notion_target_basics/README.md b/examples/notion_target_basics/README.md new file mode 100644 index 000000000..edbb63b1f --- /dev/null +++ b/examples/notion_target_basics/README.md @@ -0,0 +1,96 @@ +# Notion target — basics + +Minimal example for the cocoindex Notion target connector. Declares three rows +of `Person` data; cocoindex creates the Notion database for you +(`managed_by="system"`, the default) and syncs the rows — creating new pages, +patching changed ones, and archiving pages whose source row goes away. + +## What it shows + +| Behavior | How to trigger | What you see in Notion | +|---|---|---| +| Initial insert | First `cocoindex update` | Three pages created | +| Idempotent re-run | Re-run with no changes | ~0.2s, zero Notion writes | +| Field update | Change a field value on one row | That page PATCHed | +| Automatic archive | Remove a row from `PEOPLE` and re-run | That page archived in Notion | +| Re-add | Add the row back | New page (or revived) | + +The archive step is the key win over hand-rolled `notion-client` plumbing — +cocoindex tracks what it wrote on the previous run and reconciles it against +what's declared this run. + +## Setup + +1. **Create (or pick) a Notion page** to hold the database. CocoIndex creates + the database — titled `People`, with the `Name` / `Email` / `Role` / + `Active` properties this example declares — under it on the first run. + + The sandbox page used while developing this example looked like this: + + ![Sandbox parent page](https://cocoindex.io/blobs/docs/img/examples/notion_target_basics/test-sandbox-page.png) + +2. **Share the page** with your Notion integration: top-right `···` → + Connections → `+ Add connections` → select your integration. Every parent + page in the path must be shared — Notion checks access at the page level. + +3. **Grab the page ID** from the page URL (the 32-char hex after the title). + +4. **Export tokens** and run: + + ```sh + export NOTION_TOKEN=ntn_... + export NOTION_PARENT_PAGE= + cocoindex update main.py:NotionTargetBasics + ``` + +After the first run, the database fills up: + +![Demo database after first run, with Alan archived](https://cocoindex.io/blobs/docs/img/examples/notion_target_basics/demo-database-after-archive.png) + +(Pictured: after the `Alan Turing` row was removed from `PEOPLE` and the example +was re-run — his page was automatically archived, leaving only Ada and Grace.) + +## Try the lifecycle + +Edit `main.py`'s `PEOPLE` list and re-run `cocoindex update` after each change: + +```python +# 1. Add a new row -> CocoIndex creates a new page +Person(name="Margaret Hamilton", email="margaret@example.com", + role="Engineer", active=True), + +# 2. Change a value -> CocoIndex PATCHes that page +Person(name="Ada Lovelace", email="ada@new.example.com", ...), + +# 3. Remove a row -> CocoIndex archives that page +# (delete the line) +``` + +## Switch the delete behavior + +Pass `on_delete=...` to change what happens when a row is removed: + +```python +target = await notion.mount_database_target( + notion_client, + schema=schema, + parent_page_id=os.environ["NOTION_PARENT_PAGE"], + title="People", + on_delete=notion.OnDelete.HARD, # send page to trash + # on_delete=notion.OnDelete.IGNORE # leave page alone +) +``` + +Default is `OnDelete.ARCHIVE` — reversible, matches what Notion users expect. + +## Image files to add (developer note) + +The two screenshots referenced above live in the `cocoindex-io/blobs` repo: + +| README link → file path (in blobs repo) | +|---| +| `public/docs/img/examples/notion_target_basics/test-sandbox-page.png` | +| `public/docs/img/examples/notion_target_basics/demo-database-after-archive.png` | + +Drop the images at those paths, then `git add . && git commit && git push` from +the blobs repo — the cocoindex.io Pages workflow will publish them. diff --git a/examples/notion_target_basics/main.py b/examples/notion_target_basics/main.py new file mode 100644 index 000000000..29e377de7 --- /dev/null +++ b/examples/notion_target_basics/main.py @@ -0,0 +1,86 @@ +"""Minimal example for the cocoindex Notion target connector. + +CocoIndex manages the database for you (``managed_by="system"``, the default): +on the first run it creates a database titled "People" under the parent page +you point it at, then keeps the rows in sync on every run. Modify the +``PEOPLE`` list below to see how reconciliation works: + +- Edit a row's value -> CocoIndex PATCHes the corresponding Notion page. +- Remove a row -> CocoIndex archives the page (or hard-deletes, or leaves + it untouched, depending on the ``on_delete`` strategy). +- Add a row -> CocoIndex creates a new page. +- Add a field to ``Person`` -> CocoIndex adds the property to the database. + +Setup +----- +1. Create (or pick) a Notion page to hold the database. +2. Share it with your integration (top-right ··· -> Connections). Every parent + page in the path must be shared, not just the database. +3. Export ``NOTION_TOKEN`` and ``NOTION_PARENT_PAGE`` (the page's ID, from its + URL). +4. ``cocoindex update main.py:NotionTargetBasics`` + +To point at an existing database instead of creating one, pass +``managed_by="user"`` with a ``data_source_id`` — see the connector docs. +""" + +import os +import pathlib +from dataclasses import dataclass +from typing import AsyncIterator + +from typing_extensions import Annotated + +import cocoindex as coco +from cocoindex.connectors import notion + +notion_client = coco.ContextKey[notion.NotionClient]("notion") + + +@dataclass(frozen=True) +class Person: + """One row in the Notion database.""" + + name: Annotated[str, notion.TitleProp("Name")] + email: Annotated[str, notion.EmailProp("Email")] + role: Annotated[str, notion.SelectProp("Role")] + active: Annotated[bool, notion.CheckboxProp("Active")] + + +PEOPLE: list[Person] = [ + Person(name="Ada Lovelace", email="ada@example.com", role="Engineer", active=True), + Person( + name="Grace Hopper", email="grace@example.com", role="Engineer", active=True + ), + Person( + name="Alan Turing", email="alan@example.com", role="Researcher", active=False + ), +] + + +@coco.lifespan +async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]: + builder.settings.db_path = pathlib.Path("./cocoindex.db") + async with notion.NotionClient(token=os.environ["NOTION_TOKEN"]) as client: + builder.provide(notion_client, client) + yield + + +@coco.fn +async def app_main() -> None: + # managed_by="system" is the default — CocoIndex creates the "People" + # database under the parent page on first run and evolves it as Person grows. + target = await notion.mount_database_target( + notion_client, + schema=await notion.DatabaseSchema.from_class(Person, primary_key=["name"]), + parent_page_id=os.environ["NOTION_PARENT_PAGE"], + title="People", + ) + for person in PEOPLE: + target.declare_row(row=person) + + +app = coco.App( + coco.AppConfig(name="NotionTargetBasics"), + app_main, +) diff --git a/examples/notion_target_basics/pyproject.toml b/examples/notion_target_basics/pyproject.toml new file mode 100644 index 000000000..9315f37e8 --- /dev/null +++ b/examples/notion_target_basics/pyproject.toml @@ -0,0 +1,9 @@ +[project] +name = "notion-target-hello-world" +version = "0.1.0" +description = "Hello-world for the cocoindex Notion target connector." +requires-python = ">=3.11" +dependencies = ["cocoindex[notion]>=1.0.7"] + +[tool.setuptools] +packages = [] diff --git a/pyproject.toml b/pyproject.toml index b75ff7ad9..3243c23b2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -90,6 +90,7 @@ falkordb = ["falkordb>=1.1.0"] neo4j = ["neo4j>=5.18.0"] kafka = ["confluent_kafka>=2.6"] iggy = ["apache-iggy>=0.8.0"] +notion = ["aiohttp>=3.9.0"] oci = ["oci>=2.0"] entity_resolution = ["faiss-cpu>=1.7"] entity_resolution_llm = [ diff --git a/python/cocoindex/connectors/notion/__init__.py b/python/cocoindex/connectors/notion/__init__.py new file mode 100644 index 000000000..b3c187694 --- /dev/null +++ b/python/cocoindex/connectors/notion/__init__.py @@ -0,0 +1,81 @@ +"""Notion target connector for CocoIndex. + +Declarative target for Notion databases (data sources) with automatic +upsert + delete semantics, mirroring ``connectors.postgres``. + +Quick start:: + + from cocoindex.connectors import notion + + notion_client = coco.ContextKey[notion.NotionClient]("notion_main") + + @coco.lifespan + async def lifespan(builder): + async with notion.NotionClient(token=os.environ["NOTION_TOKEN"]) as c: + builder.provide(notion_client, c) + yield + + @dataclass(frozen=True) + class AccountRow: + domain: Annotated[str, notion.UrlProp("Domain")] + name: Annotated[str, notion.TitleProp("Name")] + + @coco.fn + async def app_main(): + accounts = await notion.mount_database_target( + notion_client, + schema=await notion.DatabaseSchema.from_class( + AccountRow, primary_key=["domain"] + ), + parent_page_id=os.environ["NOTION_PARENT_PAGE_ID"], + title="Accounts", + ) + accounts.declare_row(row=AccountRow(domain="anthropic.com", name="Anthropic")) +""" + +from ._client import NotionClient, NOTION_API_VERSION +from ._target import ( + DatabaseTarget, + ManagedBy, + OnDelete, + database_target, + declare_database_target, + mount_database_target, +) +from ._types import ( + CheckboxProp, + DatabaseSchema, + DateProp, + EmailProp, + MultiSelectProp, + NumberProp, + PropType, + RelationProp, + RichTextProp, + SelectProp, + TitleProp, + UrlProp, +) + +__all__ = [ + "CheckboxProp", + "DatabaseSchema", + "DatabaseTarget", + "DateProp", + "ManagedBy", + "EmailProp", + "MultiSelectProp", + "NOTION_API_VERSION", + "NotionClient", + "NumberProp", + "OnDelete", + "PropType", + "RelationProp", + "RichTextProp", + "SelectProp", + "TitleProp", + "UrlProp", + "database_target", + "declare_database_target", + "mount_database_target", +] diff --git a/python/cocoindex/connectors/notion/_client.py b/python/cocoindex/connectors/notion/_client.py new file mode 100644 index 000000000..c4840d1c7 --- /dev/null +++ b/python/cocoindex/connectors/notion/_client.py @@ -0,0 +1,224 @@ +"""Notion HTTP client used by the target connector. + +Thin wrapper around the Notion REST API (version 2025-09-03) with the +concurrency + retry behavior required by CocoIndex's high-fanout sinks: + +- shared per-client asyncio.Semaphore caps concurrent calls +- honors Retry-After on 429s +- bounded exponential backoff +""" + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass, field +from typing import Any, AsyncIterator + +import aiohttp + +NOTION_API_VERSION = "2025-09-03" +NOTION_BASE_URL = "https://api.notion.com/v1" + +# Manual retry: 10 attempts with exponential backoff capped at 60s. Implemented +# inline (rather than via tenacity) so mypy can fully type the decorated method +# without an untyped-decorator escape hatch. +_MAX_ATTEMPTS = 10 +_MIN_WAIT = 2.0 +_MAX_WAIT = 60.0 + + +@dataclass +class NotionClient: + """A token-scoped Notion API client with built-in rate limiting. + + Use as an async context manager so the underlying aiohttp session is + closed deterministically:: + + async with NotionClient(token=...) as client: + ... + + Or pass an explicit ``session`` if you want to reuse one across clients. + + ``max_concurrency`` defaults to 3, matching Notion's documented sustained + rate limit; tune lower if the integration is shared with other workloads. + """ + + token: str + max_concurrency: int = 3 + session: aiohttp.ClientSession | None = None + _owns_session: bool = field(default=False, init=False) + _sem: asyncio.Semaphore = field(init=False) + + def __post_init__(self) -> None: + self._sem = asyncio.Semaphore(self.max_concurrency) + + async def __aenter__(self) -> "NotionClient": + if self.session is None: + self.session = aiohttp.ClientSession() + self._owns_session = True + return self + + async def __aexit__(self, *exc: Any) -> None: + await self.close() + + async def close(self) -> None: + if self._owns_session and self.session is not None: + await self.session.close() + self.session = None + self._owns_session = False + + def _headers(self) -> dict[str, str]: + return { + "Authorization": f"Bearer {self.token}", + "Notion-Version": NOTION_API_VERSION, + "Content-Type": "application/json", + } + + def _require_session(self) -> aiohttp.ClientSession: + if self.session is None: + raise RuntimeError( + "NotionClient session not started; use 'async with NotionClient(...)' " + "or assign a session explicitly." + ) + return self.session + + async def _request( + self, method: str, path: str, json_body: dict[str, Any] | None = None + ) -> dict[str, Any]: + session = self._require_session() + url = f"{NOTION_BASE_URL}{path}" + last_error: Exception | None = None + for attempt in range(_MAX_ATTEMPTS): + retry_after: float | None = None + async with self._sem: + async with session.request( + method, url, json=json_body, headers=self._headers() + ) as r: + if r.status == 429: + # Notion rate-limit docs, checked 2026-06-01: + # Retry-After is an integer number of seconds. Sleep + # after releasing the semaphore so unrelated requests + # can still use the connection's concurrency budget. + retry_after = float(r.headers.get("Retry-After", "1")) + last_error = RuntimeError("notion rate_limited") + else: + r.raise_for_status() + result: dict[str, Any] = await r.json() + return result + if retry_after is not None: + await asyncio.sleep(retry_after) + else: + # Exponential backoff between attempts, capped at _MAX_WAIT. + await asyncio.sleep(min(_MIN_WAIT * (2**attempt), _MAX_WAIT)) + raise RuntimeError( + f"Notion request {method} {path} failed after {_MAX_ATTEMPTS} attempts" + ) from last_error + + async def get_data_source(self, data_source_id: str) -> dict[str, Any]: + return await self._request("GET", f"/data_sources/{data_source_id}") + + async def get_database(self, database_id: str) -> dict[str, Any]: + return await self._request("GET", f"/databases/{database_id}") + + async def create_database( + self, + *, + parent_page_id: str, + title: str, + properties: dict[str, dict[str, Any]], + ) -> dict[str, Any]: + """POST /v1/databases — create a new database under a parent page, + with one initial data source. + + Returns the full database object, including ``data_sources`` — the + new data source's id is ``result["data_sources"][0]["id"]``. + """ + body: dict[str, Any] = { + "parent": {"type": "page_id", "page_id": parent_page_id}, + "title": [{"type": "text", "text": {"content": title}}], + "initial_data_source": {"properties": properties}, + } + return await self._request("POST", "/databases", body) + + async def create_data_source( + self, + *, + parent_database_id: str, + title: str, + properties: dict[str, dict[str, Any]], + ) -> dict[str, Any]: + """POST /v1/data_sources — add a new data source to an existing + database. + """ + body: dict[str, Any] = { + "parent": {"type": "database_id", "database_id": parent_database_id}, + "title": [{"type": "text", "text": {"content": title}}], + "properties": properties, + } + return await self._request("POST", "/data_sources", body) + + async def update_data_source_properties( + self, + data_source_id: str, + properties: dict[str, dict[str, Any] | None], + ) -> dict[str, Any]: + """PATCH /v1/data_sources/{id} — add / rename / change-type / remove + properties. To remove, pass ``None`` as the value for that name. + """ + return await self._request( + "PATCH", f"/data_sources/{data_source_id}", {"properties": properties} + ) + + async def query_data_source( + self, + data_source_id: str, + *, + filter: dict[str, Any] | None = None, + page_size: int = 100, + start_cursor: str | None = None, + ) -> dict[str, Any]: + body: dict[str, Any] = {"page_size": page_size} + if filter is not None: + body["filter"] = filter + if start_cursor is not None: + body["start_cursor"] = start_cursor + return await self._request( + "POST", f"/data_sources/{data_source_id}/query", body + ) + + async def query_all( + self, data_source_id: str, *, filter: dict[str, Any] | None = None + ) -> AsyncIterator[dict[str, Any]]: + cursor: str | None = None + while True: + res = await self.query_data_source( + data_source_id, filter=filter, start_cursor=cursor + ) + for page in res.get("results", []): + yield page + if not res.get("has_more"): + return + cursor = res.get("next_cursor") + + async def create_page( + self, data_source_id: str, properties: dict[str, Any] + ) -> dict[str, Any]: + body: dict[str, Any] = { + "parent": {"data_source_id": data_source_id}, + "properties": properties, + } + return await self._request("POST", "/pages", body) + + async def update_page_properties( + self, page_id: str, properties: dict[str, Any] + ) -> dict[str, Any]: + return await self._request( + "PATCH", f"/pages/{page_id}", {"properties": properties} + ) + + async def archive_page(self, page_id: str) -> dict[str, Any]: + return await self._request("PATCH", f"/pages/{page_id}", {"archived": True}) + + async def delete_page(self, page_id: str) -> dict[str, Any]: + # Notion treats DELETE on a block (a page IS a block) as a trash operation. + return await self._request("DELETE", f"/blocks/{page_id}") diff --git a/python/cocoindex/connectors/notion/_target.py b/python/cocoindex/connectors/notion/_target.py new file mode 100644 index 000000000..3fb5fe542 --- /dev/null +++ b/python/cocoindex/connectors/notion/_target.py @@ -0,0 +1,792 @@ +"""Notion database target. + +Two-level structure that mirrors ``connectors.postgres`` / +``connectors.sqlite``: + +- ``_DatabaseHandler`` (root) owns the data source. Supports two modes: + + - ``managed_by="system"`` (default): the connector looks under the parent + (page or database) for a data source with the given title; creates it on + first run if missing, and PATCH-adds new properties on subsequent runs + when the dataclass grows. Destructive schema changes are rejected unless + ``allow_destructive=True``. + - ``managed_by="user"``: the data source must already exist; the connector + validates that the declared property schema matches the live data source + on every apply. + +- ``_RowHandler`` (child) reconciles individual pages against their + fingerprints and applies upsert / archive actions. + +Page-id persistence uses the query-on-miss approach: when an action +needs a page_id and the in-memory cache doesn't have one, the row +handler queries Notion with a primary-key filter (one HTTP call per +unique missing PK). New rows POST and cache the returned page_id; +unchanged rows never need a query because cocoindex's tracking +records short-circuit reconcile before the sink runs. Trade-off: +heavier per-miss cost than pre-fetching the whole data source, but +much cheaper than pre-fetch on the common case where most signals +are no-ops between runs. +""" + +from __future__ import annotations + +import asyncio +from collections.abc import Collection, Sequence +from dataclasses import dataclass +from datetime import date +from enum import Enum +from typing import Any, Generic, Literal, NamedTuple +from urllib.parse import urlencode + +from typing_extensions import TypeVar + +import cocoindex as coco +from cocoindex.connectorkits.fingerprint import fingerprint_object +from cocoindex._internal.context_keys import ContextKey, ContextProvider + +from ._client import NotionClient +from ._types import DatabaseSchema, PropType + +RowT = TypeVar("RowT") + + +# --------------------------------------------------------------------------- +# Row-level +# --------------------------------------------------------------------------- + +_PageKey = tuple[Any, ...] +_PageValue = dict[str, Any] # Python field name -> Python value +_PageFingerprint = bytes + + +def _date_filter_value(value: Any) -> str: + if isinstance(value, date): + return value.isoformat() + return str(value) + + +class OnDelete(Enum): + """What to do with a Notion page whose source row is no longer declared.""" + + ARCHIVE = "archive" # PATCH archived=true — reversible (default). + HARD = "hard" # DELETE /blocks/{id} — moves to trash, recoverable for 30d. + IGNORE = "ignore" # Leave the page alone; tracking record drops only. + + +class _PageAction(NamedTuple): + """Action on a single page: upsert (value!=None) or delete (value=None).""" + + key: _PageKey + value: _PageValue | None + + +def _property_filter(prop: PropType, value: Any) -> dict[str, Any]: + """Build a Notion data-source query filter for ``prop == value``. + + Each Notion property type wants a different filter shape; this dispatches + on the property's ``notion_type``. Only the property types usable as + primary keys need to be listed here — title (almost always), rich_text, + url, email, number, date, checkbox, select. + """ + name = prop.name + nt = prop.notion_type + if nt == "title": + return {"property": name, "title": {"equals": str(value)}} + if nt == "rich_text": + return {"property": name, "rich_text": {"equals": str(value)}} + if nt == "url": + return {"property": name, "url": {"equals": str(value)}} + if nt == "email": + return {"property": name, "email": {"equals": str(value)}} + if nt == "number": + return {"property": name, "number": {"equals": float(value)}} + if nt == "checkbox": + return {"property": name, "checkbox": {"equals": bool(value)}} + if nt == "date": + return {"property": name, "date": {"equals": _date_filter_value(value)}} + if nt == "select": + return {"property": name, "select": {"equals": str(value)}} + raise ValueError( + f"Notion property type {nt!r} (prop {name!r}) is not supported as a " + "primary key. Use title / rich_text / url / email / number / date / " + "checkbox / select instead." + ) + + +class _RowHandler(coco.TargetHandler[_PageValue, _PageFingerprint]): + """Handler for each page within a Notion data source.""" + + def __init__( + self, + client: NotionClient, + data_source_id: str, + schema: DatabaseSchema[Any], + on_delete: OnDelete, + ) -> None: + self._client = client + self._data_source_id = data_source_id + self._schema = schema + self._on_delete = on_delete + self._sink = coco.TargetActionSink[_PageAction, None].from_async_fn( + self._apply_actions + ) + # In-memory cache. Values are page_id strings; sentinel _ABSENT marks + # PKs we've confirmed don't exist in Notion (avoids re-querying). + self._page_id_cache: dict[_PageKey, str] = {} + # One lock per PK so concurrent actions for the same key serialize their + # lookup-or-create, but actions for different keys still run in parallel. + self._key_locks: dict[_PageKey, asyncio.Lock] = {} + self._key_locks_lock = asyncio.Lock() + + async def _lock_for(self, key: _PageKey) -> asyncio.Lock: + async with self._key_locks_lock: + lock = self._key_locks.get(key) + if lock is None: + lock = asyncio.Lock() + self._key_locks[key] = lock + return lock + + async def _resolve_page_id(self, key: _PageKey) -> str | None: + """Return the page_id for ``key``, querying Notion on cache miss. + + Returns ``None`` if no page with this primary key exists in the data + source. The caller decides whether to POST a new page in that case. + """ + cached = self._page_id_cache.get(key) + if cached is not None: + return cached + pk_filter = self._build_pk_filter(key) + res = await self._client.query_data_source( + self._data_source_id, filter=pk_filter, page_size=1 + ) + results = res.get("results", []) + if not results: + return None + page_id: str = results[0]["id"] + self._page_id_cache[key] = page_id + return page_id + + def _build_pk_filter(self, key: _PageKey) -> dict[str, Any]: + pk_fields = self._schema.primary_key + prop_by_field = self._schema.properties_by_field + terms = [ + _property_filter(prop_by_field[field], key[i]) + for i, field in enumerate(pk_fields) + ] + return terms[0] if len(terms) == 1 else {"and": terms} + + async def _apply_actions( + self, + context_provider: ContextProvider, + actions: Sequence[_PageAction], + ) -> None: + if not actions: + return + + async def handle(action: _PageAction) -> None: + lock = await self._lock_for(action.key) + async with lock: + existing_id = await self._resolve_page_id(action.key) + + if action.value is None: + if existing_id is None: + return + if self._on_delete is OnDelete.IGNORE: + return + if self._on_delete is OnDelete.HARD: + await self._client.delete_page(existing_id) + else: + await self._client.archive_page(existing_id) + self._page_id_cache.pop(action.key, None) + return + + properties = self._schema.encode_row(action.value) + if existing_id: + await self._client.update_page_properties(existing_id, properties) + else: + created = await self._client.create_page( + self._data_source_id, properties + ) + self._page_id_cache[action.key] = created["id"] + + await asyncio.gather(*(handle(a) for a in actions)) + + def reconcile( + self, + key: coco.StableKey, + desired_state: _PageValue | coco.NonExistenceType, + prev_possible_records: Collection[_PageFingerprint], + prev_may_be_missing: bool, + /, + ) -> coco.TargetReconcileOutput[_PageAction, _PageFingerprint] | None: + # Coerce StableKey -> _PageKey + if not isinstance(key, tuple): + page_key: _PageKey = (key,) + else: + page_key = key + + if coco.is_non_existence(desired_state): + if not prev_possible_records and not prev_may_be_missing: + return None + return coco.TargetReconcileOutput( + action=_PageAction(key=page_key, value=None), + sink=self._sink, + tracking_record=coco.NON_EXISTENCE, + ) + + target_fp = fingerprint_object(desired_state) + if ( + not prev_may_be_missing + and prev_possible_records + and all(prev == target_fp for prev in prev_possible_records) + ): + return None + + return coco.TargetReconcileOutput( + action=_PageAction(key=page_key, value=desired_state), + sink=self._sink, + tracking_record=target_fp, + ) + + +# --------------------------------------------------------------------------- +# Database-level (parent) +# --------------------------------------------------------------------------- + + +ManagedBy = Literal["system", "user"] + + +@dataclass(frozen=True) +class _DatabaseSpec: + """User-declared spec for a Notion database (data source) target. + + In ``user`` mode, ``data_source_id`` identifies a pre-existing data source. + In ``system`` mode, the connector creates (and additively evolves) the data + source under ``parent_page_id`` or ``parent_database_id`` with ``title``. + """ + + schema: DatabaseSchema[Any] + on_delete: OnDelete = OnDelete.ARCHIVE + managed_by: ManagedBy = "system" + # user mode: + data_source_id: str | None = None + # system mode: + parent_page_id: str | None = None + parent_database_id: str | None = None + title: str | None = None + allow_destructive: bool = False + + +class _DatabaseKey(NamedTuple): + """Stable identity for a database target across runs. + + For user mode: ``identity`` is the data_source_id. + For system mode: ``identity`` is ``"system::"`` — + derived from the (immutable, user-supplied) inputs so cocoindex tracking + records survive even before the data source has been created. + """ + + client_key: str # ContextKey.key for the NotionClient + identity: str + + +class _DatabaseTracking(NamedTuple): + """Persisted tracking record for a database target.""" + + identity: str + + +class _DatabaseAction(NamedTuple): + """Parent-level action: ensure (spec set) or teardown (spec=NON_EXISTENCE).""" + + key: _DatabaseKey + spec: _DatabaseSpec | coco.NonExistenceType + + +def _system_identity(spec: _DatabaseSpec) -> str: + """Stable string identity for a system-managed target.""" + parent = spec.parent_page_id or spec.parent_database_id or "" + return f"system:{parent}:{spec.title or ''}" + + +def _read_notion_title(title_array: list[dict[str, Any]] | None) -> str: + """Extract plain text from a Notion ``title`` rich-text array.""" + return "".join(p.get("plain_text", "") for p in (title_array or [])) + + +async def _find_or_create_data_source(client: NotionClient, spec: _DatabaseSpec) -> str: + """Resolve the data_source_id for a system-managed target. + + Strategy: look under the parent for a database / data source with + matching title. If found, reuse it. If not, create it with the declared + schema. Returns the data_source_id. + """ + assert spec.managed_by == "system" + assert spec.title is not None + + if spec.parent_page_id is not None: + # Notion pagination docs, checked 2026-06-01: + # GET list endpoints return at most 100 results and require passing + # next_cursor as start_cursor to continue. + cursor: str | None = None + matching_data_source_ids: list[str] = [] + fallback_single_source_ids: list[str] = [] + while True: + params = {"page_size": "100"} + if cursor is not None: + params["start_cursor"] = cursor + children = await client._request( + "GET", + f"/blocks/{spec.parent_page_id}/children?{urlencode(params)}", + ) + for child in children.get("results", []): + if child.get("type") != "child_database": + continue + try: + db = await client.get_database(child["id"]) + except Exception: + continue + data_sources = db.get("data_sources") or [] + matching_data_sources = [ + ds for ds in data_sources if ds.get("name") == spec.title + ] + matching_data_source_ids.extend( + str(ds["id"]) for ds in matching_data_sources + ) + if ( + len(data_sources) == 1 + and _read_notion_title(db.get("title")) == spec.title + ): + fallback_single_source_ids.append(str(data_sources[0]["id"])) + if not children.get("has_more"): + break + next_cursor = children.get("next_cursor") + if next_cursor is None: + break + cursor = str(next_cursor) + if len(matching_data_source_ids) == 1: + return matching_data_source_ids[0] + if len(matching_data_source_ids) > 1: + raise ValueError( + f"Found multiple Notion data sources named {spec.title!r} " + f"under parent {spec.parent_page_id!r}. Pass " + "parent_database_id or choose a unique data source title." + ) + if len(fallback_single_source_ids) == 1: + return fallback_single_source_ids[0] + if len(fallback_single_source_ids) > 1: + raise ValueError( + f"Found multiple single-source Notion databases titled " + f"{spec.title!r} under parent {spec.parent_page_id!r}. Pass " + "parent_database_id or choose a unique data source title." + ) + # Not found — create. + new_db = await client.create_database( + parent_page_id=spec.parent_page_id, + title=spec.title, + properties=spec.schema.to_notion_properties(), + ) + created_id: str = (new_db.get("data_sources") or [{}])[0]["id"] + return created_id + + if spec.parent_database_id is not None: + # Look for an existing data source with matching name on the database. + db = await client.get_database(spec.parent_database_id) + for ds_info in db.get("data_sources") or []: + if ds_info.get("name") == spec.title: + existing_id: str = ds_info["id"] + return existing_id + new_ds = await client.create_data_source( + parent_database_id=spec.parent_database_id, + title=spec.title, + properties=spec.schema.to_notion_properties(), + ) + new_id: str = new_ds["id"] + return new_id + + raise ValueError( + "managed_by='system' requires parent_page_id or parent_database_id." + ) + + +async def _evolve_schema_if_needed( + client: NotionClient, data_source_id: str, spec: _DatabaseSpec +) -> None: + """For system mode: PATCH-add any properties the dataclass declares that + aren't yet on the live data source. Destructive changes (type mismatches) + are rejected unless ``allow_destructive=True``. + """ + ds = await client.get_data_source(data_source_id) + notion_props = ds.get("properties") or {} + missing, type_mismatch = spec.schema.diff_against(notion_props) + prop_by_name = spec.schema.properties_by_notion_name + + # Notion data-source property docs, checked 2026-06-01: + # every data source requires exactly one title property, and its type + # cannot be changed or added/removed through schema PATCHes. + missing_title = [ + name for name in missing if prop_by_name[name].notion_type == "title" + ] + mismatched_title = [ + name + for name, declared, actual in type_mismatch + if declared == "title" or actual == "title" + ] + if missing_title or mismatched_title: + raise ValueError( + f"{spec.schema.record_type.__name__}: Notion title property cannot " + "be added or type-changed via the API. Edit the data source title " + "property in Notion's UI to match the declared schema." + ) + + if type_mismatch and not spec.allow_destructive: + details = ", ".join( + f"{name!r} declared {declared!r} but Notion has {actual!r}" + for name, declared, actual in type_mismatch + ) + raise ValueError( + f"{spec.schema.record_type.__name__}: destructive schema change " + f"rejected ({details}). Either edit the schema in Notion's UI to " + "match, or pass allow_destructive=True to apply the change." + ) + + updates: dict[str, dict[str, Any] | None] = { + name: prop_by_name[name].to_notion_schema() for name in missing + } + if spec.allow_destructive: + # Notion update-data-source docs, checked 2026-06-01: + # property type changes are PATCHed by sending the new type schema. + # Not all conversions are accepted by Notion; surface that API error. + for name, _, _ in type_mismatch: + updates[name] = prop_by_name[name].to_notion_schema() + if updates: + await client.update_data_source_properties(data_source_id, updates) + + +async def _apply_database_actions( + context_provider: ContextProvider, + actions: Sequence[_DatabaseAction], +) -> list[coco.ChildTargetDef[_RowHandler] | None]: + outputs: list[coco.ChildTargetDef[_RowHandler] | None] = [None] * len(actions) + + for i, action in enumerate(actions): + if coco.is_non_existence(action.spec): + # Target un-mounted. We don't touch the data source itself — even + # in system mode, the user may want to recover the data. Pages + # declared via this target on prior runs are not archived here + # because we drop the child handler; keep the target declared and + # stop declaring rows individually for that. + outputs[i] = None + continue + + spec = action.spec + client = context_provider.get(action.key.client_key, NotionClient) + + # Resolve the data source. In user mode, it's spec.data_source_id and + # must already exist. In system mode, we find-or-create it. + if spec.managed_by == "user": + assert spec.data_source_id is not None + data_source_id = spec.data_source_id + try: + ds = await client.get_data_source(data_source_id) + except Exception as e: + raise RuntimeError( + f"Notion data source {data_source_id!r} is not " + "accessible. Confirm the integration is shared with the " + "containing page (top-right ··· → Connections)." + ) from e + spec.schema.validate_against(ds.get("properties") or {}) + else: + data_source_id = await _find_or_create_data_source(client, spec) + await _evolve_schema_if_needed(client, data_source_id, spec) + + outputs[i] = coco.ChildTargetDef( + handler=_RowHandler( + client=client, + data_source_id=data_source_id, + schema=spec.schema, + on_delete=spec.on_delete, + ) + ) + + return outputs + + +_database_action_sink = coco.TargetActionSink[ + _DatabaseAction, _RowHandler +].from_async_fn(_apply_database_actions) + + +class _DatabaseHandler( + coco.TargetHandler[_DatabaseSpec, _DatabaseTracking, _RowHandler] +): + """Parent handler — owns the data source identity across runs.""" + + def reconcile( + self, + key: coco.StableKey, + desired_state: _DatabaseSpec | coco.NonExistenceType, + prev_possible_records: Collection[_DatabaseTracking], + prev_may_be_missing: bool, + /, + ) -> ( + coco.TargetReconcileOutput[_DatabaseAction, _DatabaseTracking, _RowHandler] + | None + ): + # StableKey -> _DatabaseKey + if isinstance(key, tuple) and len(key) == 2: + db_key = _DatabaseKey(client_key=str(key[0]), identity=str(key[1])) + elif isinstance(key, _DatabaseKey): + db_key = key + else: + raise TypeError( + f"_DatabaseHandler: expected _DatabaseKey, got {type(key).__name__}" + ) + + if coco.is_non_existence(desired_state): + return coco.TargetReconcileOutput( + action=_DatabaseAction(key=db_key, spec=coco.NON_EXISTENCE), + sink=_database_action_sink, + tracking_record=coco.NON_EXISTENCE, + ) + + tracking = _DatabaseTracking(identity=db_key.identity) + return coco.TargetReconcileOutput( + action=_DatabaseAction(key=db_key, spec=desired_state), + sink=_database_action_sink, + tracking_record=tracking, + ) + + +_database_provider = coco.register_root_target_states_provider( + "cocoindex/notion/database", _DatabaseHandler() +) + + +# --------------------------------------------------------------------------- +# User-facing API +# --------------------------------------------------------------------------- + + +class DatabaseTarget( + Generic[RowT, coco.MaybePendingS], + coco.ResolvesTo["DatabaseTarget[RowT]"], +): + """A target for writing rows to a Notion database (data source). + + Acquired via :func:`mount_database_target` or :func:`declare_database_target`. + Call :meth:`declare_row` for each row to upsert; rows declared in a previous + run but not in this run are automatically archived (subject to ``on_delete``). + """ + + _provider: "coco.TargetStateProvider[_PageValue, None, coco.MaybePendingS]" + _schema: DatabaseSchema[RowT] + + def __init__( + self, + provider: "coco.TargetStateProvider[_PageValue, None, coco.MaybePendingS]", + schema: DatabaseSchema[RowT], + ) -> None: + self._provider = provider + self._schema = schema + + def declare_row(self: "DatabaseTarget[RowT]", *, row: RowT) -> None: + """Declare a row to be upserted to this database. + + ``row`` is a dataclass / NamedTuple / dict / Pydantic model bound to + the :class:`DatabaseSchema`. Primary-key fields must be set. + """ + # PK tuple + if isinstance(row, dict): + pk_values: _PageKey = tuple(row.get(pk) for pk in self._schema.primary_key) + else: + pk_values = tuple(getattr(row, pk) for pk in self._schema.primary_key) + + # Field-name -> Python value (the row itself, frozen as a dict) + value: _PageValue = {} + for field_name, _ in self._schema.properties: + if isinstance(row, dict): + value[field_name] = row.get(field_name) + else: + value[field_name] = getattr(row, field_name, None) + + coco.declare_target_state(self._provider.target_state(pk_values, value)) + + def __coco_memo_key__(self) -> str: + key: str = self._provider.memo_key + return key + + +def _build_spec( + data_source_id: str | None, + schema: DatabaseSchema[RowT], + on_delete: OnDelete, + managed_by: ManagedBy, + parent_page_id: str | None, + parent_database_id: str | None, + title: str | None, + allow_destructive: bool, +) -> tuple[str, _DatabaseSpec]: + """Validate the combination of args and return ``(identity, spec)``. + + User mode requires ``data_source_id`` and forbids the system-mode kwargs. + System mode requires ``title`` plus exactly one of ``parent_page_id`` / + ``parent_database_id``. + """ + if managed_by == "user": + if data_source_id is None: + raise ValueError( + "managed_by='user' requires data_source_id (the existing " + "Notion data source's ID)." + ) + if any(v is not None for v in (parent_page_id, parent_database_id, title)): + raise ValueError( + "parent_page_id / parent_database_id / title are only valid " + "with managed_by='system'." + ) + return data_source_id, _DatabaseSpec( + schema=schema, + on_delete=on_delete, + managed_by="user", + data_source_id=data_source_id, + ) + + # managed_by == "system" + if data_source_id is not None: + raise ValueError( + "managed_by='system' creates the data source; don't pass " + "data_source_id. Use parent_page_id or parent_database_id + title." + ) + if title is None: + raise ValueError("managed_by='system' requires title.") + if (parent_page_id is None) == (parent_database_id is None): + raise ValueError( + "managed_by='system' requires exactly one of parent_page_id or " + "parent_database_id." + ) + spec = _DatabaseSpec( + schema=schema, + on_delete=on_delete, + managed_by="system", + parent_page_id=parent_page_id, + parent_database_id=parent_database_id, + title=title, + allow_destructive=allow_destructive, + ) + return _system_identity(spec), spec + + +def database_target( + client: ContextKey[NotionClient], + data_source_id: str | None = None, + schema: DatabaseSchema[RowT] | None = None, + *, + managed_by: ManagedBy = "system", + parent_page_id: str | None = None, + parent_database_id: str | None = None, + title: str | None = None, + on_delete: OnDelete = OnDelete.ARCHIVE, + allow_destructive: bool = False, +) -> "coco.TargetState[_RowHandler]": + """Create a TargetState for a Notion database target. Prefer the + :func:`mount_database_target` wrapper. + """ + if schema is None: + raise ValueError("schema is required.") + identity, spec = _build_spec( + data_source_id, + schema, + on_delete, + managed_by, + parent_page_id, + parent_database_id, + title, + allow_destructive, + ) + key = _DatabaseKey(client_key=client.key, identity=identity) + return _database_provider.target_state(key, spec) + + +def declare_database_target( + client: ContextKey[NotionClient], + data_source_id: str | None = None, + schema: DatabaseSchema[RowT] | None = None, + *, + managed_by: ManagedBy = "system", + parent_page_id: str | None = None, + parent_database_id: str | None = None, + title: str | None = None, + on_delete: OnDelete = OnDelete.ARCHIVE, + allow_destructive: bool = False, +) -> "DatabaseTarget[RowT, coco.PendingS]": + """Declare a database target and return a ready-to-declare DatabaseTarget.""" + if schema is None: + raise ValueError("schema is required.") + provider = coco.declare_target_state_with_child( + database_target( + client, + data_source_id, + schema, + managed_by=managed_by, + parent_page_id=parent_page_id, + parent_database_id=parent_database_id, + title=title, + on_delete=on_delete, + allow_destructive=allow_destructive, + ) + ) + return DatabaseTarget(provider, schema) + + +async def mount_database_target( + client: ContextKey[NotionClient], + data_source_id: str | None = None, + schema: DatabaseSchema[RowT] | None = None, + *, + managed_by: ManagedBy = "system", + parent_page_id: str | None = None, + parent_database_id: str | None = None, + title: str | None = None, + on_delete: OnDelete = OnDelete.ARCHIVE, + allow_destructive: bool = False, +) -> "DatabaseTarget[RowT]": + """Mount a Notion database target and return a ready-to-use DatabaseTarget. + + Two modes: + + - ``managed_by="system"`` (default): pass ``parent_page_id`` or + ``parent_database_id`` plus ``title``. The connector creates the data + source on first run if it doesn't exist, and PATCH-adds new properties + on subsequent runs when the dataclass grows. Destructive changes + (existing property type changed) are rejected unless + ``allow_destructive=True``. + - ``managed_by="user"``: pass an existing ``data_source_id``. The + connector validates that the live property schema matches. + """ + if schema is None: + raise ValueError("schema is required.") + provider = await coco.mount_target( + database_target( + client, + data_source_id, + schema, + managed_by=managed_by, + parent_page_id=parent_page_id, + parent_database_id=parent_database_id, + title=title, + on_delete=on_delete, + allow_destructive=allow_destructive, + ) + ) + return DatabaseTarget(provider, schema) + + +__all__ = [ + "DatabaseTarget", + "ManagedBy", + "OnDelete", + "database_target", + "declare_database_target", + "mount_database_target", +] diff --git a/python/cocoindex/connectors/notion/_types.py b/python/cocoindex/connectors/notion/_types.py new file mode 100644 index 000000000..97ee7b0af --- /dev/null +++ b/python/cocoindex/connectors/notion/_types.py @@ -0,0 +1,414 @@ +"""Property type system + schema for the Notion target. + +Each PropType encodes a single Notion property: its Notion API type name, +how to encode a Python value into the property's JSON shape, and how to +decode it back (used by query-on-miss to extract the primary key from an +existing page returned by ``POST /v1/data_sources/{id}/query``). + +The MVP supports the property types you'd reach for first: +title, rich_text, number, url, email, date, select, multi_select, checkbox, +relation. People and files can land in a follow-up. +""" + +from __future__ import annotations + +from collections import Counter +from dataclasses import dataclass +from datetime import date, datetime +from typing import Any, Generic, get_args, get_origin, get_type_hints + +from typing_extensions import Annotated, TypeVar + +RowT = TypeVar("RowT") + + +# --------------------------------------------------------------------------- +# Property type base + concrete classes +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class PropType: + """Base class for Notion property type bindings. + + Subclasses set `notion_type` to the Notion API's property type string + (e.g. ``"title"``, ``"rich_text"``, ``"select"``) and implement + ``encode``/``decode``. + """ + + name: str # The property's name as it appears in the Notion data source + notion_type: str = "" + + def encode(self, value: Any) -> dict[str, Any]: + """Return the Notion property JSON body for this value.""" + raise NotImplementedError + + def decode(self, prop_json: dict[str, Any]) -> Any: + """Extract the Python value from a Notion property JSON body.""" + raise NotImplementedError + + def to_notion_schema(self) -> dict[str, Any]: + """Return the schema body for this property when creating it via the + Notion API. Used by ``managed_by="system"`` mode at + ``POST /v1/databases`` (initial schema) and ``PATCH /v1/data_sources/{id}`` + (additive new column). Returns just the type-config dict; the caller + wraps it as ``{name: schema}``. + """ + return {self.notion_type: {}} + + +@dataclass(frozen=True) +class TitleProp(PropType): + """The data source's single title property (every Notion DS has exactly one).""" + + notion_type: str = "title" + + def encode(self, value: Any) -> dict[str, Any]: + text = "" if value is None else str(value) + return {"title": [{"text": {"content": text[:2000]}}]} + + def decode(self, prop_json: dict[str, Any]) -> Any: + parts = prop_json.get("title") or [] + return "".join(p.get("plain_text", "") for p in parts) + + +@dataclass(frozen=True) +class RichTextProp(PropType): + notion_type: str = "rich_text" + + def encode(self, value: Any) -> dict[str, Any]: + text = "" if value is None else str(value) + return {"rich_text": [{"text": {"content": text[:2000]}}]} + + def decode(self, prop_json: dict[str, Any]) -> Any: + parts = prop_json.get("rich_text") or [] + return "".join(p.get("plain_text", "") for p in parts) + + +@dataclass(frozen=True) +class NumberProp(PropType): + notion_type: str = "number" + + def encode(self, value: Any) -> dict[str, Any]: + return {"number": None if value is None else float(value)} + + def decode(self, prop_json: dict[str, Any]) -> Any: + return prop_json.get("number") + + +@dataclass(frozen=True) +class UrlProp(PropType): + notion_type: str = "url" + + def encode(self, value: Any) -> dict[str, Any]: + return {"url": None if value is None else str(value)} + + def decode(self, prop_json: dict[str, Any]) -> Any: + return prop_json.get("url") + + +@dataclass(frozen=True) +class EmailProp(PropType): + notion_type: str = "email" + + def encode(self, value: Any) -> dict[str, Any]: + return {"email": None if value is None else str(value)} + + def decode(self, prop_json: dict[str, Any]) -> Any: + return prop_json.get("email") + + +@dataclass(frozen=True) +class SelectProp(PropType): + notion_type: str = "select" + # Optional pre-declared options. Notion auto-creates options on write if + # they're not pre-declared, so this is only needed when you want to control + # the option color/order at managed_by="system" create time. + options: tuple[str, ...] = () + + def encode(self, value: Any) -> dict[str, Any]: + return {"select": None if value is None else {"name": str(value)}} + + def decode(self, prop_json: dict[str, Any]) -> Any: + sel = prop_json.get("select") + return sel.get("name") if sel else None + + def to_notion_schema(self) -> dict[str, Any]: + return {"select": {"options": [{"name": o} for o in self.options]}} + + +@dataclass(frozen=True) +class MultiSelectProp(PropType): + notion_type: str = "multi_select" + options: tuple[str, ...] = () + + def encode(self, value: Any) -> dict[str, Any]: + items = list(value or []) + return {"multi_select": [{"name": str(v)} for v in items]} + + def decode(self, prop_json: dict[str, Any]) -> Any: + return [opt.get("name") for opt in (prop_json.get("multi_select") or [])] + + def to_notion_schema(self) -> dict[str, Any]: + return {"multi_select": {"options": [{"name": o} for o in self.options]}} + + +@dataclass(frozen=True) +class CheckboxProp(PropType): + notion_type: str = "checkbox" + + def encode(self, value: Any) -> dict[str, Any]: + return {"checkbox": bool(value)} + + def decode(self, prop_json: dict[str, Any]) -> Any: + return bool(prop_json.get("checkbox")) + + +@dataclass(frozen=True) +class RelationProp(PropType): + """A Notion relation property pointing at another data source. + + The Python value is a list of page IDs (or a single page ID string) of + pages in the related data source. The connector does not look these up + for you — when porting from notion-client-style code, you typically pass + the page IDs you got from an upstream upsert. + + For system mode, ``target_data_source_id`` is required to create the + relation column on the live Notion data source. + """ + + notion_type: str = "relation" + target_data_source_id: str | None = None + + def encode(self, value: Any) -> dict[str, Any]: + if value is None: + return {"relation": []} + if isinstance(value, str): + ids = [value] + else: + ids = [str(v) for v in value if v] + return {"relation": [{"id": pid} for pid in ids]} + + def decode(self, prop_json: dict[str, Any]) -> Any: + return [r["id"] for r in (prop_json.get("relation") or []) if r.get("id")] + + def to_notion_schema(self) -> dict[str, Any]: + if self.target_data_source_id is None: + raise ValueError( + f"RelationProp {self.name!r} requires target_data_source_id when " + "CocoIndex creates or adds the Notion relation property." + ) + return {"relation": {"data_source_id": self.target_data_source_id}} + + +@dataclass(frozen=True) +class DateProp(PropType): + notion_type: str = "date" + + def encode(self, value: Any) -> dict[str, Any]: + if value is None: + return {"date": None} + if isinstance(value, datetime): + return {"date": {"start": value.isoformat()}} + if isinstance(value, date): + return {"date": {"start": value.isoformat()}} + return {"date": {"start": str(value)}} + + def decode(self, prop_json: dict[str, Any]) -> Any: + d = prop_json.get("date") + if not d: + return None + return d.get("start") + + +# --------------------------------------------------------------------------- +# Schema +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class DatabaseSchema(Generic[RowT]): + """A bound mapping from a Python class's fields onto Notion properties. + + Built via :meth:`from_class` — the user either annotates each field with + ``Annotated[T, SomePropType(...)]`` or passes a ``property_map`` override. + """ + + record_type: type[RowT] + primary_key: tuple[str, ...] + properties: tuple[tuple[str, PropType], ...] # ordered: (field_name, prop) + + @property + def properties_by_field(self) -> dict[str, PropType]: + return {f: p for f, p in self.properties} + + @property + def properties_by_notion_name(self) -> dict[str, PropType]: + return {p.name: p for _, p in self.properties} + + @classmethod + async def from_class( + cls, + record_type: type[RowT], + primary_key: list[str], + *, + property_map: dict[str, PropType] | None = None, + ) -> "DatabaseSchema[RowT]": + """Build a DatabaseSchema from a dataclass / NamedTuple / similar. + + Order of precedence for the property binding of each field: + 1. ``property_map[field_name]`` if provided + 2. ``Annotated[T, PropType(...)]`` metadata on the field + + Either source must yield exactly one ``PropType`` per field. Fields + with neither are skipped (allowing the dataclass to carry transient + state that doesn't land in Notion). + """ + property_map = property_map or {} + hints = get_type_hints(record_type, include_extras=True) + + # Catch property_map keys that don't name an actual field — almost + # always a typo and silently swallowing it leads to "why are my rows + # missing this column?" debugging. + unknown_pm_keys = set(property_map) - set(hints) + if unknown_pm_keys: + raise ValueError( + f"{record_type.__name__}: property_map keys {sorted(unknown_pm_keys)} " + f"do not match any field. Known fields: {sorted(hints)}." + ) + + bindings: list[tuple[str, PropType]] = [] + for field_name, type_hint in hints.items(): + if field_name in property_map: + bindings.append((field_name, property_map[field_name])) + continue + if get_origin(type_hint) is Annotated: + for meta in get_args(type_hint)[1:]: + if isinstance(meta, PropType): + bindings.append((field_name, meta)) + break + + if not bindings: + raise ValueError( + f"{record_type.__name__}: no Notion properties found. " + "Annotate fields with Annotated[T, notion.SomeProp(...)] " + "or pass property_map={...}." + ) + + # Sanity: at most one title property. + titles = [f for f, p in bindings if p.notion_type == "title"] + if len(titles) > 1: + raise ValueError( + f"{record_type.__name__}: only one TitleProp allowed; got {titles}" + ) + + notion_names = Counter(prop.name for _, prop in bindings) + duplicate_notion_names = sorted( + name for name, count in notion_names.items() if count > 1 + ) + if duplicate_notion_names: + raise ValueError( + f"{record_type.__name__}: multiple fields map to the same Notion " + f"property name(s): {duplicate_notion_names}." + ) + + # Sanity: every PK field must be in bindings. + binding_fields = {f for f, _ in bindings} + missing_pk = [pk for pk in primary_key if pk not in binding_fields] + if missing_pk: + raise ValueError( + f"{record_type.__name__}: primary_key fields {missing_pk} " + "have no Notion property binding." + ) + + return cls( + record_type=record_type, + primary_key=tuple(primary_key), + properties=tuple(bindings), + ) + + def encode_row(self, row: Any) -> dict[str, dict[str, Any]]: + """Encode a row instance into Notion's ``properties`` body.""" + out: dict[str, dict[str, Any]] = {} + for field_name, prop in self.properties: + if isinstance(row, dict): + value = row.get(field_name) + else: + value = getattr(row, field_name, None) + out[prop.name] = prop.encode(value) + return out + + def extract_pk(self, page_properties: dict[str, Any]) -> tuple[Any, ...]: + """Pull the primary-key tuple out of a Notion page's properties payload.""" + prop_by_field = self.properties_by_field + return tuple( + prop_by_field[pk].decode(page_properties.get(prop_by_field[pk].name, {})) + for pk in self.primary_key + ) + + def to_notion_properties(self) -> dict[str, dict[str, Any]]: + """Build the ``properties`` body for ``POST /v1/databases`` / + ``POST /v1/data_sources``. Keys are Notion property names, values are + the per-type schema configs (e.g. ``{"select": {"options": [...]}}``). + """ + return {prop.name: prop.to_notion_schema() for _, prop in self.properties} + + def diff_against( + self, notion_schema: dict[str, dict[str, Any]] + ) -> tuple[list[str], list[tuple[str, str, str]]]: + """Return ``(missing_property_names, type_mismatches)`` between the + declared schema and the live Notion data source. Used by both + :meth:`validate_against` (user-managed, both lists become errors) and + the system-managed reconcile path (missing → PATCH-add, mismatches → + reject unless ``allow_destructive``). + """ + missing: list[str] = [] + type_mismatch: list[tuple[str, str, str]] = [] + for _, prop in self.properties: + notion_prop = notion_schema.get(prop.name) + if notion_prop is None: + missing.append(prop.name) + continue + actual_type = notion_prop.get("type") + if actual_type != prop.notion_type: + type_mismatch.append((prop.name, prop.notion_type, str(actual_type))) + return missing, type_mismatch + + def validate_against(self, notion_schema: dict[str, dict[str, Any]]) -> None: + """Check the declared schema against the live Notion data source. + + ``notion_schema`` is the ``properties`` map from + ``GET /v1/data_sources/{id}`` — keys are property names, values look like + ``{"id": ..., "name": ..., "type": "select", "select": {...}}``. + + Errors loudly on two cases that would otherwise fail silently at write + time: + + - Declared property name doesn't exist in Notion. + - Declared property name exists but with a different Notion type. + + Extra properties in Notion that the schema doesn't declare are left + alone (extra-columns-on-target is fine — they aren't touched). + """ + missing, type_mismatch = self.diff_against(notion_schema) + if not missing and not type_mismatch: + return + + parts: list[str] = [] + if missing: + available = sorted(notion_schema) + parts.append( + f"missing properties {sorted(missing)} (data source has: {available})" + ) + if type_mismatch: + parts.append( + "type mismatches: " + + ", ".join( + f"{name!r} declared {declared!r} but Notion has {actual!r}" + for name, declared, actual in type_mismatch + ) + ) + raise ValueError( + f"{self.record_type.__name__} schema does not match the Notion data " + f"source: " + "; ".join(parts) + ) diff --git a/python/tests/connectors/test_notion_target.py b/python/tests/connectors/test_notion_target.py new file mode 100644 index 000000000..e254ced8d --- /dev/null +++ b/python/tests/connectors/test_notion_target.py @@ -0,0 +1,1176 @@ +"""Tests for the Notion target connector. + +Requires a real Notion workspace. Gated by env vars: + + NOTION_TEST_TOKEN — internal-integration secret + NOTION_TEST_PARENT_PAGE — page UUID, shared with the integration; tests + create temporary databases under it and archive + them on teardown. + +The whole suite is skipped if either is missing — match the optional-dep +pattern used elsewhere in this repo. +""" + +from __future__ import annotations + +import asyncio +import os +import time +import uuid +from dataclasses import dataclass +from datetime import date, datetime +from typing import Any, AsyncIterator, Callable, Coroutine, cast + +import pytest +import pytest_asyncio +from typing_extensions import Annotated + +import cocoindex as coco +from cocoindex._internal.context_keys import ContextProvider +from cocoindex.connectors import notion +from cocoindex.connectors.notion._target import ( # pyright: ignore[reportPrivateUsage] + _DatabaseSpec, + _evolve_schema_if_needed, + _find_or_create_data_source, + _property_filter, +) + +from tests import common + +NOTION_TEST_TOKEN = os.environ.get("NOTION_TEST_TOKEN") +NOTION_TEST_PARENT_PAGE = os.environ.get("NOTION_TEST_PARENT_PAGE") + +requires_notion_env = pytest.mark.skipif( + not (NOTION_TEST_TOKEN and NOTION_TEST_PARENT_PAGE), + reason="NOTION_TEST_TOKEN and NOTION_TEST_PARENT_PAGE not set", +) + +NOTION_CK = coco.ContextKey[notion.NotionClient]("notion_test_client") + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _unique_title(test_name: str) -> str: + """Distinct title per test invocation so reruns don't collide.""" + return f"cocoindex-test-{test_name}-{int(time.time())}-{uuid.uuid4().hex[:6]}" + + +async def _create_test_db( + client: notion.NotionClient, + parent_page_id: str, + title: str, + properties: dict[str, dict[str, Any]], +) -> str: + """Create a test database and return the data_source_id.""" + res = await client.create_database( + parent_page_id=parent_page_id, title=title, properties=properties + ) + data_source_id: str = (res.get("data_sources") or [{}])[0]["id"] + return data_source_id + + +async def _archive_db(client: notion.NotionClient, data_source_id: str) -> None: + """Best-effort archive — never raise from teardown.""" + try: + ds = await client.get_data_source(data_source_id) + db_id = (ds.get("parent") or {}).get("database_id") + if db_id: + await client._request("PATCH", f"/databases/{db_id}", {"in_trash": True}) + except Exception: + pass + + +async def _active_pages( + client: notion.NotionClient, data_source_id: str +) -> list[dict[str, Any]]: + pages = [] + async for page in client.query_all(data_source_id): + pages.append(page) + return pages + + +def _title_of(page: dict[str, Any]) -> str: + parts = page["properties"].get("Name", {}).get("title") or [] + return "".join(p.get("plain_text", "") for p in parts) + + +def _make_env(client: notion.NotionClient, suffix: str) -> coco.Environment: + ctx = ContextProvider() + ctx.provide(NOTION_CK, client) + settings = coco.Settings.from_env( + db_path=common.get_env_db_path(f"connectors__test_notion_target__{suffix}") + ) + return coco.Environment(settings, context_provider=ctx) + + +# --------------------------------------------------------------------------- +# Row types +# --------------------------------------------------------------------------- + + +@dataclass +class Person: + name: Annotated[str, notion.TitleProp("Name")] + email: Annotated[str, notion.EmailProp("Email")] + role: Annotated[str, notion.SelectProp("Role")] + active: Annotated[bool, notion.CheckboxProp("Active")] + + +@dataclass +class PersonWithNotes(Person): + notes: Annotated[str, notion.RichTextProp("Notes")] + + +PERSON_SCHEMA_PROPS: dict[str, dict[str, Any]] = { + "Name": {"title": {}}, + "Email": {"email": {}}, + "Role": { + "select": { + "options": [ + {"name": "Engineer"}, + {"name": "Designer"}, + ] + } + }, + "Active": {"checkbox": {}}, +} + + +class _FakeSystemClient(notion.NotionClient): + def __init__(self) -> None: + super().__init__(token="fake", session=cast(Any, object())) + self.child_pages: list[dict[str, Any]] = [] + self.databases: dict[str, dict[str, Any]] = {} + self.data_source: dict[str, Any] = {"properties": {}} + self.created_databases: list[dict[str, Any]] = [] + self.schema_updates: list[dict[str, dict[str, Any] | None]] = [] + self.request_paths: list[str] = [] + + async def _request( + self, method: str, path: str, json_body: dict[str, Any] | None = None + ) -> dict[str, Any]: + self.request_paths.append(path) + assert method == "GET" + assert path.startswith("/blocks/parent-page/children?") + if not self.child_pages: + return {"results": [], "has_more": False, "next_cursor": None} + if "start_cursor=cursor-2" in path: + return { + "results": [self.child_pages[1]], + "has_more": False, + "next_cursor": None, + } + return { + "results": [self.child_pages[0]], + "has_more": len(self.child_pages) > 1, + "next_cursor": "cursor-2" if len(self.child_pages) > 1 else None, + } + + async def get_database(self, database_id: str) -> dict[str, Any]: + return self.databases[database_id] + + async def create_database( + self, + *, + parent_page_id: str, + title: str, + properties: dict[str, dict[str, Any]], + ) -> dict[str, Any]: + self.created_databases.append( + { + "parent_page_id": parent_page_id, + "title": title, + "properties": properties, + } + ) + return {"data_sources": [{"id": "created-ds"}]} + + async def get_data_source(self, data_source_id: str) -> dict[str, Any]: + assert data_source_id == "ds-1" + return self.data_source + + async def update_data_source_properties( + self, + data_source_id: str, + properties: dict[str, dict[str, Any] | None], + ) -> dict[str, Any]: + assert data_source_id == "ds-1" + self.schema_updates.append(properties) + return {"id": data_source_id, "properties": properties} + + +class _FakeResponse: + def __init__(self, status: int, body: dict[str, Any]) -> None: + self.status = status + self.headers: dict[str, str] = {} + self._body = body + + async def __aenter__(self) -> "_FakeResponse": + return self + + async def __aexit__(self, *exc: object) -> None: + return None + + def raise_for_status(self) -> None: + return None + + async def json(self) -> dict[str, Any]: + return self._body + + +class _FakeSession: + def __init__(self, second_request_started: asyncio.Event) -> None: + self.second_request_started = second_request_started + self.calls = 0 + + def request( + self, + method: str, + url: str, + *, + json: dict[str, Any] | None, + headers: dict[str, str], + ) -> _FakeResponse: + self.calls += 1 + if self.calls == 1: + response = _FakeResponse(429, {}) + response.headers["Retry-After"] = "1" + return response + self.second_request_started.set() + return _FakeResponse(200, {"ok": True}) + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest_asyncio.fixture +async def client() -> AsyncIterator[notion.NotionClient]: + """Per-test NotionClient with the test token.""" + assert NOTION_TEST_TOKEN is not None + async with notion.NotionClient(token=NOTION_TEST_TOKEN, max_concurrency=1) as c: + yield c + + +@pytest_asyncio.fixture +async def user_db( + client: notion.NotionClient, request: pytest.FixtureRequest +) -> AsyncIterator[str]: + """Create a pre-existing data source for user-mode tests; teardown archives it.""" + assert NOTION_TEST_PARENT_PAGE is not None + title = _unique_title(request.node.name) + ds_id = await _create_test_db( + client, NOTION_TEST_PARENT_PAGE, title, PERSON_SCHEMA_PROPS + ) + yield ds_id + await _archive_db(client, ds_id) + + +# --------------------------------------------------------------------------- +# Schema validation (no Notion access needed) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_property_map_typo_raises() -> None: + """Bug fix #2: typoed key in property_map should error, not silently drop.""" + + @dataclass + class R: + name: str + email: str + + with pytest.raises(ValueError, match="property_map keys"): + await notion.DatabaseSchema.from_class( + R, + primary_key=["name"], + property_map={"naame": notion.TitleProp("Name")}, + ) + + +@pytest.mark.asyncio +async def test_schema_requires_at_most_one_title() -> None: + @dataclass + class TwoTitles: + name: Annotated[str, notion.TitleProp("Name")] + other: Annotated[str, notion.TitleProp("Other")] + + with pytest.raises(ValueError, match="only one TitleProp"): + await notion.DatabaseSchema.from_class(TwoTitles, primary_key=["name"]) + + +@pytest.mark.asyncio +async def test_duplicate_notion_property_names_raise() -> None: + @dataclass + class DuplicateNames: + name: Annotated[str, notion.TitleProp("Name")] + display_name: Annotated[str, notion.RichTextProp("Name")] + + with pytest.raises(ValueError, match="same Notion property"): + await notion.DatabaseSchema.from_class(DuplicateNames, primary_key=["name"]) + + +@pytest.mark.asyncio +async def test_relation_prop_encode_decode() -> None: + """RelationProp writes page-id lists and creates data-source relations.""" + p = notion.RelationProp("Account") + assert p.encode(["page1", "page2"]) == { + "relation": [{"id": "page1"}, {"id": "page2"}] + } + assert p.encode("page1") == {"relation": [{"id": "page1"}]} + assert p.encode(None) == {"relation": []} + assert p.encode([]) == {"relation": []} + assert p.decode({"relation": [{"id": "p1"}, {}, {"id": "p2"}]}) == [ + "p1", + "p2", + ] + with pytest.raises(ValueError, match="target_data_source_id"): + p.to_notion_schema() + + p_typed = notion.RelationProp("Account", target_data_source_id="ds-related") + assert p_typed.to_notion_schema() == {"relation": {"data_source_id": "ds-related"}} + + +def test_date_primary_key_filter_uses_isoformat() -> None: + prop = notion.DateProp("Published") + + assert _property_filter(prop, date(2026, 6, 1)) == { + "property": "Published", + "date": {"equals": "2026-06-01"}, + } + assert _property_filter(prop, datetime(2026, 6, 1, 12, 34, 56)) == { + "property": "Published", + "date": {"equals": "2026-06-01T12:34:56"}, + } + + +@pytest.mark.asyncio +async def test_managed_by_args_validation() -> None: + """User mode rejects parent_/title kwargs; system mode rejects data_source_id.""" + + @dataclass + class R: + name: Annotated[str, notion.TitleProp("Name")] + + client_key = coco.ContextKey[notion.NotionClient]("validation_test") + schema = await notion.DatabaseSchema.from_class(R, primary_key=["name"]) + + # User mode: missing data_source_id + with pytest.raises(ValueError, match="managed_by='user' requires data_source_id"): + notion.database_target(client_key, None, schema, managed_by="user") + + # User mode: extra system kwargs + with pytest.raises(ValueError, match="only valid with managed_by='system'"): + notion.database_target( + client_key, "ds-id", schema, managed_by="user", title="Foo" + ) + + # System mode: missing title + with pytest.raises(ValueError, match="managed_by='system' requires title"): + notion.database_target( + client_key, None, schema, managed_by="system", parent_page_id="p" + ) + + # System mode: needs exactly one parent + with pytest.raises(ValueError, match="exactly one of parent_page_id"): + notion.database_target( + client_key, None, schema, managed_by="system", title="Foo" + ) + + +# --------------------------------------------------------------------------- +# API-contract regressions (no Notion access) +# --------------------------------------------------------------------------- +# +# The approved design's acceptance path is real Notion API coverage. These unit +# tests are intentionally narrower: they cover hard-to-force edge contracts from +# Notion's public API docs, while the integration tests below keep exercising +# the actual connector lifecycle against a real workspace when env vars exist. +# + + +@pytest.mark.asyncio +async def test_system_lookup_paginates_parent_children() -> None: + @dataclass + class R: + name: Annotated[str, notion.TitleProp("Name")] + + schema = await notion.DatabaseSchema.from_class(R, primary_key=["name"]) + client = _FakeSystemClient() + client.child_pages = [ + {"type": "child_database", "id": "db-other"}, + {"type": "child_database", "id": "db-match"}, + ] + client.databases = { + "db-other": { + "title": [{"plain_text": "Other"}], + "data_sources": [{"id": "other-ds"}], + }, + "db-match": { + "title": [{"plain_text": "Wanted"}], + "data_sources": [{"id": "wanted-ds"}], + }, + } + + data_source_id = await _find_or_create_data_source( + client, + _DatabaseSpec( + schema=schema, + managed_by="system", + parent_page_id="parent-page", + title="Wanted", + ), + ) + + assert data_source_id == "wanted-ds" + assert client.created_databases == [] + assert len(client.request_paths) == 2 + assert "start_cursor=cursor-2" in client.request_paths[1] + + +@pytest.mark.asyncio +async def test_system_lookup_matches_data_source_name_when_database_has_many() -> None: + @dataclass + class R: + name: Annotated[str, notion.TitleProp("Name")] + + schema = await notion.DatabaseSchema.from_class(R, primary_key=["name"]) + client = _FakeSystemClient() + client.child_pages = [{"type": "child_database", "id": "db-match"}] + client.databases = { + "db-match": { + "title": [{"plain_text": "People"}], + "data_sources": [ + {"id": "wrong-ds", "name": "Archive"}, + {"id": "right-ds", "name": "People"}, + ], + }, + } + + data_source_id = await _find_or_create_data_source( + client, + _DatabaseSpec( + schema=schema, + managed_by="system", + parent_page_id="parent-page", + title="People", + ), + ) + + assert data_source_id == "right-ds" + + +@pytest.mark.asyncio +async def test_system_lookup_matches_data_source_when_database_title_differs() -> None: + @dataclass + class R: + name: Annotated[str, notion.TitleProp("Name")] + + schema = await notion.DatabaseSchema.from_class(R, primary_key=["name"]) + client = _FakeSystemClient() + client.child_pages = [{"type": "child_database", "id": "db-match"}] + client.databases = { + "db-match": { + "title": [{"plain_text": "Workspace CRM"}], + "data_sources": [ + {"id": "right-ds", "name": "People"}, + ], + }, + } + + data_source_id = await _find_or_create_data_source( + client, + _DatabaseSpec( + schema=schema, + managed_by="system", + parent_page_id="parent-page", + title="People", + ), + ) + + assert data_source_id == "right-ds" + + +@pytest.mark.asyncio +async def test_system_lookup_rejects_duplicate_data_source_names() -> None: + @dataclass + class R: + name: Annotated[str, notion.TitleProp("Name")] + + schema = await notion.DatabaseSchema.from_class(R, primary_key=["name"]) + client = _FakeSystemClient() + client.child_pages = [{"type": "child_database", "id": "db-match"}] + client.databases = { + "db-match": { + "title": [{"plain_text": "Workspace CRM"}], + "data_sources": [ + {"id": "people-a", "name": "People"}, + {"id": "people-b", "name": "People"}, + ], + }, + } + + with pytest.raises(ValueError, match="multiple Notion data sources"): + await _find_or_create_data_source( + client, + _DatabaseSpec( + schema=schema, + managed_by="system", + parent_page_id="parent-page", + title="People", + ), + ) + + +@pytest.mark.asyncio +async def test_system_lookup_rejects_duplicate_data_sources_across_databases() -> None: + @dataclass + class R: + name: Annotated[str, notion.TitleProp("Name")] + + schema = await notion.DatabaseSchema.from_class(R, primary_key=["name"]) + client = _FakeSystemClient() + client.child_pages = [ + {"type": "child_database", "id": "db-a"}, + {"type": "child_database", "id": "db-b"}, + ] + client.databases = { + "db-a": { + "title": [{"plain_text": "First CRM"}], + "data_sources": [{"id": "people-a", "name": "People"}], + }, + "db-b": { + "title": [{"plain_text": "Second CRM"}], + "data_sources": [{"id": "people-b", "name": "People"}], + }, + } + + with pytest.raises(ValueError, match="multiple Notion data sources"): + await _find_or_create_data_source( + client, + _DatabaseSpec( + schema=schema, + managed_by="system", + parent_page_id="parent-page", + title="People", + ), + ) + + +@pytest.mark.asyncio +async def test_system_lookup_rejects_duplicate_single_source_title_fallbacks() -> None: + @dataclass + class R: + name: Annotated[str, notion.TitleProp("Name")] + + schema = await notion.DatabaseSchema.from_class(R, primary_key=["name"]) + client = _FakeSystemClient() + client.child_pages = [ + {"type": "child_database", "id": "db-a"}, + {"type": "child_database", "id": "db-b"}, + ] + client.databases = { + "db-a": { + "title": [{"plain_text": "People"}], + "data_sources": [{"id": "ds-a"}], + }, + "db-b": { + "title": [{"plain_text": "People"}], + "data_sources": [{"id": "ds-b"}], + }, + } + + with pytest.raises(ValueError, match="single-source Notion databases"): + await _find_or_create_data_source( + client, + _DatabaseSpec( + schema=schema, + managed_by="system", + parent_page_id="parent-page", + title="People", + ), + ) + + +@pytest.mark.asyncio +async def test_system_destructive_evolution_patches_type_change() -> None: + @dataclass + class R: + name: Annotated[str, notion.TitleProp("Name")] + email: Annotated[str, notion.RichTextProp("Email")] + + schema = await notion.DatabaseSchema.from_class(R, primary_key=["name"]) + client = _FakeSystemClient() + client.data_source = { + "properties": { + "Name": {"type": "title"}, + "Email": {"type": "email"}, + } + } + spec = _DatabaseSpec( + schema=schema, + managed_by="system", + parent_page_id="parent-page", + title="People", + allow_destructive=True, + ) + + await _evolve_schema_if_needed(client, "ds-1", spec) + + assert client.schema_updates == [{"Email": {"rich_text": {}}}] + + +@pytest.mark.asyncio +async def test_system_destructive_evolution_still_rejects_title_changes() -> None: + @dataclass + class R: + name: Annotated[str, notion.TitleProp("Name")] + + schema = await notion.DatabaseSchema.from_class(R, primary_key=["name"]) + client = _FakeSystemClient() + client.data_source = {"properties": {"Title": {"type": "title"}}} + spec = _DatabaseSpec( + schema=schema, + managed_by="system", + parent_page_id="parent-page", + title="People", + allow_destructive=True, + ) + + with pytest.raises(ValueError, match="title property cannot"): + await _evolve_schema_if_needed(client, "ds-1", spec) + + +@pytest.mark.asyncio +async def test_rate_limit_sleep_releases_concurrency_slot( + monkeypatch: pytest.MonkeyPatch, +) -> None: + second_request_started = asyncio.Event() + first_sleep_started = asyncio.Event() + finish_sleep = asyncio.Event() + session = _FakeSession(second_request_started) + client = notion.NotionClient( + token="fake", max_concurrency=1, session=cast(Any, session) + ) + + async def fake_sleep(delay: float) -> None: + assert delay == 1.0 + first_sleep_started.set() + await finish_sleep.wait() + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + + first = asyncio.create_task(client._request("GET", "/first")) + await first_sleep_started.wait() + second = asyncio.create_task(client._request("GET", "/second")) + await asyncio.wait_for(second_request_started.wait(), timeout=1) + assert await second == {"ok": True} + finish_sleep.set() + assert await first == {"ok": True} + + +# --------------------------------------------------------------------------- +# User-mode end-to-end (require Notion env) +# --------------------------------------------------------------------------- + + +def _user_mode_main( + user_db_id: str, + rows: list[Person], + on_delete: notion.OnDelete = notion.OnDelete.ARCHIVE, +) -> Callable[[], Coroutine[Any, Any, None]]: + """Return an async main fn that mounts a user-mode target and declares ``rows``.""" + + async def app_main() -> None: + target = await notion.mount_database_target( + NOTION_CK, + user_db_id, + await notion.DatabaseSchema.from_class(Person, primary_key=["name"]), + managed_by="user", + on_delete=on_delete, + ) + for r in rows: + target.declare_row(row=r) + + return app_main + + +@requires_notion_env +@pytest.mark.asyncio +async def test_insert_update_archive( + client: notion.NotionClient, + user_db: str, + request: pytest.FixtureRequest, +) -> None: + """End-to-end: insert 3 rows, change one, drop one — see PATCH and archive.""" + env = _make_env(client, request.node.name) + + rows = [ + Person(name="Ada", email="ada@x.com", role="Engineer", active=True), + Person(name="Grace", email="grace@x.com", role="Engineer", active=True), + Person(name="Alan", email="alan@x.com", role="Engineer", active=False), + ] + + # All three steps reuse the same App name so cocoindex's tracking record + # carries across; without that, step 3 wouldn't know Alan was previously + # declared and the archive wouldn't fire. + app_name = "lifecycle" + + # 1. Insert + await coco.App( + coco.AppConfig(name=app_name, environment=env), + _user_mode_main(user_db, rows), + ).update() + pages = await _active_pages(client, user_db) + assert {_title_of(p) for p in pages} == {"Ada", "Grace", "Alan"} + + # 2. Update Ada's email + rows[0] = Person(name="Ada", email="ada@new.com", role="Engineer", active=True) + await coco.App( + coco.AppConfig(name=app_name, environment=env), + _user_mode_main(user_db, rows), + ).update() + pages = await _active_pages(client, user_db) + ada = next(p for p in pages if _title_of(p) == "Ada") + assert ada["properties"]["Email"]["email"] == "ada@new.com" + + # 3. Drop Alan -> page archived + rows.pop() # remove Alan + await coco.App( + coco.AppConfig(name=app_name, environment=env), + _user_mode_main(user_db, rows), + ).update() + pages = await _active_pages(client, user_db) + assert {_title_of(p) for p in pages} == {"Ada", "Grace"} + + +@requires_notion_env +@pytest.mark.asyncio +async def test_on_delete_ignore_leaves_page( + client: notion.NotionClient, + user_db: str, + request: pytest.FixtureRequest, +) -> None: + env = _make_env(client, request.node.name) + rows = [Person(name="Ada", email="ada@x.com", role="Engineer", active=True)] + app_name = "ignore" + await coco.App( + coco.AppConfig(name=app_name, environment=env), + _user_mode_main(user_db, rows, on_delete=notion.OnDelete.IGNORE), + ).update() + assert len(await _active_pages(client, user_db)) == 1 + + # Undeclare — reuse the same app name so cocoindex's prior tracking is + # carried across; otherwise it wouldn't know the row "went missing". + await coco.App( + coco.AppConfig(name=app_name, environment=env), + _user_mode_main(user_db, [], on_delete=notion.OnDelete.IGNORE), + ).update() + # Page is still there — IGNORE doesn't archive. + assert len(await _active_pages(client, user_db)) == 1 + + +@requires_notion_env +@pytest.mark.asyncio +async def test_noop_when_no_changes( + client: notion.NotionClient, + user_db: str, + request: pytest.FixtureRequest, +) -> None: + """Second run with identical data must not touch Notion. + + Concretely: capture last_edited_time after run 1, run 2 with the same rows, + confirm the timestamps are unchanged (no PATCH was issued). + """ + env = _make_env(client, request.node.name) + rows = [ + Person(name="Ada", email="ada@x.com", role="Engineer", active=True), + Person(name="Grace", email="grace@x.com", role="Engineer", active=True), + ] + app_name = "noop" + await coco.App( + coco.AppConfig(name=app_name, environment=env), + _user_mode_main(user_db, rows), + ).update() + timestamps_run1 = { + _title_of(p): p["last_edited_time"] + for p in await _active_pages(client, user_db) + } + + # Re-run with identical rows; cocoindex's fingerprint should short-circuit + # the reconcile and no PATCH should be issued. + await coco.App( + coco.AppConfig(name=app_name, environment=env), + _user_mode_main(user_db, rows), + ).update() + timestamps_run2 = { + _title_of(p): p["last_edited_time"] + for p in await _active_pages(client, user_db) + } + assert timestamps_run1 == timestamps_run2, "no-op run somehow touched the pages" + + +@requires_notion_env +@pytest.mark.asyncio +async def test_on_delete_hard( + client: notion.NotionClient, + user_db: str, + request: pytest.FixtureRequest, +) -> None: + """OnDelete.HARD trashes the page (DELETE /blocks/{id}). Verify it's gone + from active queries (same as archive from the user POV, but the page is + in trash rather than archived).""" + env = _make_env(client, request.node.name) + app_name = "hard" + rows = [Person(name="Ada", email="ada@x.com", role="Engineer", active=True)] + await coco.App( + coco.AppConfig(name=app_name, environment=env), + _user_mode_main(user_db, rows, on_delete=notion.OnDelete.HARD), + ).update() + assert len(await _active_pages(client, user_db)) == 1 + + await coco.App( + coco.AppConfig(name=app_name, environment=env), + _user_mode_main(user_db, [], on_delete=notion.OnDelete.HARD), + ).update() + assert len(await _active_pages(client, user_db)) == 0 + + +@requires_notion_env +@pytest.mark.asyncio +async def test_property_types_roundtrip( + client: notion.NotionClient, + request: pytest.FixtureRequest, +) -> None: + """Title + rich_text + number + url + checkbox + select + date all round-trip + through encode -> Notion -> query -> decode without corruption. + """ + + @dataclass + class AllTypes: + name: Annotated[str, notion.TitleProp("Name")] + notes: Annotated[str, notion.RichTextProp("Notes")] + score: Annotated[float, notion.NumberProp("Score")] + homepage: Annotated[str, notion.UrlProp("Homepage")] + active: Annotated[bool, notion.CheckboxProp("Active")] + role: Annotated[str, notion.SelectProp("Role")] + joined: Annotated[date, notion.DateProp("Joined")] + + assert NOTION_TEST_PARENT_PAGE is not None + title = _unique_title(request.node.name) + ds_id = await _create_test_db( + client, + NOTION_TEST_PARENT_PAGE, + title, + { + "Name": {"title": {}}, + "Notes": {"rich_text": {}}, + "Score": {"number": {}}, + "Homepage": {"url": {}}, + "Active": {"checkbox": {}}, + "Role": {"select": {"options": [{"name": "Engineer"}]}}, + "Joined": {"date": {}}, + }, + ) + try: + env = _make_env(client, request.node.name) + row = AllTypes( + name="Alice", + notes="Likes long walks", + score=3.14, + homepage="https://example.com", + active=True, + role="Engineer", + joined=date(2026, 1, 15), + ) + + async def app_main() -> None: + target = await notion.mount_database_target( + NOTION_CK, + ds_id, + await notion.DatabaseSchema.from_class(AllTypes, primary_key=["name"]), + managed_by="user", + ) + target.declare_row(row=row) + + await coco.App( + coco.AppConfig(name="alltypes", environment=env), app_main + ).update() + + pages = await _active_pages(client, ds_id) + assert len(pages) == 1 + props = pages[0]["properties"] + assert _title_of(pages[0]) == "Alice" + assert ( + "".join(p.get("plain_text", "") for p in props["Notes"]["rich_text"]) + == "Likes long walks" + ) + assert props["Score"]["number"] == 3.14 + assert props["Homepage"]["url"] == "https://example.com" + assert props["Active"]["checkbox"] is True + assert props["Role"]["select"]["name"] == "Engineer" + assert props["Joined"]["date"]["start"] == "2026-01-15" + finally: + await _archive_db(client, ds_id) + + +@requires_notion_env +@pytest.mark.asyncio +async def test_first_run_against_existing_page( + client: notion.NotionClient, + user_db: str, + request: pytest.FixtureRequest, +) -> None: + """If a page with the declared PK already exists in Notion (e.g. the user + pre-seeded it), the first run should PATCH it — not create a duplicate. + Exercises the query-on-miss path returning a hit on first attempt. + """ + # Pre-seed: create a page directly via the API. + await client.create_page( + user_db, + { + "Name": {"title": [{"text": {"content": "Ada"}}]}, + "Email": {"email": "ada@old.com"}, + "Role": {"select": {"name": "Engineer"}}, + "Active": {"checkbox": False}, + }, + ) + assert len(await _active_pages(client, user_db)) == 1 + + env = _make_env(client, request.node.name) + await coco.App( + coco.AppConfig(name="preseed", environment=env), + _user_mode_main( + user_db, + [Person(name="Ada", email="ada@updated.com", role="Engineer", active=True)], + ), + ).update() + pages = await _active_pages(client, user_db) + assert len(pages) == 1, "should have updated the pre-existing page, not duplicated" + assert pages[0]["properties"]["Email"]["email"] == "ada@updated.com" + assert pages[0]["properties"]["Active"]["checkbox"] is True + + +@requires_notion_env +@pytest.mark.asyncio +async def test_schema_validation_type_mismatch( + client: notion.NotionClient, + user_db: str, + request: pytest.FixtureRequest, +) -> None: + """Bug fix #1: declared rich_text on a notion email field → mount fails.""" + env = _make_env(client, request.node.name) + + @dataclass + class WrongPerson: + name: Annotated[str, notion.TitleProp("Name")] + # Email is email type in Notion; we declare it as rich_text → mismatch. + email: Annotated[str, notion.RichTextProp("Email")] + + async def app_main() -> None: + await notion.mount_database_target( + NOTION_CK, + user_db, + await notion.DatabaseSchema.from_class(WrongPerson, primary_key=["name"]), + managed_by="user", + ) + + app = coco.App(coco.AppConfig(name="mismatch", environment=env), app_main) + with pytest.raises(Exception, match="type mismatches"): + await app.update() + + +@requires_notion_env +@pytest.mark.asyncio +async def test_schema_validation_missing_property( + client: notion.NotionClient, + user_db: str, + request: pytest.FixtureRequest, +) -> None: + env = _make_env(client, request.node.name) + + @dataclass + class ExtraField: + name: Annotated[str, notion.TitleProp("Name")] + ghost: Annotated[str, notion.RichTextProp("Ghost")] # not in DS + + async def app_main() -> None: + await notion.mount_database_target( + NOTION_CK, + user_db, + await notion.DatabaseSchema.from_class(ExtraField, primary_key=["name"]), + managed_by="user", + ) + + app = coco.App(coco.AppConfig(name="missing", environment=env), app_main) + with pytest.raises(Exception, match="missing properties"): + await app.update() + + +@requires_notion_env +@pytest.mark.asyncio +async def test_multiple_targets_in_one_app( + client: notion.NotionClient, + request: pytest.FixtureRequest, +) -> None: + """Two ``mount_database_target`` calls in one app sync independently. + + Catches the class of bug where per-target state (page_id cache, locks, + tracking record identity) would accidentally be shared across targets. + Verifies it both at insert time (each target gets its own row) and on + undeclare (dropping rows from one target doesn't affect the other). + """ + assert NOTION_TEST_PARENT_PAGE is not None + title_a = _unique_title(request.node.name + "_a") + title_b = _unique_title(request.node.name + "_b") + ds_a = await _create_test_db( + client, NOTION_TEST_PARENT_PAGE, title_a, PERSON_SCHEMA_PROPS + ) + ds_b = await _create_test_db( + client, NOTION_TEST_PARENT_PAGE, title_b, PERSON_SCHEMA_PROPS + ) + try: + env = _make_env(client, request.node.name) + rows_a = [Person(name="A1", email="a1@x.com", role="Engineer", active=True)] + rows_b = [Person(name="B1", email="b1@x.com", role="Engineer", active=True)] + + async def app_main() -> None: + schema = await notion.DatabaseSchema.from_class( + Person, primary_key=["name"] + ) + target_a = await coco.use_mount( + coco.component_subpath("setup", "target_a"), + notion.declare_database_target, + NOTION_CK, + ds_a, + schema, + managed_by="user", + ) + target_b = await coco.use_mount( + coco.component_subpath("setup", "target_b"), + notion.declare_database_target, + NOTION_CK, + ds_b, + schema, + managed_by="user", + ) + for r in rows_a: + target_a.declare_row(row=r) + for r in rows_b: + target_b.declare_row(row=r) + + app_name = "multitarget" + + # Step 1: each target gets its own row. + await coco.App( + coco.AppConfig(name=app_name, environment=env), app_main + ).update() + assert {_title_of(p) for p in await _active_pages(client, ds_a)} == {"A1"} + assert {_title_of(p) for p in await _active_pages(client, ds_b)} == {"B1"} + + # Step 2: drop A1 only. ds_a should empty out; ds_b should keep B1. + rows_a.clear() + await coco.App( + coco.AppConfig(name=app_name, environment=env), app_main + ).update() + assert {_title_of(p) for p in await _active_pages(client, ds_a)} == set() + assert {_title_of(p) for p in await _active_pages(client, ds_b)} == {"B1"} + finally: + await _archive_db(client, ds_a) + await _archive_db(client, ds_b) + + +# --------------------------------------------------------------------------- +# System mode +# --------------------------------------------------------------------------- + + +@requires_notion_env +@pytest.mark.asyncio +async def test_system_creates_and_evolves( + client: notion.NotionClient, + request: pytest.FixtureRequest, +) -> None: + """First run: DS doesn't exist -> connector creates it. + Second run with extended schema: connector PATCHes the new property. + """ + assert NOTION_TEST_PARENT_PAGE is not None + title = _unique_title(request.node.name) + env = _make_env(client, request.node.name) + + async def app_create() -> None: + target = await notion.mount_database_target( + NOTION_CK, + schema=await notion.DatabaseSchema.from_class(Person, primary_key=["name"]), + managed_by="system", + parent_page_id=NOTION_TEST_PARENT_PAGE, + title=title, + ) + target.declare_row( + row=Person(name="Seed", email="seed@x.com", role="Engineer", active=True) + ) + + created_db_id: str | None = None + try: + await coco.App(coco.AppConfig(name="s1", environment=env), app_create).update() + + # Find the DS the connector just created by enumerating children. + children = await client._request( + "GET", f"/blocks/{NOTION_TEST_PARENT_PAGE}/children?page_size=100" + ) + ds_id: str | None = None + for c in children.get("results", []): + if c.get("type") != "child_database": + continue + db = await client.get_database(c["id"]) + if _title_of_db(db) == title: + created_db_id = c["id"] + ds_id = (db.get("data_sources") or [{}])[0].get("id") + break + assert ds_id is not None, "system mode should have created the DS" + ds = await client.get_data_source(ds_id) + assert set(ds["properties"].keys()) >= {"Name", "Email", "Role", "Active"} + assert len(await _active_pages(client, ds_id)) == 1 + + # Now extend the schema with a Notes column and re-run. + async def app_evolve() -> None: + target = await notion.mount_database_target( + NOTION_CK, + schema=await notion.DatabaseSchema.from_class( + PersonWithNotes, primary_key=["name"] + ), + managed_by="system", + parent_page_id=NOTION_TEST_PARENT_PAGE, + title=title, + ) + target.declare_row( + row=PersonWithNotes( + name="Seed", + email="seed@x.com", + role="Engineer", + active=True, + notes="updated", + ) + ) + + await coco.App(coco.AppConfig(name="s2", environment=env), app_evolve).update() + ds = await client.get_data_source(ds_id) + assert "Notes" in ds["properties"] + assert ds["properties"]["Notes"]["type"] == "rich_text" + finally: + if created_db_id: + try: + await client._request( + "PATCH", f"/databases/{created_db_id}", {"in_trash": True} + ) + except Exception: + pass + + +def _title_of_db(db: dict[str, Any]) -> str: + parts = db.get("title") or [] + return "".join(p.get("plain_text", "") for p in parts) diff --git a/uv.lock b/uv.lock index 37ccfd780..51e949967 100644 --- a/uv.lock +++ b/uv.lock @@ -632,6 +632,9 @@ litellm = [ neo4j = [ { name = "neo4j" }, ] +notion = [ + { name = "aiohttp" }, +] oci = [ { name = "oci" }, ] @@ -722,6 +725,7 @@ requires-dist = [ { name = "aiobotocore", marker = "extra == 'amazon-s3'", specifier = ">=2.0.0" }, { name = "aiohttp", marker = "extra == 'all'", specifier = ">=3.9.0" }, { name = "aiohttp", marker = "extra == 'doris'", specifier = ">=3.9.0" }, + { name = "aiohttp", marker = "extra == 'notion'", specifier = ">=3.9.0" }, { name = "aiomysql", marker = "extra == 'all'", specifier = ">=0.2.0" }, { name = "aiomysql", marker = "extra == 'doris'", specifier = ">=0.2.0" }, { name = "apache-iggy", marker = "extra == 'all'", specifier = ">=0.8.0" }, @@ -780,7 +784,7 @@ requires-dist = [ { name = "typing-extensions", specifier = ">=4.12" }, { name = "watchdog", specifier = ">=6.0.0" }, ] -provides-extras = ["all", "amazon-s3", "colpali", "doris", "entity-resolution", "entity-resolution-llm", "falkordb", "google-drive", "iggy", "kafka", "lancedb", "litellm", "neo4j", "oci", "postgres", "qdrant", "sentence-transformers", "sqlite", "surrealdb", "turbopuffer"] +provides-extras = ["all", "amazon-s3", "colpali", "doris", "entity-resolution", "entity-resolution-llm", "falkordb", "google-drive", "iggy", "kafka", "lancedb", "litellm", "neo4j", "notion", "oci", "postgres", "qdrant", "sentence-transformers", "sqlite", "surrealdb", "turbopuffer"] [package.metadata.requires-dev] build-test = [