deerflow-code/offline-backend-20260512/backend/packages/harness/deerflow/persistence/light_apps/base.py
2026-09-07 18:24:55 +08:00

47 lines
1.5 KiB
Python

"""Abstract light-app registry persistence API.
Global / shared store: no ``user_id`` scoping. Reads are open to any caller; the
router gates writes behind admin. ``created_by`` / ``updated_by`` are audit-only.
"""
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import Any
class DuplicateLightAppNameError(ValueError):
"""Raised when creating/updating an app would collide on the unique name."""
class LightAppStore(ABC):
@abstractmethod
async def list_apps(
self,
*,
app_type: str | None = None,
mount_location: str | None = None,
enabled_only: bool = False,
) -> list[dict[str, Any]]:
"""Return registered apps (newest first), optionally filtered."""
raise NotImplementedError
@abstractmethod
async def get_app(self, app_id: str) -> dict[str, Any] | None:
raise NotImplementedError
@abstractmethod
async def create_app(self, data: dict[str, Any]) -> dict[str, Any]:
"""Create an app. Raises ``DuplicateLightAppNameError`` on a name clash."""
raise NotImplementedError
@abstractmethod
async def update_app(self, app_id: str, data: dict[str, Any]) -> dict[str, Any] | None:
"""Partial update. Returns None if the app does not exist. Raises
``DuplicateLightAppNameError`` when renaming onto an existing name."""
raise NotImplementedError
@abstractmethod
async def delete_app(self, app_id: str) -> bool:
raise NotImplementedError