refactor(models): pass session into CustomizedSnippet accessors (#40379)

api/controllers/console/workspace/snippets.py

@@ -5,7 +5,7 @@ from uuid import UUID
 from flask import Response, request
 from flask_restx import Resource
 from sqlalchemy.orm import Session, sessionmaker
-from werkzeug.exceptions import NotFound
+from werkzeug.exceptions import BadRequest, NotFound
 
 from controllers.common.fields import TextFileResponse
 from controllers.common.rbac import RBACCheck, Workspace
@@ -36,7 +36,13 @@ from controllers.console.wraps import (
 from core.plugin.entities.plugin import PluginDependency
 from extensions.ext_database import db
 from fields.base import ResponseModel
-from fields.snippet_fields import SnippetListItemResponse, SnippetPaginationResponse, SnippetResponse
+from fields.snippet_fields import (
+    SnippetListItemResponse,
+    SnippetPaginationResponse,
+    SnippetResponse,
+    snippet_list_item_responses,
+    snippet_response,
+)
 from libs.helper import dump_response
 from libs.login import login_required
 from models import Account
@@ -112,14 +118,15 @@ class CustomizedSnippetsApi(Resource):
     @login_required
     @account_initialization_required
     @with_current_tenant_id
-    def get(self, current_tenant_id: str):
+    @with_session(write=False)
+    def get(self, session: Session, current_tenant_id: str):
         """List customized snippets with pagination and search."""
         query = _snippet_list_query_from_request()
 
         snippet_service = _snippet_service()
         snippets, total, has_more = snippet_service.get_snippets(
             tenant_id=current_tenant_id,
-            session=db.session(),
+            session=session,
             page=query.page,
             limit=query.limit,
             keyword=query.keyword,
@@ -131,7 +138,7 @@ class CustomizedSnippetsApi(Resource):
         return dump_response(
             SnippetPaginationResponse,
             {
-                "data": snippets,
+                "data": snippet_list_item_responses(snippets, session=session),
                 "page": query.page,
                 "limit": query.limit,
                 "total": total,
@@ -150,8 +157,9 @@ class CustomizedSnippetsApi(Resource):
     @rbac_permission_required(RBACCheck(RBACPermission.SNIPPETS_CREATE_AND_MODIFY, Workspace()))
     @with_current_user
     @with_current_tenant_id
+    @with_session
     @model_validate(CreateSnippetPayload)
-    def post(self, req_data: CreateSnippetPayload, current_tenant_id: str, current_user: Account):
+    def post(self, req_data: CreateSnippetPayload, session: Session, current_tenant_id: str, current_user: Account):
         """Create a new customized snippet."""
         try:
             snippet_type = SnippetType(req_data.type)
@@ -175,7 +183,7 @@ class CustomizedSnippetsApi(Resource):
         except ValueError as e:
             return {"message": str(e)}, 400
 
-        return dump_response(SnippetResponse, snippet), 201
+        return dump_response(SnippetResponse, snippet_response(snippet, session=session)), 201
 
 
 @console_ns.route("/workspaces/current/customized-snippets/<uuid:snippet_id>")
@@ -187,7 +195,8 @@ class CustomizedSnippetDetailApi(Resource):
     @login_required
     @account_initialization_required
     @with_current_tenant_id
-    def get(self, current_tenant_id: str, snippet_id: UUID):
+    @with_session(write=False)
+    def get(self, session: Session, current_tenant_id: str, snippet_id: UUID):
         """Get customized snippet details."""
         snippet_service = _snippet_service()
         snippet = snippet_service.get_snippet_by_id(
@@ -198,7 +207,7 @@ class CustomizedSnippetDetailApi(Resource):
         if not snippet:
             raise NotFound("Snippet not found")
 
-        return dump_response(SnippetResponse, snippet), 200
+        return dump_response(SnippetResponse, snippet_response(snippet, session=session)), 200
 
     @console_ns.doc("update_customized_snippet")
     @console_ns.expect(console_ns.models.get(UpdateSnippetPayload.__name__))
@@ -212,8 +221,16 @@ class CustomizedSnippetDetailApi(Resource):
     @rbac_permission_required(RBACCheck(RBACPermission.SNIPPETS_CREATE_AND_MODIFY, Workspace()))
     @with_current_user
     @with_current_tenant_id
+    @with_session
     @model_validate(UpdateSnippetPayload)
-    def patch(self, req_data: UpdateSnippetPayload, current_tenant_id: str, current_user: Account, snippet_id: str):
+    def patch(
+        self,
+        req_data: UpdateSnippetPayload,
+        session: Session,
+        current_tenant_id: str,
+        current_user: Account,
+        snippet_id: str,
+    ):
         """Update customized snippet."""
         snippet_service = _snippet_service()
         snippet = snippet_service.get_snippet_by_id(
@@ -233,19 +250,22 @@ class CustomizedSnippetDetailApi(Resource):
             return {"message": "No valid fields to update"}, 400
 
         try:
-            with Session(db.engine, expire_on_commit=False) as session:
+            snippet = session.merge(snippet)
-                snippet = session.merge(snippet)
+            snippet = SnippetService.update_snippet(
-                snippet = SnippetService.update_snippet(
+                session=session,
-                    session=session,
+                snippet=snippet,
-                    snippet=snippet,
+                account_id=current_user.id,
-                    account_id=current_user.id,
+                data=update_data,
-                    data=update_data,
+            )
-                )
+            session.commit()
-                session.commit()
         except ValueError as e:
-            return {"message": str(e)}, 400
+            # Raise rather than return: `with_session` commits on a normal return, so returning here
+            # would persist whatever the update wrote before it rejected the payload. Raising routes
+            # through the decorator's rollback. Status stays 400 and `message` is unchanged; the body
+            # picks up the standard error envelope, as on every other BadRequest in the console API.
+            raise BadRequest(str(e)) from e
 
-        return dump_response(SnippetResponse, snippet), 200
+        return dump_response(SnippetResponse, snippet_response(snippet, session=session)), 200
 
     @console_ns.doc("delete_customized_snippet")
     @console_ns.response(204, "Snippet deleted successfully")

api/fields/snippet_fields.py

@@ -1,12 +1,14 @@
+from collections.abc import Iterable
 from datetime import datetime
 from typing import Any
 
 from pydantic import Field, field_validator
+from sqlalchemy.orm import Session
 
 from fields.base import ResponseModel
 from fields.member_fields import SimpleAccountResponse
 from libs.helper import to_timestamp
-from models.snippet import SnippetType
+from models.snippet import CustomizedSnippet, SnippetType
 
 
 class SnippetTagResponse(ResponseModel):
@@ -72,3 +74,54 @@ class SnippetPaginationResponse(ResponseModel):
     limit: int
     total: int
     has_more: bool
+
+
+def snippet_response(snippet: CustomizedSnippet, *, session: Session) -> SnippetResponse:
+    """Build the snippet detail response, resolving session-backed lookups at the request boundary."""
+    return SnippetResponse.model_validate(
+        {
+            "id": snippet.id,
+            "name": snippet.name,
+            "description": snippet.description,
+            "type": snippet.type,
+            "version": snippet.version,
+            "use_count": snippet.use_count,
+            "is_published": snippet.is_published,
+            "icon_info": snippet.icon_info,
+            "graph": snippet.get_graph_dict(session=session),
+            "input_fields": snippet.input_fields_list,
+            "tags": snippet.get_tags(session=session),
+            "created_by": snippet.get_created_by_account(session=session),
+            "created_at": snippet.created_at,
+            "updated_by": snippet.get_updated_by_account(session=session),
+            "updated_at": snippet.updated_at,
+        }
+    )
+
+
+def snippet_list_item_response(snippet: CustomizedSnippet, *, session: Session) -> SnippetListItemResponse:
+    """Build one snippet list row, resolving session-backed lookups at the request boundary."""
+    return SnippetListItemResponse.model_validate(
+        {
+            "id": snippet.id,
+            "name": snippet.name,
+            "description": snippet.description,
+            "type": snippet.type,
+            "version": snippet.version,
+            "use_count": snippet.use_count,
+            "is_published": snippet.is_published,
+            "icon_info": snippet.icon_info,
+            "tags": snippet.get_tags(session=session),
+            "created_by": snippet.created_by,
+            "author_name": snippet.get_author_name(session=session),
+            "created_at": snippet.created_at,
+            "updated_by": snippet.updated_by,
+            "updated_at": snippet.updated_at,
+        }
+    )
+
+
+def snippet_list_item_responses(
+    snippets: Iterable[CustomizedSnippet], *, session: Session
+) -> list[SnippetListItemResponse]:
+    return [snippet_list_item_response(snippet, session=session) for snippet in snippets]

api/models/snippet.py

@@ -1,17 +1,17 @@
 import json
+from collections.abc import Sequence
 from datetime import datetime
 from enum import StrEnum
 from typing import Any
 
 import sqlalchemy as sa
 from sqlalchemy import DateTime, String, func
-from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy.orm import Mapped, Session, mapped_column
 
 from libs.uuid_utils import uuidv7
 
 from .account import Account
 from .base import Base
-from .engine import db
 from .model import Tag, TagBinding
 from .types import AdjustedJSON, LongText, StringUUID
 
@@ -65,13 +65,12 @@ class CustomizedSnippet(Base):
         DateTime, nullable=False, server_default=func.current_timestamp(), onupdate=func.current_timestamp()
     )
 
-    @property
+    def get_graph_dict(self, *, session: Session) -> dict[str, Any]:
-    def graph_dict(self) -> dict[str, Any]:
         """Get graph from associated workflow."""
         if self.workflow_id:
             from .workflow import Workflow
 
-            workflow = db.session.get(Workflow, self.workflow_id)
+            workflow = session.get(Workflow, self.workflow_id)
             if workflow:
                 return json.loads(workflow.graph) if workflow.graph else {}
         return {}
@@ -81,10 +80,9 @@ class CustomizedSnippet(Base):
         """Parse input_fields JSON to list."""
         return json.loads(self.input_fields) if self.input_fields else []
 
-    @property
+    def get_tags(self, *, session: Session) -> Sequence[Tag]:
-    def tags(self):
         """Get snippet tags."""
-        tags = db.session.scalars(
+        tags = session.scalars(
             sa.select(Tag)
             .join(TagBinding, Tag.id == TagBinding.tag_id)
             .where(
@@ -97,24 +95,21 @@ class CustomizedSnippet(Base):
 
         return tags or []
 
-    @property
+    def get_created_by_account(self, *, session: Session) -> Account | None:
-    def created_by_account(self) -> Account | None:
         """Get the account that created this snippet."""
         if self.created_by:
-            return db.session.get(Account, self.created_by)
+            return session.get(Account, self.created_by)
         return None
 
-    @property
+    def get_author_name(self, *, session: Session) -> str | None:
-    def author_name(self) -> str | None:
         """Get the creator account name."""
-        account = self.created_by_account
+        account = self.get_created_by_account(session=session)
         return account.name if account else None
 
-    @property
+    def get_updated_by_account(self, *, session: Session) -> Account | None:
-    def updated_by_account(self) -> Account | None:
         """Get the account that last updated this snippet."""
         if self.updated_by:
-            return db.session.get(Account, self.updated_by)
+            return session.get(Account, self.updated_by)
         return None
 
     @property

api/tests/unit_tests/controllers/console/workspace/test_snippets.py

@@ -1,14 +1,17 @@
 from datetime import UTC, datetime
 from inspect import unwrap
 from types import SimpleNamespace
-from unittest.mock import ANY, Mock
+from unittest.mock import Mock
 
 import pytest
 from flask import Flask
-from werkzeug.exceptions import NotFound
+from pydantic import ValidationError
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import BadRequest, NotFound
 
 from controllers.console.workspace import snippets as snippets_module
 from models.account import Account, TenantAccountRole
+from models.snippet import CustomizedSnippet
 from services.snippet_dsl_service import ImportStatus, SnippetImportInfo
 
 
@@ -39,30 +42,31 @@ def _account(account_id: str = "account-1") -> Account:
     return account
 
 
-def _snippet(**overrides) -> SimpleNamespace:
+def _snippet(**overrides) -> CustomizedSnippet:
-    data = {
+    """Build a real ``CustomizedSnippet`` row so session-backed accessors run against the test schema.
-        "id": "snippet-1",
+
-        "tenant_id": "tenant-1",
+    The SQLite fixtures in ``tests/unit_tests/conftest.py`` provide the full schema, so the
-        "name": "Snippet",
+    ``get_*(session=...)`` accessors resolve through real queries instead of hand-written stubs.
-        "description": "Description",
+    """
-        "type": snippets_module.SnippetType.NODE,
+    snippet = CustomizedSnippet(
-        "version": 1,
+        tenant_id="tenant-1",
-        "use_count": 0,
+        name="Snippet",
-        "is_published": False,
+        description="Description",
-        "icon_info": None,
+        type=snippets_module.SnippetType.NODE,
-        "graph_dict": {},
+        version=1,
-        "input_fields_list": [],
+        use_count=0,
-        "tags": [],
+        is_published=False,
-        "created_by": None,
+        icon_info=None,
-        "author_name": None,
+        input_fields=None,
-        "created_by_account": None,
+        created_by=None,
-        "created_at": datetime.fromtimestamp(1_704_067_200, UTC),
+        created_at=datetime.fromtimestamp(1_704_067_200, UTC),
-        "updated_by": None,
+        updated_by=None,
-        "updated_by_account": None,
+        updated_at=datetime.fromtimestamp(1_704_153_600, UTC),
-        "updated_at": datetime.fromtimestamp(1_704_153_600, UTC),
+    )
-    }
+    snippet.id = "snippet-1"
-    data.update(overrides)
+    for name, value in overrides.items():
-    return SimpleNamespace(**data)
+        setattr(snippet, name, value)
+    return snippet
 
 
 def test_snippet_list_query_reads_repeated_values(app: Flask):
@@ -98,7 +102,7 @@ def test_snippet_list_query_ignores_indexed_values(app: Flask):
     assert query.creators is None
 
 
-def test_list_snippets_returns_pagination(app: Flask, monkeypatch: pytest.MonkeyPatch):
+def test_list_snippets_returns_pagination(app: Flask, monkeypatch: pytest.MonkeyPatch, sqlite_session: Session):
     snippets = [_snippet()]
     tag_id = "11111111-1111-1111-1111-111111111111"
     get_snippets = Mock(return_value=(snippets, 1, False))
@@ -110,7 +114,7 @@ def test_list_snippets_returns_pagination(app: Flask, monkeypatch: pytest.Monkey
     with app.test_request_context(
         f"/workspaces/current/customized-snippets?page=2&limit=10&tag_ids={tag_id}&creators=account-2"
     ):
-        response, status_code = handler(api, "tenant-1")
+        response, status_code = handler(api, sqlite_session, "tenant-1")
 
     assert status_code == 200
     assert response == {
@@ -139,7 +143,7 @@ def test_list_snippets_returns_pagination(app: Flask, monkeypatch: pytest.Monkey
     }
     get_snippets.assert_called_once_with(
         tenant_id="tenant-1",
-        session=ANY,
+        session=sqlite_session,
         page=2,
         limit=10,
         keyword=None,
@@ -149,7 +153,9 @@ def test_list_snippets_returns_pagination(app: Flask, monkeypatch: pytest.Monkey
     )
 
 
-def test_create_snippet_defaults_unknown_type_and_returns_created(app: Flask, monkeypatch: pytest.MonkeyPatch):
+def test_create_snippet_defaults_unknown_type_and_returns_created(
+    app: Flask, monkeypatch: pytest.MonkeyPatch, sqlite_session: Session
+):
     user = _account("account-1")
     snippet = _snippet()
     create_snippet = Mock(return_value=snippet)
@@ -172,7 +178,7 @@ def test_create_snippet_defaults_unknown_type_and_returns_created(app: Flask, mo
         method="POST",
         json={"name": "Snippet", "type": "node", "description": "Description"},
     ):
-        response, status_code = handler(api, req_data, "tenant-1", user)
+        response, status_code = handler(api, req_data, sqlite_session, "tenant-1", user)
 
     assert status_code == 201
     assert response["id"] == "snippet-1"
@@ -180,7 +186,7 @@ def test_create_snippet_defaults_unknown_type_and_returns_created(app: Flask, mo
     assert create_snippet.call_args.kwargs["snippet_type"] == snippets_module.SnippetType.NODE
 
 
-def test_create_snippet_rejects_forbidden_nodes(app: Flask, monkeypatch: pytest.MonkeyPatch):
+def test_create_snippet_rejects_forbidden_nodes(app: Flask, monkeypatch: pytest.MonkeyPatch, sqlite_session: Session):
     user = _account("account-1")
     create_snippet = Mock()
     monkeypatch.setattr(snippets_module.SnippetService, "create_snippet", create_snippet)
@@ -213,14 +219,14 @@ def test_create_snippet_rejects_forbidden_nodes(app: Flask, monkeypatch: pytest.
             },
         },
     ):
-        response, status_code = handler(api, req_data, "tenant-1", user)
+        response, status_code = handler(api, req_data, sqlite_session, "tenant-1", user)
 
     assert status_code == 400
     assert "knowledge-retrieval" in response["message"]
     create_snippet.assert_not_called()
 
 
-def test_get_snippet_detail_raises_when_missing(app: Flask, monkeypatch: pytest.MonkeyPatch):
+def test_get_snippet_detail_raises_when_missing(app: Flask, monkeypatch: pytest.MonkeyPatch, sqlite_session: Session):
     monkeypatch.setattr(snippets_module.SnippetService, "get_snippet_by_id", Mock(return_value=None))
 
     api = snippets_module.CustomizedSnippetDetailApi()
@@ -228,10 +234,10 @@ def test_get_snippet_detail_raises_when_missing(app: Flask, monkeypatch: pytest.
 
     with app.test_request_context("/workspaces/current/customized-snippets/snippet-1"):
         with pytest.raises(NotFound, match="Snippet not found"):
-            handler(api, "tenant-1", snippet_id="snippet-1")
+            handler(api, sqlite_session, "tenant-1", snippet_id="snippet-1")
 
 
-def test_get_snippet_detail_returns_snippet(app: Flask, monkeypatch: pytest.MonkeyPatch):
+def test_get_snippet_detail_returns_snippet(app: Flask, monkeypatch: pytest.MonkeyPatch, sqlite_session: Session):
     snippet = _snippet()
     monkeypatch.setattr(snippets_module.SnippetService, "get_snippet_by_id", Mock(return_value=snippet))
 
@@ -239,14 +245,38 @@ def test_get_snippet_detail_returns_snippet(app: Flask, monkeypatch: pytest.Monk
     handler = unwrap(api.get)
 
     with app.test_request_context("/workspaces/current/customized-snippets/snippet-1"):
-        response, status_code = handler(api, "tenant-1", snippet_id="snippet-1")
+        response, status_code = handler(api, sqlite_session, "tenant-1", snippet_id="snippet-1")
 
     assert status_code == 200
     assert response["id"] == "snippet-1"
     assert response["name"] == "Snippet"
 
 
-def test_patch_snippet_returns_400_for_empty_payload(app: Flask, monkeypatch: pytest.MonkeyPatch):
+def test_get_snippet_detail_resolves_creator_through_the_request_session(
+    app: Flask, monkeypatch: pytest.MonkeyPatch, sqlite_session: Session
+):
+    """The injected session is the one the accessors query, so a persisted creator resolves."""
+    author = _account("11111111-1111-1111-1111-111111111111")
+    sqlite_session.add(author)
+    sqlite_session.commit()
+
+    snippet = _snippet(created_by=author.id, updated_by=author.id)
+    monkeypatch.setattr(snippets_module.SnippetService, "get_snippet_by_id", Mock(return_value=snippet))
+
+    api = snippets_module.CustomizedSnippetDetailApi()
+    handler = unwrap(api.get)
+
+    with app.test_request_context("/workspaces/current/customized-snippets/snippet-1"):
+        response, status_code = handler(api, sqlite_session, "tenant-1", snippet_id="snippet-1")
+
+    assert status_code == 200
+    assert response["created_by"] == {"id": author.id, "name": "Test User", "email": author.email}
+    assert response["updated_by"] == {"id": author.id, "name": "Test User", "email": author.email}
+
+
+def test_patch_snippet_returns_400_for_empty_payload(
+    app: Flask, monkeypatch: pytest.MonkeyPatch, sqlite_session: Session
+):
     snippet = _snippet()
     user = _account("user-1")
     monkeypatch.setattr(snippets_module.SnippetService, "get_snippet_by_id", Mock(return_value=snippet))
@@ -261,27 +291,20 @@ def test_patch_snippet_returns_400_for_empty_payload(app: Flask, monkeypatch: py
         method="PATCH",
         json={},
     ):
-        response, status_code = handler(api, req_data, "tenant-1", user, snippet_id="snippet-1")
+        response, status_code = handler(api, req_data, sqlite_session, "tenant-1", user, snippet_id="snippet-1")
 
     assert status_code == 400
     assert response == {"message": "No valid fields to update"}
 
 
-def test_patch_snippet_updates_and_commits(app: Flask, monkeypatch: pytest.MonkeyPatch):
+def test_patch_snippet_updates_and_commits(app: Flask, monkeypatch: pytest.MonkeyPatch, sqlite_session: Session):
+    _persist_snippet(sqlite_session)
     user = _account("account-1")
     snippet = _snippet()
-    updated_snippet = _snippet(name="New")
+    update_snippet = Mock(side_effect=_apply_update)
-    session = SimpleNamespace(merge=Mock(return_value=snippet), commit=Mock())
-    update_snippet = Mock(return_value=updated_snippet)
-
-    class SessionContext(_SessionContext):
-        def __init__(self, engine, *args, **kwargs):
-            super().__init__(engine, *args, session=session, **kwargs)
 
     monkeypatch.setattr(snippets_module.SnippetService, "get_snippet_by_id", Mock(return_value=snippet))
     monkeypatch.setattr(snippets_module.SnippetService, "update_snippet", update_snippet)
-    monkeypatch.setattr(snippets_module, "Session", SessionContext)
-    monkeypatch.setattr(snippets_module, "db", SimpleNamespace(engine=object()))
 
     req_data = snippets_module.UpdateSnippetPayload(name="New", icon_info={"icon": "star"})
 
@@ -293,17 +316,125 @@ def test_patch_snippet_updates_and_commits(app: Flask, monkeypatch: pytest.Monke
         method="PATCH",
         json={"name": "New", "icon_info": {"icon": "star"}},
     ):
-        response, status_code = handler(api, req_data, "tenant-1", user, snippet_id="snippet-1")
+        response, status_code = handler(api, req_data, sqlite_session, "tenant-1", user, snippet_id="snippet-1")
 
     assert status_code == 200
     assert response["id"] == "snippet-1"
     assert response["name"] == "New"
     update_snippet.assert_called_once()
+    assert update_snippet.call_args.kwargs["session"] is sqlite_session
     assert update_snippet.call_args.kwargs["data"] == {
         "name": "New",
         "icon_info": {"icon": "star", "icon_background": None, "icon_type": None, "icon_url": None},
     }
-    session.commit.assert_called_once()
+    assert _persisted_name(sqlite_session) == "New"
+
+
+def test_patch_snippet_does_not_report_a_committed_write_as_a_bad_request(
+    app: Flask, monkeypatch: pytest.MonkeyPatch, sqlite_session: Session
+):
+    """The ``except ValueError`` must scope the update call only, never the serialization.
+
+    ``ValidationError`` subclasses ``ValueError``. Both routes end at a 400 either way — the app
+    registers a ``ValueError`` handler in ``libs/external_api.py`` — so what matters is *where* it
+    is handled: covering the serialization would blame the client's payload for a write that already
+    succeeded, and swallow the failure before it ever reaches the error handlers. The committed row
+    must survive regardless.
+    """
+    _persist_snippet(sqlite_session)
+    user = _account("account-1")
+    snippet = _snippet()
+
+    monkeypatch.setattr(snippets_module.SnippetService, "get_snippet_by_id", Mock(return_value=snippet))
+    monkeypatch.setattr(snippets_module.SnippetService, "update_snippet", Mock(side_effect=_apply_update))
+    monkeypatch.setattr(CustomizedSnippet, "get_graph_dict", _unserializable_graph)
+
+    req_data = snippets_module.UpdateSnippetPayload(name="New")
+
+    api = snippets_module.CustomizedSnippetDetailApi()
+    handler = unwrap(api.patch)
+
+    with app.test_request_context(
+        "/workspaces/current/customized-snippets/snippet-1",
+        method="PATCH",
+        json={"name": "New"},
+    ):
+        with pytest.raises(ValidationError):
+            handler(api, req_data, sqlite_session, "tenant-1", user, snippet_id="snippet-1")
+
+    assert _persisted_name(sqlite_session) == "New"
+
+
+def test_patch_snippet_does_not_persist_a_rejected_update(
+    app: Flask, monkeypatch: pytest.MonkeyPatch, sqlite_session: Session
+):
+    """A rejected update must raise, not return a 400 tuple.
+
+    ``with_session`` commits on any normal return, so returning ``{"message": ...}, 400`` would
+    durably persist whatever ``update_snippet`` wrote before it rejected the payload; raising routes
+    through the decorator's rollback instead. That commit lives *in the decorator*, so this test runs
+    the handler through it rather than through ``unwrap`` — the autouse ``_sqlite_session_factory``
+    fixture already points ``with_session`` at the same database as ``sqlite_session``.
+    """
+    _persist_snippet(sqlite_session)
+    user = _account("account-1")
+    snippet = _snippet()
+
+    monkeypatch.setattr(snippets_module.SnippetService, "get_snippet_by_id", Mock(return_value=snippet))
+    monkeypatch.setattr(snippets_module.SnippetService, "update_snippet", Mock(side_effect=_reject_update))
+
+    req_data = snippets_module.UpdateSnippetPayload(name="New")
+
+    api = snippets_module.CustomizedSnippetDetailApi()
+    view = unwrap(api.patch)
+
+    @snippets_module.with_session
+    def patch_through_decorator(resource, session: Session, snippet_id: str):
+        return view(resource, req_data, session, "tenant-1", user, snippet_id=snippet_id)
+
+    with app.test_request_context(
+        "/workspaces/current/customized-snippets/snippet-1",
+        method="PATCH",
+        json={"name": "New"},
+    ):
+        with pytest.raises(BadRequest, match="name already in use"):
+            patch_through_decorator(api, snippet_id="snippet-1")
+
+    assert _persisted_name(sqlite_session) == "Snippet"
+
+
+def _apply_update(*, session: Session, snippet: CustomizedSnippet, account_id: str, data: dict) -> CustomizedSnippet:
+    """Stand in for ``SnippetService.update_snippet``: write the payload onto the merged row."""
+    del session, account_id
+    for field, value in data.items():
+        setattr(snippet, field, value)
+    return snippet
+
+
+def _reject_update(*, session: Session, snippet: CustomizedSnippet, account_id: str, data: dict) -> CustomizedSnippet:
+    """Stand in for an ``update_snippet`` that writes some fields and then rejects the payload."""
+    _apply_update(session=session, snippet=snippet, account_id=account_id, data=data)
+    raise ValueError("name already in use")
+
+
+def _unserializable_graph(self: CustomizedSnippet, *, session: Session) -> str:
+    """Return a non-dict graph so response validation fails after the write is committed."""
+    del self, session
+    return "not-a-dict"
+
+
+def _persist_snippet(session: Session) -> None:
+    """Persist the baseline row so the handler's ``merge`` takes the UPDATE path, as in production."""
+    with Session(bind=session.get_bind()) as setup_session:
+        setup_session.add(_snippet())
+        setup_session.commit()
+
+
+def _persisted_name(session: Session) -> str | None:
+    """Read the snippet name back through a second session to prove what was committed."""
+    with Session(bind=session.get_bind()) as verification_session:
+        stored = verification_session.get(CustomizedSnippet, "snippet-1")
+        return stored.name if stored else None
 
 
 def test_delete_snippet_deletes_and_commits(app: Flask, monkeypatch: pytest.MonkeyPatch):

api/tests/unit_tests/fields/test_snippet_fields.py

@@ -1,17 +1,26 @@
+import json
 from datetime import UTC, datetime
 
 import pytest
 from sqlalchemy.orm import Session
 
-from fields.snippet_fields import SnippetListItemResponse
+from fields.snippet_fields import snippet_list_item_response, snippet_response
-from libs.helper import dump_response
-from models import snippet as snippet_module
 from models.account import Account
-from models.snippet import CustomizedSnippet
+from models.enums import TagType
+from models.model import Tag, TagBinding
+from models.snippet import CustomizedSnippet, SnippetType
+from models.workflow import Workflow, WorkflowType
+
+TENANT_ID = "11111111-1111-1111-1111-111111111111"
+WORKFLOW_ID = "22222222-2222-2222-2222-222222222222"
+APP_ID = "33333333-3333-3333-3333-333333333333"
+SNIPPET_ID = "44444444-4444-4444-4444-444444444444"
+ACCOUNT_1_ID = "55555555-5555-5555-5555-555555555555"
+ACCOUNT_2_ID = "55555555-5555-5555-5555-555555555556"
 
 
 @pytest.mark.parametrize("sqlite_session", [(CustomizedSnippet, Account)], indirect=True)
-def test_snippet_list_fields_include_author_name(sqlite_session: Session, monkeypatch: pytest.MonkeyPatch) -> None:
+def test_snippet_list_fields_include_author_name(sqlite_session: Session) -> None:
     account = Account(name="Alice", email="alice@example.com")
     account.id = "account-1"
     snippet = CustomizedSnippet(
@@ -31,8 +40,72 @@ def test_snippet_list_fields_include_author_name(sqlite_session: Session, monkey
     )
     sqlite_session.add_all([account, snippet])
     sqlite_session.flush()
-    monkeypatch.setattr(snippet_module.db, "session", sqlite_session)
 
-    result = dump_response(SnippetListItemResponse, snippet)
+    result = snippet_list_item_response(snippet, session=sqlite_session).model_dump(mode="json")
 
     assert result["author_name"] == "Alice"
+
+
+@pytest.fixture
+def populated_snippet(sqlite_session: Session) -> CustomizedSnippet:
+    """Persist a snippet plus the workflow, accounts and tag its response resolves."""
+    workflow = Workflow(
+        tenant_id=TENANT_ID,
+        app_id=APP_ID,
+        type=WorkflowType.WORKFLOW,
+        version="1",
+        graph=json.dumps({"nodes": [{"id": "llm-1"}], "edges": []}),
+        _features="{}",
+        created_by=ACCOUNT_1_ID,
+    )
+    workflow.id = WORKFLOW_ID
+    author = Account(name="Ada", email="ada@example.com")
+    author.id = ACCOUNT_1_ID
+    editor = Account(name="Grace", email="grace@example.com")
+    editor.id = ACCOUNT_2_ID
+    tag = Tag(tenant_id=TENANT_ID, type=TagType.SNIPPET, name="Reusable", created_by=ACCOUNT_1_ID)
+    binding = TagBinding(tenant_id=TENANT_ID, tag_id=tag.id, target_id=SNIPPET_ID, created_by=ACCOUNT_1_ID)
+    sqlite_session.add_all((workflow, author, editor, tag, binding))
+    sqlite_session.commit()
+
+    return CustomizedSnippet(
+        id=SNIPPET_ID,
+        tenant_id=TENANT_ID,
+        name="Snippet",
+        description="Reusable node",
+        type=SnippetType.NODE,
+        workflow_id=WORKFLOW_ID,
+        is_published=True,
+        version=1,
+        use_count=0,
+        icon_info=None,
+        input_fields=json.dumps([{"variable": "query"}]),
+        created_by=ACCOUNT_1_ID,
+        updated_by=ACCOUNT_2_ID,
+        created_at=datetime.fromtimestamp(1704067200, tz=UTC),
+        updated_at=datetime.fromtimestamp(1704067201, tz=UTC),
+    )
+
+
+def test_snippet_response_resolves_fields_from_the_given_session(
+    populated_snippet: CustomizedSnippet, sqlite_session: Session
+) -> None:
+    result = snippet_response(populated_snippet, session=sqlite_session).model_dump(mode="json")
+
+    assert result["graph"] == {"nodes": [{"id": "llm-1"}], "edges": []}
+    assert result["input_fields"] == [{"variable": "query"}]
+    assert result["created_by"]["name"] == "Ada"
+    assert result["updated_by"]["name"] == "Grace"
+    assert [tag["name"] for tag in result["tags"]] == ["Reusable"]
+
+
+def test_snippet_list_item_resolves_author_and_tags_from_the_given_session(
+    populated_snippet: CustomizedSnippet, sqlite_session: Session
+) -> None:
+    result = snippet_list_item_response(populated_snippet, session=sqlite_session).model_dump(mode="json")
+
+    assert result["author_name"] == "Ada"
+    assert [tag["name"] for tag in result["tags"]] == ["Reusable"]
+    # The list row carries the raw audit ids; only the detail response resolves them to accounts.
+    assert result["created_by"] == ACCOUNT_1_ID
+    assert result["updated_by"] == ACCOUNT_2_ID

api/tests/unit_tests/models/test_snippet.py

@@ -1,8 +1,7 @@
-"""Snippet model properties backed by the shared SQLite test session."""
+"""Snippet model accessors backed by the shared SQLite test session."""
 
 import json
 
-import pytest
 from sqlalchemy.orm import Session
 
 from models import snippet as snippet_module
@@ -18,24 +17,15 @@ APP_ID = "33333333-3333-3333-3333-333333333333"
 SNIPPET_ID = "44444444-4444-4444-4444-444444444444"
 ACCOUNT_1_ID = "55555555-5555-5555-5555-555555555555"
 ACCOUNT_2_ID = "55555555-5555-5555-5555-555555555556"
-SQLITE_MODELS = (Workflow, Tag, TagBinding, Account)
 
 
-@pytest.fixture
+def test_get_graph_dict_returns_empty_without_workflow_id(sqlite_session: Session) -> None:
-def snippet_session(sqlite_session: Session, monkeypatch: pytest.MonkeyPatch) -> Session:
-    """Expose the shared SQLite session to model properties that use the global Flask session."""
-    monkeypatch.setattr(snippet_module.db, "session", sqlite_session)
-    return sqlite_session
-
-
-def test_graph_dict_returns_empty_without_workflow_id() -> None:
     snippet = CustomizedSnippet(workflow_id=None)
 
-    assert snippet.graph_dict == {}
+    assert snippet.get_graph_dict(session=sqlite_session) == {}
 
 
-@pytest.mark.parametrize("sqlite_session", [SQLITE_MODELS], indirect=True)
+def test_get_graph_dict_loads_published_workflow_graph(sqlite_session: Session) -> None:
-def test_graph_dict_loads_published_workflow_graph(snippet_session: Session) -> None:
     workflow = Workflow(
         tenant_id=TENANT_ID,
         app_id=APP_ID,
@@ -46,18 +36,17 @@ def test_graph_dict_loads_published_workflow_graph(snippet_session: Session) ->
         created_by=ACCOUNT_1_ID,
     )
     workflow.id = WORKFLOW_ID
-    snippet_session.add(workflow)
+    sqlite_session.add(workflow)
-    snippet_session.commit()
+    sqlite_session.commit()
     snippet = CustomizedSnippet(workflow_id=WORKFLOW_ID)
 
-    assert snippet.graph_dict == {"nodes": [{"id": "llm-1"}], "edges": []}
+    assert snippet.get_graph_dict(session=sqlite_session) == {"nodes": [{"id": "llm-1"}], "edges": []}
 
 
-@pytest.mark.parametrize("sqlite_session", [SQLITE_MODELS], indirect=True)
+def test_get_graph_dict_returns_empty_when_workflow_missing(sqlite_session: Session) -> None:
-def test_graph_dict_returns_empty_when_workflow_missing(snippet_session: Session) -> None:
     snippet = CustomizedSnippet(workflow_id=WORKFLOW_ID)
 
-    assert snippet.graph_dict == {}
+    assert snippet.get_graph_dict(session=sqlite_session) == {}
 
 
 def test_input_fields_list_parses_json_or_returns_empty() -> None:
@@ -67,45 +56,50 @@ def test_input_fields_list_parses_json_or_returns_empty() -> None:
     ]
 
 
-@pytest.mark.parametrize("sqlite_session", [SQLITE_MODELS], indirect=True)
+def test_get_tags_returns_query_results_or_empty(sqlite_session: Session) -> None:
-def test_tags_returns_query_results_or_empty(snippet_session: Session) -> None:
     tag = Tag(tenant_id=TENANT_ID, type=TagType.SNIPPET, name="Reusable", created_by=ACCOUNT_1_ID)
     binding = TagBinding(tenant_id=TENANT_ID, tag_id=tag.id, target_id=SNIPPET_ID, created_by=ACCOUNT_1_ID)
-    snippet_session.add_all((tag, binding))
+    sqlite_session.add_all((tag, binding))
-    snippet_session.commit()
+    sqlite_session.commit()
     snippet = CustomizedSnippet(id=SNIPPET_ID, tenant_id=TENANT_ID)
 
-    assert snippet.tags == [tag]
+    assert snippet.get_tags(session=sqlite_session) == [tag]
 
-    snippet_session.delete(binding)
+    sqlite_session.delete(binding)
-    snippet_session.commit()
+    sqlite_session.commit()
-    assert snippet.tags == []
+    assert snippet.get_tags(session=sqlite_session) == []
 
 
-@pytest.mark.parametrize("sqlite_session", [SQLITE_MODELS], indirect=True)
+def test_get_account_accessors_and_author_name(sqlite_session: Session) -> None:
-def test_account_properties_and_author_name(snippet_session: Session) -> None:
     account = Account(name="Ada", email="ada@example.com")
     account.id = ACCOUNT_1_ID
     updated_account = Account(name="Grace", email="grace@example.com")
     updated_account.id = ACCOUNT_2_ID
-    snippet_session.add_all((account, updated_account))
+    sqlite_session.add_all((account, updated_account))
-    snippet_session.commit()
+    sqlite_session.commit()
     snippet = CustomizedSnippet(created_by=ACCOUNT_1_ID, updated_by=ACCOUNT_2_ID)
 
-    assert snippet.created_by_account is account
+    assert snippet.get_created_by_account(session=sqlite_session) is account
-    assert snippet.author_name == "Ada"
+    assert snippet.get_author_name(session=sqlite_session) == "Ada"
-    assert snippet.updated_by_account is updated_account
+    assert snippet.get_updated_by_account(session=sqlite_session) is updated_account
 
 
-def test_account_properties_return_none_without_account_ids() -> None:
+def test_get_account_accessors_return_none_without_account_ids(sqlite_session: Session) -> None:
     snippet = CustomizedSnippet(created_by=None, updated_by=None)
 
-    assert snippet.created_by_account is None
+    assert snippet.get_created_by_account(session=sqlite_session) is None
-    assert snippet.author_name is None
+    assert snippet.get_author_name(session=sqlite_session) is None
-    assert snippet.updated_by_account is None
+    assert snippet.get_updated_by_account(session=sqlite_session) is None
 
 
 def test_version_str_returns_string_value() -> None:
     snippet = CustomizedSnippet(version=7)
 
     assert snippet.version_str == "7"
+
+
+def test_session_backed_lookups_are_not_exposed_as_properties() -> None:
+    """Callers must pass a session; the model no longer reads the Flask-global ``db.session``."""
+    assert not hasattr(snippet_module, "db")
+    for name in ("graph_dict", "tags", "created_by_account", "author_name", "updated_by_account"):
+        assert not hasattr(CustomizedSnippet, name)