106 lines
2.7 KiB
Python
106 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from .config import BACKEND_MODE
|
|
from .csharp_sidecar import call_sidecar_action
|
|
|
|
|
|
def _require_hybrid_mode() -> None:
|
|
if BACKEND_MODE != "csharp-hybrid":
|
|
raise RuntimeError("Fragments backend requires JOURNAL_BACKEND_MODE=csharp-hybrid.")
|
|
|
|
|
|
def list_fragments() -> list[dict[str, Any]]:
|
|
_require_hybrid_mode()
|
|
result = call_sidecar_action("fragments.list")
|
|
if not isinstance(result, list):
|
|
return []
|
|
return [item for item in result if isinstance(item, dict)]
|
|
|
|
|
|
def create_fragment(
|
|
type_: str,
|
|
description: str,
|
|
tags: list[str] | None = None,
|
|
time: str | None = None,
|
|
) -> dict[str, Any]:
|
|
_require_hybrid_mode()
|
|
payload: dict[str, Any] = {
|
|
"type": type_,
|
|
"description": description,
|
|
}
|
|
if time:
|
|
payload["time"] = time
|
|
if tags:
|
|
payload["tags"] = tags
|
|
result = call_sidecar_action("fragments.create", payload=payload)
|
|
return result if isinstance(result, dict) else {"result": result}
|
|
|
|
|
|
def get_fragment(fragment_id: str) -> dict[str, Any] | None:
|
|
_require_hybrid_mode()
|
|
result = call_sidecar_action(
|
|
"fragments.get",
|
|
command_fields={"id": fragment_id},
|
|
)
|
|
return result if isinstance(result, dict) else None
|
|
|
|
|
|
def update_fragment(
|
|
fragment_id: str,
|
|
*,
|
|
type_: str | None = None,
|
|
description: str | None = None,
|
|
time: str | None = None,
|
|
tags: list[str] | None = None,
|
|
clear_tags: bool = False,
|
|
) -> bool:
|
|
_require_hybrid_mode()
|
|
payload: dict[str, Any] = {}
|
|
if type_ is not None:
|
|
payload["type"] = type_
|
|
if description is not None:
|
|
payload["description"] = description
|
|
if time is not None:
|
|
payload["time"] = time
|
|
if clear_tags:
|
|
payload["tags"] = []
|
|
elif tags is not None:
|
|
payload["tags"] = tags
|
|
|
|
if not payload:
|
|
raise ValueError(
|
|
"No update fields provided. Use --type/--description/--time/--tag/--clear-tags."
|
|
)
|
|
|
|
result = call_sidecar_action(
|
|
"fragments.update",
|
|
payload=payload,
|
|
command_fields={"id": fragment_id},
|
|
)
|
|
return bool(result)
|
|
|
|
|
|
def delete_fragment(fragment_id: str) -> bool:
|
|
_require_hybrid_mode()
|
|
result = call_sidecar_action(
|
|
"fragments.delete",
|
|
command_fields={"id": fragment_id},
|
|
)
|
|
return bool(result)
|
|
|
|
|
|
def search_fragments(type_: str | None = None, tag: str | None = None) -> list[dict[str, Any]]:
|
|
_require_hybrid_mode()
|
|
result = call_sidecar_action(
|
|
"fragments.search",
|
|
command_fields={
|
|
"type": type_,
|
|
"tag": tag,
|
|
},
|
|
)
|
|
if not isinstance(result, list):
|
|
return []
|
|
return [item for item in result if isinstance(item, dict)]
|