mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-29 09:26:00 +00:00
refactor(feedback): move the secondary adapters to app/adapters
Group them by bounded context instead of by technology, one file per
port, and align the directory name with the AWS Prescriptive Guidance
layout (entrypoints / domain-with-ports / adapters).
app/infra/persistence/feedback.py
-> app/adapters/feedback/feedback_repository.py owned persistence
-> app/adapters/feedback/run_lookup.py anti-corruption layer
`persistence/` promised a technology-first classification that its own
contents contradicted: RunStoreRunLookup lived there while its docstring
said "no new SQL". Splitting per port makes that distinction structural.
SqlFeedbackRepository and _tz_aware move unchanged -- verified by
comparing their AST against the original rather than by eye. run_lookup.py
additionally gains a RunStore annotation behind TYPE_CHECKING (the module
is imported lazily by the composition root, so this keeps the runtime
import cost at zero), a docstring stating that this context owns no table
and writes no SQL against it, and a TODO recording the condition under
which the body is replaced: when the run context publishes a contract of
its own, the RunLookup port itself does not move.
Each module docstring opens with a fixed marker so the two kinds of
secondary adapter stay greppable:
grep -rl "anti-corruption layer" app/adapters/
Filenames deliberately carry no sql_ / acl_ prefix: a prefix encodes an
implementation property, so switching storage would force a rename even
though the port -- and therefore the import path -- has not changed. The
class name already carries it. A prefix earns its place once one port has
several production implementations, which is not yet the case here.
app/infra/ held nothing else and is removed.
This commit is contained in:
parent
0137722a29
commit
cb49dd67dc
@ -1,11 +1,15 @@
|
|||||||
"""SQL adapters for the feedback bounded context.
|
"""Secondary adapter (owned persistence) -- the feedback table in SQL.
|
||||||
|
|
||||||
Secondary adapters implementing the ports declared in
|
Implements ``FeedbackRepository`` from ``deerflow.domain.feedback.ports``.
|
||||||
``deerflow.domain.feedback.ports``. SQL/ORM vocabulary stops at this
|
This context owns the ``feedback`` table and writes its own queries, so
|
||||||
file: methods exchange domain objects, translate ``IntegrityError`` into
|
SQL/ORM vocabulary stops at this file: methods exchange domain objects,
|
||||||
domain errors, and normalize SQLite's tz-naive reads. Queries were
|
translate ``IntegrityError`` into domain errors, and normalize SQLite's
|
||||||
migrated unchanged from the legacy repository (now removed) to preserve
|
tz-naive reads. Queries were migrated unchanged from the legacy repository
|
||||||
behavior.
|
(now removed) to preserve behavior.
|
||||||
|
|
||||||
|
Sibling adapter: ``run_lookup.py`` serves the same context but owns no
|
||||||
|
table and writes no SQL -- see its docstring for why that distinction is
|
||||||
|
worth keeping visible.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
@ -17,10 +21,10 @@ from sqlalchemy.exc import IntegrityError
|
|||||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
||||||
|
|
||||||
from deerflow.domain.feedback.model import DuplicateFeedbackError, Feedback
|
from deerflow.domain.feedback.model import DuplicateFeedbackError, Feedback
|
||||||
from deerflow.domain.feedback.ports import FeedbackRepository, RunLookup
|
from deerflow.domain.feedback.ports import FeedbackRepository
|
||||||
|
|
||||||
# Transitional: the ORM row stays in the harness until PR-N moves engine,
|
# Transitional: the ORM row stays in the harness until PR-N moves engine,
|
||||||
# models, and migrations into app/infra.
|
# models, and migrations into app/adapters.
|
||||||
from deerflow.persistence.feedback.model import FeedbackRow
|
from deerflow.persistence.feedback.model import FeedbackRow
|
||||||
|
|
||||||
|
|
||||||
@ -137,18 +141,3 @@ class SqlFeedbackRepository(FeedbackRepository):
|
|||||||
await session.delete(row)
|
await session.delete(row)
|
||||||
await session.commit()
|
await session.commit()
|
||||||
return True
|
return True
|
||||||
|
|
||||||
|
|
||||||
class RunStoreRunLookup(RunLookup):
|
|
||||||
"""Adapts the framework ``RunStore`` (wide interface) to the narrow
|
|
||||||
``RunLookup`` port -- reuses the existing lookup, no new SQL.
|
|
||||||
|
|
||||||
Used by the service for run-ownership checks before writing feedback.
|
|
||||||
"""
|
|
||||||
|
|
||||||
def __init__(self, run_store) -> None:
|
|
||||||
self._run_store = run_store
|
|
||||||
|
|
||||||
async def thread_of(self, run_id: str) -> str | None:
|
|
||||||
run = await self._run_store.get(run_id)
|
|
||||||
return run.get("thread_id") if run else None
|
|
||||||
47
backend/app/adapters/feedback/run_lookup.py
Normal file
47
backend/app/adapters/feedback/run_lookup.py
Normal file
@ -0,0 +1,47 @@
|
|||||||
|
"""Secondary adapter (anti-corruption layer) -- narrowing RunStore to RunLookup.
|
||||||
|
|
||||||
|
Implements ``RunLookup`` from ``deerflow.domain.feedback.ports``. Unlike its
|
||||||
|
sibling ``feedback_repository.py``, this context does not own the ``runs``
|
||||||
|
table and writes no SQL against it: it asks its one question through the
|
||||||
|
component the run context already provides.
|
||||||
|
|
||||||
|
Being the downstream consumer of an upstream context is the normal shape
|
||||||
|
here, not debt. Until that context publishes a contract of its own, a thin
|
||||||
|
translation layer is what keeps its wide interface -- and its untyped dict
|
||||||
|
rows -- out of the domain.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from typing import TYPE_CHECKING
|
||||||
|
|
||||||
|
from deerflow.domain.feedback.ports import RunLookup
|
||||||
|
|
||||||
|
if TYPE_CHECKING:
|
||||||
|
from deerflow.runtime.runs.store import RunStore
|
||||||
|
|
||||||
|
|
||||||
|
class RunStoreRunLookup(RunLookup):
|
||||||
|
"""Adapts the framework ``RunStore`` (wide interface) to the narrow
|
||||||
|
``RunLookup`` port -- reuses the existing lookup, no new SQL.
|
||||||
|
|
||||||
|
Used by the service to confirm a run belongs to the thread before
|
||||||
|
writing feedback. That is referential integrity, not authorization:
|
||||||
|
the router's ``@require_permission(..., owner_check=True)`` proves the
|
||||||
|
caller owns the thread, and this check proves the run belongs to it --
|
||||||
|
only together do they establish that the caller owns the run. Which is
|
||||||
|
why the port takes no ``user_id``.
|
||||||
|
|
||||||
|
TODO(hexagonal): this depends on ``RunStore``, an infrastructure
|
||||||
|
component, rather than on a contract published by the run context --
|
||||||
|
that context has not been through a hexagonal slice yet. When it
|
||||||
|
publishes one (a DTO, not its aggregate and not its repository),
|
||||||
|
replace the body of this class. The ``RunLookup`` port does not move.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, run_store: RunStore) -> None:
|
||||||
|
self._run_store = run_store
|
||||||
|
|
||||||
|
async def thread_of(self, run_id: str) -> str | None:
|
||||||
|
run = await self._run_store.get(run_id)
|
||||||
|
return run.get("thread_id") if run else None
|
||||||
@ -400,10 +400,8 @@ async def langgraph_runtime(app: FastAPI, startup_config: AppConfig) -> AsyncGen
|
|||||||
# Initialize repositories — one get_session_factory() call for all.
|
# Initialize repositories — one get_session_factory() call for all.
|
||||||
sf = get_session_factory()
|
sf = get_session_factory()
|
||||||
if sf is not None:
|
if sf is not None:
|
||||||
from app.infra.persistence.feedback import (
|
from app.adapters.feedback.feedback_repository import SqlFeedbackRepository
|
||||||
RunStoreRunLookup,
|
from app.adapters.feedback.run_lookup import RunStoreRunLookup
|
||||||
SqlFeedbackRepository,
|
|
||||||
)
|
|
||||||
from deerflow.persistence.run import RunRepository
|
from deerflow.persistence.run import RunRepository
|
||||||
|
|
||||||
app.state.run_store = RunRepository(sf)
|
app.state.run_store = RunRepository(sf)
|
||||||
|
|||||||
@ -105,7 +105,7 @@ class TestSqlFeedbackRepository(FeedbackRepositoryContract):
|
|||||||
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
||||||
|
|
||||||
import deerflow.persistence.feedback.model # noqa: F401 (register FeedbackRow)
|
import deerflow.persistence.feedback.model # noqa: F401 (register FeedbackRow)
|
||||||
from app.infra.persistence.feedback import SqlFeedbackRepository
|
from app.adapters.feedback.feedback_repository import SqlFeedbackRepository
|
||||||
from deerflow.persistence.base import Base
|
from deerflow.persistence.base import Base
|
||||||
|
|
||||||
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'test.db'}")
|
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'test.db'}")
|
||||||
|
|||||||
@ -336,7 +336,7 @@ async def test_run_events_cross_user_delete_denied(tmp_path):
|
|||||||
async def test_feedback_cross_user_isolation(tmp_path):
|
async def test_feedback_cross_user_isolation(tmp_path):
|
||||||
"""Ownership is an explicit user_id filter on the new feedback port —
|
"""Ownership is an explicit user_id filter on the new feedback port —
|
||||||
the AUTO-sentinel context no longer applies (hexagonal slice)."""
|
the AUTO-sentinel context no longer applies (hexagonal slice)."""
|
||||||
from app.infra.persistence.feedback import SqlFeedbackRepository
|
from app.adapters.feedback.feedback_repository import SqlFeedbackRepository
|
||||||
from deerflow.domain.feedback import Feedback
|
from deerflow.domain.feedback import Feedback
|
||||||
from deerflow.persistence.engine import get_session_factory
|
from deerflow.persistence.engine import get_session_factory
|
||||||
|
|
||||||
@ -367,7 +367,7 @@ async def test_feedback_cross_user_isolation(tmp_path):
|
|||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
@pytest.mark.no_auto_user
|
@pytest.mark.no_auto_user
|
||||||
async def test_feedback_cross_user_delete_denied(tmp_path):
|
async def test_feedback_cross_user_delete_denied(tmp_path):
|
||||||
from app.infra.persistence.feedback import SqlFeedbackRepository
|
from app.adapters.feedback.feedback_repository import SqlFeedbackRepository
|
||||||
from deerflow.domain.feedback import Feedback
|
from deerflow.domain.feedback import Feedback
|
||||||
from deerflow.persistence.engine import get_session_factory
|
from deerflow.persistence.engine import get_session_factory
|
||||||
|
|
||||||
|
|||||||
@ -76,7 +76,7 @@ async def test_run_repository_emits_tz_aware_timestamps(tmp_path):
|
|||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_feedback_repository_emits_tz_aware_timestamps(tmp_path):
|
async def test_feedback_repository_emits_tz_aware_timestamps(tmp_path):
|
||||||
from app.infra.persistence.feedback import SqlFeedbackRepository
|
from app.adapters.feedback.feedback_repository import SqlFeedbackRepository
|
||||||
from deerflow.domain.feedback import Feedback
|
from deerflow.domain.feedback import Feedback
|
||||||
|
|
||||||
repo = SqlFeedbackRepository(await _init_sqlite(tmp_path))
|
repo = SqlFeedbackRepository(await _init_sqlite(tmp_path))
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user