From 5093d3d123d8e918573241a55e53e9b7fdc66db9 Mon Sep 17 00:00:00 2001 From: 1445043649 <> Date: Sun, 28 Jun 2026 11:26:34 +0800 Subject: [PATCH] feat(persistence): ExternalUserRepository scaffold (Stage 1 PR1) Co-Authored-By: Claude Opus 4.8 (1M context) --- .../persistence/external_user/__init__.py | 3 +- .../deerflow/persistence/external_user/sql.py | 98 +++++++++++++++++++ backend/tests/test_external_user_repo.py | 77 +++++++++++++++ 3 files changed, 177 insertions(+), 1 deletion(-) create mode 100644 backend/packages/harness/deerflow/persistence/external_user/sql.py create mode 100644 backend/tests/test_external_user_repo.py diff --git a/backend/packages/harness/deerflow/persistence/external_user/__init__.py b/backend/packages/harness/deerflow/persistence/external_user/__init__.py index 3a93f6b4..62d5d472 100644 --- a/backend/packages/harness/deerflow/persistence/external_user/__init__.py +++ b/backend/packages/harness/deerflow/persistence/external_user/__init__.py @@ -12,5 +12,6 @@ header parsing, and quota attribution all live in Stage 1. from __future__ import annotations from deerflow.persistence.external_user.model import ExternalUserRow +from deerflow.persistence.external_user.sql import ExternalUserRepository -__all__ = ["ExternalUserRow"] +__all__ = ["ExternalUserRepository", "ExternalUserRow"] diff --git a/backend/packages/harness/deerflow/persistence/external_user/sql.py b/backend/packages/harness/deerflow/persistence/external_user/sql.py new file mode 100644 index 00000000..bb27d66e --- /dev/null +++ b/backend/packages/harness/deerflow/persistence/external_user/sql.py @@ -0,0 +1,98 @@ +"""SQLAlchemy-backed external user repository (Stage 1 PR1). + +Built but NOT yet wired to any auth path — the X-External-User-Id +passthrough that calls ``upsert`` lands in a later track-2 PR. ``upsert`` +is idempotent on (service_account_id, external_id). +""" + +from __future__ import annotations + +import uuid +from datetime import UTC, datetime +from typing import Any + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from deerflow.persistence.external_user.model import ExternalUserRow + + +class ExternalUserRepository: + def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None: + self._sf = session_factory + + @staticmethod + def _row_to_dict(row: ExternalUserRow) -> dict[str, Any]: + return { + "id": row.id, + "workspace_id": row.workspace_id, + "service_account_id": row.service_account_id, + "external_id": row.external_id, + "display_name": row.display_name, + "metadata": dict(row.metadata_json or {}), + "created_at": row.created_at.isoformat() if row.created_at else None, + "last_seen_at": row.last_seen_at.isoformat() if row.last_seen_at else None, + } + + async def get(self, external_user_id: str) -> dict[str, Any] | None: + async with self._sf() as session: + row = await session.get(ExternalUserRow, external_user_id) + return self._row_to_dict(row) if row else None + + async def get_by_external_id(self, *, service_account_id: str, external_id: str) -> dict[str, Any] | None: + async with self._sf() as session: + result = await session.execute( + select(ExternalUserRow).where( + ExternalUserRow.service_account_id == service_account_id, + ExternalUserRow.external_id == external_id, + ) + ) + row = result.scalar_one_or_none() + return self._row_to_dict(row) if row else None + + async def list_by_workspace(self, workspace_id: str) -> list[dict[str, Any]]: + async with self._sf() as session: + result = await session.execute(select(ExternalUserRow).where(ExternalUserRow.workspace_id == workspace_id).order_by(ExternalUserRow.created_at.desc())) + return [self._row_to_dict(r) for r in result.scalars()] + + async def upsert( + self, + *, + workspace_id: str, + service_account_id: str, + external_id: str, + display_name: str | None = None, + metadata: dict[str, Any] | None = None, + ) -> dict[str, Any]: + """Insert a new external user or refresh ``last_seen_at`` on an + existing (service_account_id, external_id) row.""" + now = datetime.now(UTC) + async with self._sf() as session: + result = await session.execute( + select(ExternalUserRow).where( + ExternalUserRow.service_account_id == service_account_id, + ExternalUserRow.external_id == external_id, + ) + ) + row = result.scalar_one_or_none() + if row is None: + row = ExternalUserRow( + id=str(uuid.uuid4()), + workspace_id=workspace_id, + service_account_id=service_account_id, + external_id=external_id, + display_name=display_name, + metadata_json=metadata or {}, + created_at=now, + last_seen_at=now, + ) + session.add(row) + else: + row.last_seen_at = now + if display_name is not None: + row.display_name = display_name + if metadata is not None: + row.metadata_json = metadata + await session.commit() + await session.refresh(row) + return self._row_to_dict(row) diff --git a/backend/tests/test_external_user_repo.py b/backend/tests/test_external_user_repo.py new file mode 100644 index 00000000..a17646eb --- /dev/null +++ b/backend/tests/test_external_user_repo.py @@ -0,0 +1,77 @@ +"""Tests for ExternalUserRepository (Stage 1 PR1). + +Repository is built but not yet wired to any auth path. upsert is +idempotent on (service_account_id, external_id) per the table's +UniqueConstraint uq_external_users_sa_external. +""" + +from __future__ import annotations + +import pytest + +from deerflow.persistence.external_user import ExternalUserRepository +from deerflow.persistence.service_account.model import ServiceAccountRow +from deerflow.persistence.user.model import UserRow +from deerflow.persistence.workspace.model import WorkspaceRow + +pytestmark = pytest.mark.anyio + + +@pytest.fixture +def anyio_backend() -> str: + return "asyncio" + + +async def _make_repo(tmp_path): + from deerflow.persistence.engine import get_session_factory, init_engine + + url = f"sqlite+aiosqlite:///{tmp_path / 'test.db'}" + await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path)) + return ExternalUserRepository(get_session_factory()) + + +async def _cleanup(): + from deerflow.persistence.engine import close_engine + + await close_engine() + + +async def _seed_sa(repo) -> None: + async with repo._sf() as session: + session.add(UserRow(id="u-alice", email="alice@example.com")) + await session.commit() + async with repo._sf() as session: + session.add(WorkspaceRow(id="w-1", name="WS", slug="ws", owner_id="u-alice")) + await session.commit() + async with repo._sf() as session: + session.add(ServiceAccountRow(id="sa-1", workspace_id="w-1", name="bot", role="member", identity_mode="external_passthrough", status="active", created_by="u-alice")) + await session.commit() + + +async def test_upsert_inserts_then_updates_same_row(tmp_path): + repo = await _make_repo(tmp_path) + try: + await _seed_sa(repo) + first = await repo.upsert(workspace_id="w-1", service_account_id="sa-1", external_id="ext-42", display_name="alice") + assert first["external_id"] == "ext-42" + assert first["last_seen_at"] is not None + second = await repo.upsert(workspace_id="w-1", service_account_id="sa-1", external_id="ext-42") + # same logical row (no duplicate) + assert second["id"] == first["id"] + rows = await repo.list_by_workspace("w-1") + assert len(rows) == 1 + finally: + await _cleanup() + + +async def test_get_by_external_id(tmp_path): + repo = await _make_repo(tmp_path) + try: + await _seed_sa(repo) + await repo.upsert(workspace_id="w-1", service_account_id="sa-1", external_id="ext-7") + found = await repo.get_by_external_id(service_account_id="sa-1", external_id="ext-7") + assert found is not None + assert found["external_id"] == "ext-7" + assert await repo.get_by_external_id(service_account_id="sa-1", external_id="nope") is None + finally: + await _cleanup()