refactor(models): remove implicit sessions from workflow_run and credential accessors (#45168)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

api/core/provider_manager.py

@@ -12,6 +12,7 @@ from typing import TYPE_CHECKING, Any, Protocol, Self
 from pydantic import TypeAdapter, ValidationError
 from sqlalchemy import select
 from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session, selectinload
 
 from configs import dify_config
 from core.db.session_factory import session_factory
@@ -112,8 +113,7 @@ class _ProviderModelCacheEntry:
     encrypted_config: str | None
 
     @classmethod
-    def from_record(cls, record: ProviderModel) -> _ProviderModelCacheEntry:
-        credential = record.__dict__.get("credential")
+    def from_record(cls, record: ProviderModel, credential: ProviderModelCredential | None) -> _ProviderModelCacheEntry:
         return cls(
             id=record.id,
             provider_name=record.provider_name,
@@ -440,33 +440,38 @@ def _get_cached_or_load_records[T: _CacheEntry](
     return records
 
 
-def _attach_active_credentials(
-    *,
-    session: Any,
-    records: Sequence[Provider | ProviderModel],
-    credential_model_cls: type[ProviderCredential | ProviderModelCredential],
-) -> None:
-    credential_ids = [record.credential_id for record in records if getattr(record, "credential_id", None)]
-    if not credential_ids:
-        return
+def _load_provider_model_credentials(
+    *, session: Session, provider_models: Sequence[ProviderModel]
+) -> dict[str, ProviderModelCredential]:
+    """Load the credentials referenced by ``provider_models`` through the caller's session.
 
-    credentials = session.scalars(select(credential_model_cls).where(credential_model_cls.id.in_(credential_ids))).all()
-    credential_by_id = {credential.id: credential for credential in credentials}
-    for record in records:
-        if getattr(record, "credential_id", None):
-            record.__dict__["credential"] = credential_by_id.get(record.credential_id)
+    Precondition: ``session`` is open.
+    Postcondition: the result maps each credential ID that still exists to its row.
+    """
+    credential_ids = [
+        provider_model.credential_id for provider_model in provider_models if provider_model.credential_id
+    ]
+    if not credential_ids:
+        return {}
+
+    credentials = session.scalars(
+        select(ProviderModelCredential).where(ProviderModelCredential.id.in_(credential_ids))
+    ).all()
+    return {credential.id: credential for credential in credentials}
 
 
 def _load_provider_model_cache_entries(tenant_id: str) -> list[_ProviderModelCacheEntry]:
     with session_factory.create_session() as session:
         stmt = select(ProviderModel).where(ProviderModel.tenant_id == tenant_id, ProviderModel.is_valid.is_(True))
         provider_models = list(session.scalars(stmt))
-        _attach_active_credentials(
-            session=session,
-            records=provider_models,
-            credential_model_cls=ProviderModelCredential,
-        )
-        return [_ProviderModelCacheEntry.from_record(provider_model) for provider_model in provider_models]
+        credentials = _load_provider_model_credentials(session=session, provider_models=provider_models)
+        return [
+            _ProviderModelCacheEntry.from_record(
+                provider_model,
+                credentials.get(provider_model.credential_id) if provider_model.credential_id else None,
+            )
+            for provider_model in provider_models
+        ]
 
 
 def _load_preferred_model_provider_cache_entries(tenant_id: str) -> list[_TenantPreferredModelProviderCacheEntry]:
@@ -980,13 +985,12 @@ class ProviderManager:
     def _get_all_providers(tenant_id: str) -> dict[str, list[Provider]]:
         provider_name_to_provider_records_dict = defaultdict(list)
         with session_factory.create_session() as session:
-            stmt = select(Provider).where(Provider.tenant_id == tenant_id, Provider.is_valid.is_(True))
-            providers = list(session.scalars(stmt))
-            _attach_active_credentials(
-                session=session,
-                records=providers,
-                credential_model_cls=ProviderCredential,
+            stmt = (
+                select(Provider)
+                .where(Provider.tenant_id == tenant_id, Provider.is_valid.is_(True))
+                .options(selectinload(Provider.credential))
             )
+            providers = list(session.scalars(stmt))
             for provider in providers:
                 # Use provider name with prefix after the data migration
                 provider_name_to_provider_records_dict[str(ModelProviderID(provider.provider_name))].append(provider)
@@ -1235,6 +1239,8 @@ class ProviderManager:
                                 existed_provider_record.is_valid = True
                                 db.session.commit()
 
+                            # Load the credential after the commit above, which expires loaded attributes.
+                            db.session.refresh(existed_provider_record, ["credential"])
                             provider_name_to_provider_records_dict[provider_name].append(existed_provider_record)
 
         return provider_name_to_provider_records_dict

api/models/model.py

@@ -31,7 +31,6 @@ from models.utils.file_input_compat import build_file_from_input_mapping
 
 from .account import Account, Tenant
 from .base import Base, TypeBase, gen_uuidv4_string
-from .engine import db
 from .enums import (
     ApiTokenBindingResourceType,
     ApiTokenType,
@@ -1740,19 +1739,6 @@ class Message(TypeBase):
     def extra_contents(self) -> list[ExtraContentDict]:
         return getattr(self, "_extra_contents", [])
 
-    @property
-    def workflow_run(self):
-        if self.workflow_run_id:
-            from sqlalchemy.orm import sessionmaker
-
-            from repositories.factory import DifyAPIRepositoryFactory
-
-            session_maker = sessionmaker(bind=db.engine, expire_on_commit=False)
-            repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker)
-            return repo.get_workflow_run_by_id_without_tenant(run_id=self.workflow_run_id)
-
-        return None
-
     def to_dict(self, *, session: Session) -> MessageDict:
         return {
             "id": self.id,

api/models/provider.py

@@ -2,15 +2,13 @@ from __future__ import annotations
 
 from datetime import datetime
 from enum import StrEnum, auto
-from functools import cached_property
 from typing import override
 from uuid import uuid4
 
 import sqlalchemy as sa
-from sqlalchemy import DateTime, String, func, select, text
-from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy import DateTime, String, func, text
+from sqlalchemy.orm import Mapped, foreign, mapped_column, relationship
 
-from core.db.session_factory import session_factory
 from graphon.model_runtime.entities.model_entities import ModelType
 from libs.uuid_utils import uuidv7
 
@@ -73,6 +71,16 @@ class Provider(TypeBase):
     updated_at: Mapped[datetime] = mapped_column(
         DateTime, nullable=False, server_default=func.current_timestamp(), onupdate=func.current_timestamp(), init=False
     )
+    # The active credential row, or None when credential_id is unset or the row was deleted.
+    # lazy="raise" keeps reads from opening a session: callers load it with selectinload in their own session.
+    credential: Mapped[ProviderCredential | None] = relationship(
+        lambda: ProviderCredential,
+        primaryjoin=lambda: foreign(Provider.credential_id) == ProviderCredential.id,
+        lazy="raise",
+        viewonly=True,
+        init=False,
+        compare=False,
+    )
 
     @override
     def __repr__(self):
@@ -81,20 +89,15 @@ class Provider(TypeBase):
             f" provider_type='{self.provider_type}')>"
         )
 
-    @cached_property
-    def credential(self):
-        if self.credential_id:
-            with session_factory.create_session() as session:
-                return session.scalar(select(ProviderCredential).where(ProviderCredential.id == self.credential_id))
-
     @property
     def credential_name(self):
-        credential = self.credential
+        # Records without credential_id, such as newly created system records, never need the relationship loaded.
+        credential = self.credential if self.credential_id else None
         return credential.credential_name if credential else None
 
     @property
     def encrypted_config(self):
-        credential = self.credential
+        credential = self.credential if self.credential_id else None
         return credential.encrypted_config if credential else None
 
     @property
@@ -145,24 +148,6 @@ class ProviderModel(TypeBase):
         DateTime, nullable=False, server_default=func.current_timestamp(), onupdate=func.current_timestamp(), init=False
     )
 
-    @cached_property
-    def credential(self):
-        if self.credential_id:
-            with session_factory.create_session() as session:
-                return session.scalar(
-                    select(ProviderModelCredential).where(ProviderModelCredential.id == self.credential_id)
-                )
-
-    @property
-    def credential_name(self):
-        credential = self.credential
-        return credential.credential_name if credential else None
-
-    @property
-    def encrypted_config(self):
-        credential = self.credential
-        return credential.encrypted_config if credential else None
-
 
 class TenantDefaultModel(TypeBase):
     __tablename__ = "tenant_default_models"

api/models/workflow.py

@@ -1402,19 +1402,6 @@ class WorkflowAppLog(TypeBase):
         DateTime, nullable=False, server_default=func.current_timestamp(), init=False
     )
 
-    @property
-    def workflow_run(self):
-        if self.workflow_run_id:
-            from sqlalchemy.orm import sessionmaker
-
-            from repositories.factory import DifyAPIRepositoryFactory
-
-            session_maker = sessionmaker(bind=db.engine, expire_on_commit=False)
-            repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker)
-            return repo.get_workflow_run_by_id_without_tenant(run_id=self.workflow_run_id)
-
-        return None
-
     def created_by_account(self, session: orm.Session) -> Account | None:
         created_by_role = CreatorUserRole(self.created_by_role)
         return session.get(Account, self.created_by) if created_by_role == CreatorUserRole.ACCOUNT else None

api/tests/unit_tests/controllers/web/test_human_input_form.py

@@ -57,11 +57,10 @@ def database_session(
     sqlite_engine: Engine,
     monkeypatch: pytest.MonkeyPatch,
 ) -> Session:
-    """Bind model/controller database access to the shared SQLite session."""
+    """Bind controller database access to the shared SQLite session."""
 
     database = SimpleNamespace(engine=sqlite_engine, session=sqlite_session)
     monkeypatch.setattr(human_input_module, "db", database)
-    monkeypatch.setattr("models.model.db", database)
     return sqlite_session
 
 

api/tests/unit_tests/core/app/task_pipeline/test_message_cycle_manager.py

@@ -18,7 +18,6 @@ from core.app.task_pipeline import message_cycle_manager as message_cycle_manage
 from core.app.task_pipeline.message_cycle_manager import MessageCycleManager
 from core.rag.entities import RetrievalSourceMetadata
 from graphon.file import FileTransferMethod, FileType
-from models import model as model_module
 from models.account import Account
 from models.base import TypeBase
 from models.enums import ConversationFromSource, CreatorUserRole, MessageFileBelongsTo
@@ -48,7 +47,6 @@ def cycle_db(sqlite_engine: Engine, monkeypatch: pytest.MonkeyPatch) -> Iterator
     with owned_session_factory() as request_session:
         sqlite_db = _SQLiteDb(engine=sqlite_engine, session=request_session)
         monkeypatch.setattr(message_cycle_manager_module, "db", sqlite_db)
-        monkeypatch.setattr(model_module, "db", sqlite_db)
         monkeypatch.setattr(message_cycle_manager_module.session_factory, "create_session", owned_session_factory)
         yield request_session
 

api/tests/unit_tests/core/memory/test_token_buffer_memory.py

@@ -14,7 +14,6 @@ import pytest
 from sqlalchemy import Engine, event
 from sqlalchemy.orm import Session
 
-import models.model as model_module
 from core.app.file_access import FileAccessControllerProtocol
 from core.memory import token_buffer_memory as memory_module
 from core.memory.token_buffer_memory import TokenBufferMemory
@@ -88,7 +87,6 @@ def database(sqlite_engine: Engine, monkeypatch: pytest.MonkeyPatch) -> Iterator
     with Session(sqlite_engine, expire_on_commit=False) as session:
         database = Database(engine=sqlite_engine, session=session, statements=statements)
         monkeypatch.setattr(memory_module, "db", SimpleNamespace(engine=sqlite_engine, session=lambda: session))
-        monkeypatch.setattr(model_module, "db", database)
         yield database
     event.remove(sqlite_engine, "before_cursor_execute", record_statement)
 

api/tests/unit_tests/core/test_provider_manager.py

@@ -1,3 +1,4 @@
+from collections import defaultdict
 from collections.abc import Iterator
 from concurrent.futures import ThreadPoolExecutor
 from types import SimpleNamespace
@@ -1202,6 +1203,28 @@ def test_get_all_providers_attaches_active_credentials(provider_db: Session) ->
     assert str(ModelProviderID("anthropic")) not in result
 
 
+def test_init_trial_provider_records_attaches_credential_of_concurrently_created_record(provider_db: Session) -> None:
+    existing_provider = _build_trial_provider_record()
+    existing_provider.credential_id = "credential-id"
+    credential = ProviderCredential(
+        tenant_id="tenant-id",
+        provider_name="openai",
+        credential_name="primary",
+        encrypted_config='{"api_key": "secret"}',
+    )
+    credential.id = "credential-id"
+    provider_db.add_all([existing_provider, credential])
+    provider_db.commit()
+
+    with patch(
+        "core.provider_manager.ext_hosting_provider.hosting_configuration.provider_map",
+        {"openai": _build_hosting_provider()},
+    ):
+        result = ProviderManager._init_trial_provider_records("tenant-id", defaultdict(list))
+
+    assert result["openai"][0].encrypted_config == '{"api_key": "secret"}'
+
+
 def test_invalidate_configurations_cache_bumps_selected_source_version() -> None:
     fake_redis = _FakeRedis()
 
@@ -1271,6 +1294,50 @@ def test_provider_model_credentials_cache_returns_cache_entries(provider_db: Ses
     assert "anthropic" not in second
 
 
+def test_provider_model_cache_entries_resolve_credentials_through_owned_session(provider_db: Session) -> None:
+    credential = ProviderModelCredential(
+        tenant_id="tenant-id",
+        provider_name="openai",
+        model_name="gpt-4",
+        model_type=ModelType.LLM,
+        credential_name="primary",
+        encrypted_config='{"api_key": "secret"}',
+    )
+    credential.id = "credential-id"
+
+    def build_model(model_name: str, credential_id: str | None) -> ProviderModel:
+        return ProviderModel(
+            tenant_id="tenant-id",
+            provider_name="openai",
+            model_name=model_name,
+            model_type=ModelType.LLM,
+            credential_id=credential_id,
+            is_valid=True,
+        )
+
+    provider_db.add_all(
+        [
+            credential,
+            build_model("active", "credential-id"),
+            build_model("dangling", "deleted-credential-id"),
+            build_model("unset", None),
+        ]
+    )
+    provider_db.commit()
+
+    entries = {
+        entry.model_name: entry for entry in provider_manager_module._load_provider_model_cache_entries("tenant-id")
+    }
+
+    assert (entries["active"].credential_name, entries["active"].encrypted_config) == (
+        "primary",
+        '{"api_key": "secret"}',
+    )
+    assert entries["dangling"].credential_id == "deleted-credential-id"
+    assert (entries["dangling"].credential_name, entries["dangling"].encrypted_config) == (None, None)
+    assert (entries["unset"].credential_name, entries["unset"].encrypted_config) == (None, None)
+
+
 def test_provider_configuration_cache_skips_write_when_version_changes_during_load(
     provider_db: Session, sqlite_engine: Engine
 ) -> None:

api/tests/unit_tests/models/test_provider.py

@@ -18,6 +18,9 @@ from datetime import UTC, datetime
 from uuid import uuid4
 
 import pytest
+from sqlalchemy import select
+from sqlalchemy.exc import InvalidRequestError
+from sqlalchemy.orm import Session, selectinload
 from typing_extensions import TypedDict
 
 from graphon.model_runtime.entities.model_entities import ModelType
@@ -260,6 +263,76 @@ class TestProviderModel:
         # Act & Assert
         assert provider.token_is_set is False
 
+    def test_provider_credential_is_loaded_through_callers_session(self, sqlite_session: Session) -> None:
+        """Test credential accessors read the row loaded by the caller's query after the session closes."""
+        # Arrange
+        credential = ProviderCredential(
+            tenant_id=str(uuid4()),
+            provider_name="openai",
+            credential_name="primary",
+            encrypted_config='{"api_key": "secret"}',
+        )
+        sqlite_session.add(credential)
+        sqlite_session.flush()
+        provider = Provider(tenant_id=credential.tenant_id, provider_name="openai", credential_id=credential.id)
+        sqlite_session.add(provider)
+        sqlite_session.commit()
+        sqlite_session.expunge_all()
+
+        # Act
+        loaded = sqlite_session.scalars(
+            select(Provider).where(Provider.id == provider.id).options(selectinload(Provider.credential))
+        ).one()
+        sqlite_session.close()
+
+        # Assert
+        assert loaded.credential_name == "primary"
+        assert loaded.encrypted_config == '{"api_key": "secret"}'
+        assert loaded.token_is_set is True
+
+    def test_provider_credential_missing_row_reads_as_unset(self, sqlite_session: Session) -> None:
+        """Test a credential row deleted before loading reads as unset."""
+        # Arrange
+        provider = Provider(tenant_id=str(uuid4()), provider_name="openai", credential_id=str(uuid4()))
+        sqlite_session.add(provider)
+        sqlite_session.commit()
+        sqlite_session.expunge_all()
+
+        # Act
+        loaded = sqlite_session.scalars(
+            select(Provider).where(Provider.id == provider.id).options(selectinload(Provider.credential))
+        ).one()
+
+        # Assert
+        assert loaded.credential is None
+        assert loaded.token_is_set is False
+
+    def test_committed_provider_without_credential_reads_as_unset(self, sqlite_session: Session) -> None:
+        """Test a committed record without credential_id needs no credential load."""
+        # Arrange
+        provider = Provider(tenant_id=str(uuid4()), provider_name="openai", provider_type=ProviderType.SYSTEM)
+        sqlite_session.add(provider)
+
+        # Act
+        sqlite_session.commit()
+
+        # Assert
+        assert provider.credential_name is None
+        assert provider.encrypted_config is None
+
+    def test_provider_credential_is_not_loaded_implicitly(self, sqlite_session: Session) -> None:
+        """Test reading a credential the query did not load fails instead of opening a query."""
+        # Arrange
+        provider = Provider(tenant_id=str(uuid4()), provider_name="openai", credential_id=str(uuid4()))
+        sqlite_session.add(provider)
+        sqlite_session.commit()
+        sqlite_session.expunge_all()
+        loaded = sqlite_session.scalars(select(Provider).where(Provider.id == provider.id)).one()
+
+        # Act & Assert
+        with pytest.raises(InvalidRequestError, match="lazy='raise'"):
+            _ = loaded.token_is_set
+
     def test_provider_is_enabled_false_when_not_valid(self) -> None:
         """Test is_enabled returns False when provider is not valid."""
         # Arrange