Skip to content

Sync

sync

Utilities to synchronize other services with Notion.

ID: TypeAlias = str module-attribute

all_tasks: set[SyncTask] = set() module-attribute

All tasks that have been created so far.

ConflictMode

Conflict resolution modes.

ERROR = 'error' class-attribute instance-attribute

NEWER = 'newer' class-attribute instance-attribute

NOTION = 'notion' class-attribute instance-attribute

OTHER = 'other' class-attribute instance-attribute

State

The state of a sync task.

The state holds the synced objects and their attributes as a dictionary of Notion attributes.

ids: dict[ID, ID] = Field(default_factory=dict) class-attribute instance-attribute

Maps Notion object ids to other object ids.

objs: dict[ID, dict[str, Any]] = Field(default_factory=dict) class-attribute instance-attribute

Dictionary of Notion objects synced with other service and indexed by their ids.

SyncTask(name: str, attr_map: dict[str, str], conflict_mode: ConflictMode | str = ConflictMode.NEWER)

A task to be performed during a sync.

This is an abstract base class to allow for different kinds of sync tasks.

attr_map = attr_map instance-attribute

conflict_mode = conflict_mode instance-attribute

name = name instance-attribute

state_path = get_cfg().ultimate_notion.sync_state_dir / f'{name}.pickle' instance-attribute

__await__()

Delegate the await to the call method

__call__() async

Run the task as scheduled.

get_notion_objects() -> list[Any] abstractmethod

Get all Notion objects to sync.

get_other_objects() -> list[Any] abstractmethod

Get all other objects to sync.

in_total(times: int) -> Self

Schedule the task to run a total of so many times.

initial_sync(notion_objs: dict[ID, Any], other_objs: dict[ID, Any]) -> State

Make the initial state.

This is a two-way sync, i.e. the objects are compared and the differences are resolved.

notion_create_obj(**kwargs: Any) -> Any abstractmethod

Create a new page.

notion_delete_obj(obj: Any) -> None abstractmethod

Delete the page.

notion_hash(obj: Any) -> str abstractmethod

Get the hash of the Notion object for object mapping/linking.

notion_id(obj: Any) -> ID abstractmethod

Get the id of the Notion object.

notion_timestamp(obj: Any) -> datetime abstractmethod

Get the timestamp of the Notion object.

notion_to_dict(obj: Any) -> dict[str, Any] abstractmethod

Convert a Notion object to a dictionary.

notion_update_obj(obj: Any, attr: str, value: Any) -> None abstractmethod

Set an attribute of the Notion object, e.g. page.

other_create_obj(**kwargs: Any) -> Any abstractmethod

Create a new other object.

other_delete_obj(obj: Any) -> None abstractmethod

Delete the other object.

other_hash(obj: Any) -> str abstractmethod

Get the hash of the other object for object mapping/linking.

other_id(obj: Any) -> ID abstractmethod

Get the id of the other object.

other_timestamp(obj: Any) -> datetime abstractmethod

Get the timestamp of the other object.

other_to_dict(obj: Any) -> dict[str, Any] abstractmethod

Convert another object to a dictionary.

other_update_obj(obj: Any, attr: str, value: Any) -> None abstractmethod

Set an attribute of the other object.

resolve_conflict(notion_obj: Any, other_obj: Any, notion_attr: str, other_attr: str) -> Any

Resolve a conflict between two objects on an attribute.

run_every(hours: int = 0, minutes: int = 0, seconds: int = 0) -> Self

Schedule the task to run every so many seconds.

run_once() -> Self

schedule() -> Self

Apply the task.

sync(state: State | None) -> State

The actual sync operation.

The state holds the synced objects and their attributes as a dictionary of Notion attributes.

sync_notion_created(state: State, notion_objs: dict[ID, Any]) -> State

Sync an object not in the state and created in Notion.

sync_notion_deleted(state: State, notion_objs: dict[ID, Any], other_objs: dict[ID, Any]) -> State

Sync an object in the state that was deleted in Notion.

sync_other_created(state: State, other_objs: dict[ID, Any]) -> State

Sync an object not in the state and created in other service.

sync_other_deleted(state: State, notion_objs: dict[ID, Any], other_objs: dict[ID, Any]) -> State

Sync an object in the state that was deleted in the other service.

sync_state_changes(state: State, notion_objs: dict[ID, Any], other_objs: dict[ID, Any]) -> State

Sync changes with respect to the state and update the state.

This is a three-way sync, i.e. the objects are compared to the state as base and the differences are resolved.

run_all_tasks(*, debug: bool | None = None)

Run all scheduled tasks.