daniel156161
e115698e92
Build and Push Docker Container / build-and-push (push) Successful in 3m31s
- Serialize property set and remove commands per vault note to avoid lost frontmatter writes. - Wait briefly for successful property writes to hit disk before releasing the note lock. - Return property path, name, value, type, operation, and ok status in n8n outputs. - Include property details on continue-on-fail errors so failed items are visible in n8n. - Bump obsidian-api to 1.0.7 and n8n-nodes-obsidian-cli-api to 0.2.8. - Add regression coverage for concurrent property writes to the same note.
118 lines
3.8 KiB
Python
118 lines
3.8 KiB
Python
import asyncio, uuid, time
|
|
from pathlib import Path
|
|
|
|
from starlette.responses import JSONResponse
|
|
|
|
from app.config import CLI_BIN
|
|
from app.command_target import extract_command_target
|
|
from app.property_commands import arg_value, prepare_property_args, vault_note_path
|
|
from app.vaults import vault_root_for_target
|
|
|
|
def error(status_code:int, detail:str) -> JSONResponse:
|
|
return JSONResponse({"detail": detail}, status_code=status_code)
|
|
|
|
_PROPERTY_WRITE_LOCKS:dict[str, asyncio.Lock] = {}
|
|
|
|
async def execute_command(args:list[str], cwd:Path, stdin:str|None, timeout:float) -> JSONResponse:
|
|
command_id = str(uuid.uuid4())
|
|
try:
|
|
args = prepare_property_args(args, cwd)
|
|
except PermissionError as exc:
|
|
return error(403, str(exc))
|
|
except OSError as exc:
|
|
return error(500, f"failed to prepare property command: {exc}")
|
|
|
|
command = [CLI_BIN, *args]
|
|
started = time.perf_counter()
|
|
|
|
try:
|
|
lock_key = property_write_lock_key(args, cwd)
|
|
if lock_key:
|
|
lock = _PROPERTY_WRITE_LOCKS.setdefault(lock_key, asyncio.Lock())
|
|
async with lock:
|
|
target_file = property_write_file(args, cwd)
|
|
before_signature = file_signature(target_file)
|
|
stdout_b, stderr_b, exit_code, timed_out = await run_process(command, cwd, stdin, timeout)
|
|
if exit_code == 0 and not timed_out:
|
|
await wait_for_file_change(target_file, before_signature)
|
|
else:
|
|
stdout_b, stderr_b, exit_code, timed_out = await run_process(command, cwd, stdin, timeout)
|
|
except FileNotFoundError:
|
|
return error(500, f"configured CLI binary was not found: {CLI_BIN}")
|
|
|
|
duration_ms = int((time.perf_counter() - started) * 1000)
|
|
stdout = stdout_b.decode(errors="replace").rstrip()
|
|
stderr = stderr_b.decode(errors="replace").rstrip()
|
|
if timed_out:
|
|
exit_code = 124
|
|
stderr = (stderr + f"\nCommand timed out after {timeout}s").strip()
|
|
|
|
return JSONResponse(
|
|
{
|
|
"id": command_id,
|
|
"command": command,
|
|
"cwd": str(cwd),
|
|
"exit_code": exit_code,
|
|
"stdout": stdout,
|
|
"stderr": stderr,
|
|
"duration_ms": duration_ms,
|
|
"timed_out": timed_out,
|
|
}
|
|
)
|
|
|
|
async def run_process(command:list[str], cwd:Path, stdin:str|None, timeout:float) -> tuple[bytes, bytes, int, bool]:
|
|
try:
|
|
process = await asyncio.create_subprocess_exec(
|
|
*command,
|
|
cwd=str(cwd),
|
|
stdin=asyncio.subprocess.PIPE if stdin is not None else None,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
)
|
|
except FileNotFoundError:
|
|
raise
|
|
|
|
try:
|
|
stdout_b, stderr_b = await asyncio.wait_for(
|
|
process.communicate(None if stdin is None else stdin.encode()),
|
|
timeout=timeout,
|
|
)
|
|
timed_out = False
|
|
except asyncio.TimeoutError:
|
|
process.kill()
|
|
stdout_b, stderr_b = await process.communicate()
|
|
timed_out = True
|
|
|
|
exit_code = process.returncode if process.returncode is not None else -1
|
|
return stdout_b, stderr_b, exit_code, timed_out
|
|
|
|
def property_write_lock_key(args:list[str], cwd:Path) -> str|None:
|
|
target = extract_command_target(args)
|
|
if target.command not in {"property:set", "property:remove"}:
|
|
return None
|
|
if not target.paths:
|
|
return None
|
|
vault = target.vault or str(cwd.resolve())
|
|
return f"{vault}:{target.paths[0]}"
|
|
|
|
def property_write_file(args:list[str], cwd:Path) -> Path|None:
|
|
target = extract_command_target(args)
|
|
path = arg_value(args, "path")
|
|
if not path:
|
|
return None
|
|
return vault_note_path(vault_root_for_target(target, cwd), path)
|
|
|
|
def file_signature(path:Path|None) -> tuple[int, int]|None:
|
|
if path is None or not path.exists():
|
|
return None
|
|
stat = path.stat()
|
|
return stat.st_mtime_ns, stat.st_size
|
|
|
|
async def wait_for_file_change(path:Path|None, before:tuple[int, int]|None) -> None:
|
|
if path is None:
|
|
return
|
|
for _ in range(40):
|
|
if file_signature(path) != before:
|
|
return
|
|
await asyncio.sleep(0.05)
|