Sync
sync ¶
Utilities to sync other services with Notion.
ID: TypeAlias = str module-attribute ¶
Self = TypeVar('Self', bound='SyncTask') module-attribute ¶
all_tasks: set[SyncTask] = set() module-attribute ¶
All tasks that have been created so far.
ConflictMode ¶
State ¶
The state of a sync task.
The state holds the synched 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 synched 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 synched 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.