This commit is contained in:
hjlarry 2025-12-12 13:51:57 +08:00
parent da6fa55eed
commit fefcb1e959
3 changed files with 16 additions and 17 deletions

View File

@ -291,7 +291,7 @@ class DifyAPISQLAlchemyWorkflowNodeExecutionRepository(DifyAPIWorkflowNodeExecut
session.commit()
return result.rowcount
class _RunContext(TypedDict):
class RunContext(TypedDict):
run_id: str
tenant_id: str
app_id: str
@ -299,7 +299,7 @@ class DifyAPISQLAlchemyWorkflowNodeExecutionRepository(DifyAPIWorkflowNodeExecut
triggered_from: str
@staticmethod
def delete_by_runs(session: Session, runs: Sequence[_RunContext]) -> tuple[int, int]:
def delete_by_runs(session: Session, runs: Sequence[RunContext]) -> tuple[int, int]:
"""
Delete node executions (and offloads) for the given workflow runs using indexed columns.

View File

@ -8,6 +8,7 @@ from sqlalchemy.orm import Session, sessionmaker
from configs import dify_config
from enums.cloud_plan import CloudPlan
from extensions.ext_database import db
from models.workflow import WorkflowRun
from repositories.api_workflow_run_repository import APIWorkflowRunRepository
from repositories.sqlalchemy_api_workflow_node_execution_repository import (
DifyAPISQLAlchemyWorkflowNodeExecutionRepository,
@ -25,6 +26,7 @@ class WorkflowRunCleanup:
batch_size: int,
start_after: datetime.datetime | None = None,
end_before: datetime.datetime | None = None,
workflow_run_repo: APIWorkflowRunRepository | None = None,
):
if (start_after is None) ^ (end_before is None):
raise ValueError("start_after and end_before must be both set or both omitted.")
@ -38,12 +40,16 @@ class WorkflowRunCleanup:
self.batch_size = batch_size
self.billing_cache: dict[str, CloudPlan | None] = {}
# Lazy import to avoid circular dependency during module import
from repositories.factory import DifyAPIRepositoryFactory
if workflow_run_repo:
self.workflow_run_repo = workflow_run_repo
else:
# Lazy import to avoid circular dependencies during module import
from repositories.factory import DifyAPIRepositoryFactory
self.workflow_run_repo: APIWorkflowRunRepository = DifyAPIRepositoryFactory.create_api_workflow_run_repository(
sessionmaker(bind=db.engine, expire_on_commit=False)
)
session_maker = sessionmaker(bind=db.engine, expire_on_commit=False)
self.workflow_run_repo: APIWorkflowRunRepository = (
DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker)
)
def run(self) -> None:
click.echo(
@ -152,8 +158,8 @@ class WorkflowRunCleanup:
trigger_repo = SQLAlchemyWorkflowTriggerLogRepository(session)
return trigger_repo.delete_by_run_ids(run_ids)
def _delete_node_executions(self, session: Session, runs: Sequence[object]) -> tuple[int, int]:
run_contexts = [
def _delete_node_executions(self, session: Session, runs: Sequence[WorkflowRun]) -> tuple[int, int]:
run_contexts: list[DifyAPISQLAlchemyWorkflowNodeExecutionRepository.RunContext] = [
{
"run_id": run.id,
"tenant_id": run.tenant_id,

View File

@ -3,7 +3,6 @@ from typing import Any
import pytest
import repositories.factory as repo_factory_module
from services import clear_free_plan_expired_workflow_run_logs as cleanup_module
from services.clear_free_plan_expired_workflow_run_logs import WorkflowRunCleanup
@ -64,13 +63,7 @@ class FakeRepo:
def create_cleanup(monkeypatch: pytest.MonkeyPatch, repo: FakeRepo, **kwargs: Any) -> WorkflowRunCleanup:
monkeypatch.setattr(
repo_factory_module.DifyAPIRepositoryFactory,
"create_api_workflow_run_repository",
classmethod(lambda _cls, session_maker: repo),
)
kwargs.pop("repo", None)
return WorkflowRunCleanup(**kwargs)
return WorkflowRunCleanup(workflow_run_repo=repo, **kwargs)
def test_filter_free_tenants_billing_disabled(monkeypatch: pytest.MonkeyPatch) -> None: