feat(persistence): ServiceAccountRepository (Stage 1 PR1)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/packages/harness/deerflow/persistence/service_account/__init__.py

@@ -13,5 +13,6 @@ upgrade live in Stage 1 alongside the headless API surface.
 from __future__ import annotations
 
 from deerflow.persistence.service_account.model import ServiceAccountRow
+from deerflow.persistence.service_account.sql import ServiceAccountRepository
 
-__all__ = ["ServiceAccountRow"]
+__all__ = ["ServiceAccountRepository", "ServiceAccountRow"]

backend/packages/harness/deerflow/persistence/service_account/sql.py

@@ -0,0 +1,92 @@
+"""SQLAlchemy-backed service account repository (Stage 1 PR1).
+
+Mirrors :class:`WorkspaceRepository`: fresh session per method,
+``_row_to_dict`` static helper. Workspace scoping is enforced by the
+caller (route layer reads the workspace contextvar); the repository
+takes ``workspace_id`` explicitly.
+"""
+
+from __future__ import annotations
+
+import uuid
+from datetime import UTC, datetime
+from typing import Any
+
+from sqlalchemy import select, update
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from deerflow.persistence.service_account.model import ServiceAccountRow
+
+_VALID_STATUSES = frozenset({"active", "suspended", "deleted"})
+
+
+class ServiceAccountRepository:
+    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
+        self._sf = session_factory
+
+    @staticmethod
+    def _row_to_dict(row: ServiceAccountRow) -> dict[str, Any]:
+        return {
+            "id": row.id,
+            "workspace_id": row.workspace_id,
+            "name": row.name,
+            "role": row.role,
+            "identity_mode": row.identity_mode,
+            "status": row.status,
+            "created_by": row.created_by,
+            "created_at": row.created_at.isoformat() if row.created_at else None,
+            "updated_at": row.updated_at.isoformat() if row.updated_at else None,
+        }
+
+    async def create(
+        self,
+        *,
+        workspace_id: str,
+        name: str,
+        created_by: str,
+        role: str = "member",
+        identity_mode: str = "collapsed",
+        status: str = "active",
+    ) -> dict[str, Any]:
+        now = datetime.now(UTC)
+        row = ServiceAccountRow(
+            id=str(uuid.uuid4()),
+            workspace_id=workspace_id,
+            name=name,
+            role=role,
+            identity_mode=identity_mode,
+            status=status,
+            created_by=created_by,
+            created_at=now,
+            updated_at=now,
+        )
+        async with self._sf() as session:
+            session.add(row)
+            await session.commit()
+            await session.refresh(row)
+            return self._row_to_dict(row)
+
+    async def get(self, sa_id: str) -> dict[str, Any] | None:
+        async with self._sf() as session:
+            row = await session.get(ServiceAccountRow, sa_id)
+            return self._row_to_dict(row) if row else None
+
+    async def get_active(self, sa_id: str) -> dict[str, Any] | None:
+        """Return the row only when ``status == 'active'`` (auth hot path)."""
+        async with self._sf() as session:
+            row = await session.get(ServiceAccountRow, sa_id)
+            if row is None or row.status != "active":
+                return None
+            return self._row_to_dict(row)
+
+    async def list_by_workspace(self, workspace_id: str) -> list[dict[str, Any]]:
+        async with self._sf() as session:
+            result = await session.execute(select(ServiceAccountRow).where(ServiceAccountRow.workspace_id == workspace_id).order_by(ServiceAccountRow.created_at.desc()))
+            return [self._row_to_dict(r) for r in result.scalars()]
+
+    async def update_status(self, sa_id: str, status: str) -> None:
+        if status not in _VALID_STATUSES:
+            raise ValueError(f"status {status!r} not in {_VALID_STATUSES!r}")
+        async with self._sf() as session:
+            await session.execute(update(ServiceAccountRow).where(ServiceAccountRow.id == sa_id).values(status=status, updated_at=datetime.now(UTC)))
+            await session.commit()