From d649cccda081abfd3b8edf27db5ca7e09278d198 Mon Sep 17 00:00:00 2001 From: eryue0220 Date: Wed, 3 Jun 2026 13:25:08 +0800 Subject: [PATCH] chore: add missing @override decorato to `api/extensions` (#36941) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: Asuka Minato --- api/extensions/ext_logging.py | 4 ++++ api/extensions/ext_login.py | 3 ++- ..._api_workflow_node_execution_repository.py | 21 +++++++++++-------- .../logstore_api_workflow_run_repository.py | 20 ++++++++++++------ .../logstore_workflow_execution_repository.py | 2 ++ ...tore_workflow_node_execution_repository.py | 5 ++++- .../decorators/handlers/generate_handler.py | 2 ++ .../handlers/workflow_app_runner_handler.py | 2 ++ api/extensions/otel/instrumentation.py | 5 +++-- 9 files changed, 45 insertions(+), 19 deletions(-) diff --git a/api/extensions/ext_logging.py b/api/extensions/ext_logging.py index 978a40c503..5817e6a6ce 100644 --- a/api/extensions/ext_logging.py +++ b/api/extensions/ext_logging.py @@ -4,6 +4,7 @@ import logging import os import sys from logging.handlers import RotatingFileHandler +from typing import override from configs import dify_config from dify_app import DifyApp @@ -92,6 +93,7 @@ def _apply_timezone(handlers: list[logging.Handler]): class _TextFormatter(logging.Formatter): """Text formatter that ensures trace_id and req_id are always present.""" + @override def format(self, record: logging.LogRecord) -> str: if not hasattr(record, "req_id"): record.req_id = "" @@ -116,6 +118,7 @@ def get_request_id() -> str: class RequestIdFilter(logging.Filter): """Deprecated: Use TraceContextFilter from core.logging.filters instead.""" + @override def filter(self, record: logging.LogRecord) -> bool: from core.logging.context import get_request_id as _get_request_id from core.logging.context import get_trace_id as _get_trace_id @@ -128,6 +131,7 @@ class RequestIdFilter(logging.Filter): class RequestIdFormatter(logging.Formatter): """Deprecated: Use _TextFormatter instead.""" + @override def format(self, record: logging.LogRecord) -> str: if not hasattr(record, "req_id"): record.req_id = "" diff --git a/api/extensions/ext_login.py b/api/extensions/ext_login.py index 9f9372888f..ee469cd9a5 100644 --- a/api/extensions/ext_login.py +++ b/api/extensions/ext_login.py @@ -1,5 +1,5 @@ import json -from typing import cast +from typing import cast, override import flask_login from flask import Request, Response, request @@ -28,6 +28,7 @@ class DifyLoginManager(flask_login.LoginManager): Flask-Login's broader callback contract. """ + @override def unauthorized(self) -> Response: """Return the registered unauthorized handler result as a Flask `Response`.""" return cast(Response, super().unauthorized()) diff --git a/api/extensions/logstore/repositories/logstore_api_workflow_node_execution_repository.py b/api/extensions/logstore/repositories/logstore_api_workflow_node_execution_repository.py index 64ff0f0674..7eed1cb266 100644 --- a/api/extensions/logstore/repositories/logstore_api_workflow_node_execution_repository.py +++ b/api/extensions/logstore/repositories/logstore_api_workflow_node_execution_repository.py @@ -9,7 +9,7 @@ import logging import time from collections.abc import Sequence from datetime import datetime -from typing import Any +from typing import Any, override from sqlalchemy.orm import sessionmaker @@ -128,6 +128,7 @@ class LogstoreAPIWorkflowNodeExecutionRepository(DifyAPIWorkflowNodeExecutionRep logger.debug("LogstoreAPIWorkflowNodeExecutionRepository.__init__: initializing") self.logstore_client = AliyunLogStore() + @override def get_node_last_execution( self, tenant_id: str, @@ -160,12 +161,12 @@ class LogstoreAPIWorkflowNodeExecutionRepository(DifyAPIWorkflowNodeExecutionRep # Use PG protocol with SQL query (get latest version of each record) sql_query = f""" SELECT * FROM ( - SELECT *, + SELECT *, ROW_NUMBER() OVER (PARTITION BY id ORDER BY log_version DESC) as rn FROM "{AliyunLogStore.workflow_node_execution_logstore}" - WHERE tenant_id = '{escaped_tenant_id}' - AND app_id = '{escaped_app_id}' - AND workflow_id = '{escaped_workflow_id}' + WHERE tenant_id = '{escaped_tenant_id}' + AND app_id = '{escaped_app_id}' + AND workflow_id = '{escaped_workflow_id}' AND node_id = '{escaped_node_id}' AND __time__ > 0 ) AS subquery WHERE rn = 1 @@ -236,6 +237,7 @@ class LogstoreAPIWorkflowNodeExecutionRepository(DifyAPIWorkflowNodeExecutionRep logger.exception("Failed to get node last execution from LogStore") raise + @override def get_executions_by_workflow_run( self, tenant_id: str, @@ -265,11 +267,11 @@ class LogstoreAPIWorkflowNodeExecutionRepository(DifyAPIWorkflowNodeExecutionRep # Use PG protocol with SQL query (get latest version of each record) sql_query = f""" SELECT * FROM ( - SELECT *, + SELECT *, ROW_NUMBER() OVER (PARTITION BY id ORDER BY log_version DESC) as rn FROM "{AliyunLogStore.workflow_node_execution_logstore}" - WHERE tenant_id = '{escaped_tenant_id}' - AND app_id = '{escaped_app_id}' + WHERE tenant_id = '{escaped_tenant_id}' + AND app_id = '{escaped_app_id}' AND workflow_run_id = '{escaped_workflow_run_id}' AND __time__ > 0 ) AS subquery WHERE rn = 1 @@ -340,6 +342,7 @@ class LogstoreAPIWorkflowNodeExecutionRepository(DifyAPIWorkflowNodeExecutionRep logger.exception("Failed to get executions by workflow run from LogStore") raise + @override def get_execution_by_id( self, execution_id: str, @@ -365,7 +368,7 @@ class LogstoreAPIWorkflowNodeExecutionRepository(DifyAPIWorkflowNodeExecutionRep sql_query = f""" SELECT * FROM ( - SELECT *, + SELECT *, ROW_NUMBER() OVER (PARTITION BY id ORDER BY log_version DESC) as rn FROM "{AliyunLogStore.workflow_node_execution_logstore}" WHERE id = '{escaped_execution_id}' {tenant_filter} AND __time__ > 0 diff --git a/api/extensions/logstore/repositories/logstore_api_workflow_run_repository.py b/api/extensions/logstore/repositories/logstore_api_workflow_run_repository.py index 7f77a0437a..cdc7d129fd 100644 --- a/api/extensions/logstore/repositories/logstore_api_workflow_run_repository.py +++ b/api/extensions/logstore/repositories/logstore_api_workflow_run_repository.py @@ -18,7 +18,7 @@ import os import time from collections.abc import Sequence from datetime import datetime -from typing import Any, cast +from typing import Any, cast, override from sqlalchemy.orm import sessionmaker @@ -162,6 +162,7 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository): # Set to False for new deployments without legacy data in PostgreSQL self._enable_dual_read = os.environ.get("LOGSTORE_DUAL_READ_ENABLED", "true").lower() == "true" + @override def get_paginated_workflow_runs( self, tenant_id: str, @@ -257,6 +258,7 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository): logger.exception("Failed to get paginated workflow runs from LogStore") raise + @override def get_workflow_run_by_id( self, tenant_id: str, @@ -282,12 +284,12 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository): # Use PG protocol with SQL query (get latest version of record) sql_query = f""" SELECT * FROM ( - SELECT *, + SELECT *, ROW_NUMBER() OVER (PARTITION BY id ORDER BY log_version DESC) as rn FROM "{AliyunLogStore.workflow_execution_logstore}" - WHERE id = '{escaped_run_id}' - AND tenant_id = '{escaped_tenant_id}' - AND app_id = '{escaped_app_id}' + WHERE id = '{escaped_run_id}' + AND tenant_id = '{escaped_tenant_id}' + AND app_id = '{escaped_app_id}' AND __time__ > 0 ) AS subquery WHERE rn = 1 LIMIT 100 @@ -364,6 +366,7 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository): ) return session.scalar(stmt) + @override def get_workflow_run_by_id_without_tenant( self, run_id: str, @@ -384,7 +387,7 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository): # Use PG protocol with SQL query (get latest version of record) sql_query = f""" SELECT * FROM ( - SELECT *, + SELECT *, ROW_NUMBER() OVER (PARTITION BY id ORDER BY log_version DESC) as rn FROM "{AliyunLogStore.workflow_execution_logstore}" WHERE id = '{escaped_run_id}' AND __time__ > 0 @@ -447,6 +450,7 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository): stmt = select(WorkflowRun).where(WorkflowRun.id == run_id) return session.scalar(stmt) + @override def get_workflow_runs_count( self, tenant_id: str, @@ -594,6 +598,7 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository): logger.exception("Failed to get workflow runs count") raise + @override def get_daily_runs_statistics( self, tenant_id: str, @@ -652,6 +657,7 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository): logger.exception("Failed to get daily runs statistics") raise + @override def get_daily_terminals_statistics( self, tenant_id: str, @@ -712,6 +718,7 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository): logger.exception("Failed to get daily terminals statistics") raise + @override def get_daily_token_cost_statistics( self, tenant_id: str, @@ -772,6 +779,7 @@ class LogstoreAPIWorkflowRunRepository(APIWorkflowRunRepository): logger.exception("Failed to get daily token cost statistics") raise + @override def get_average_app_interaction_statistics( self, tenant_id: str, diff --git a/api/extensions/logstore/repositories/logstore_workflow_execution_repository.py b/api/extensions/logstore/repositories/logstore_workflow_execution_repository.py index 544109276d..d0653bec3f 100644 --- a/api/extensions/logstore/repositories/logstore_workflow_execution_repository.py +++ b/api/extensions/logstore/repositories/logstore_workflow_execution_repository.py @@ -2,6 +2,7 @@ import json import logging import os import time +from typing import override from sqlalchemy.engine import Engine from sqlalchemy.orm import sessionmaker @@ -152,6 +153,7 @@ class LogstoreWorkflowExecutionRepository(WorkflowExecutionRepository): return logstore_model + @override def save(self, execution: WorkflowExecution) -> None: """ Save or update a WorkflowExecution domain entity to the logstore. diff --git a/api/extensions/logstore/repositories/logstore_workflow_node_execution_repository.py b/api/extensions/logstore/repositories/logstore_workflow_node_execution_repository.py index dc7654a25c..6daac8a0a5 100644 --- a/api/extensions/logstore/repositories/logstore_workflow_node_execution_repository.py +++ b/api/extensions/logstore/repositories/logstore_workflow_node_execution_repository.py @@ -11,7 +11,7 @@ import os import time from collections.abc import Sequence from datetime import datetime -from typing import Any +from typing import Any, override from sqlalchemy.engine import Engine from sqlalchemy.orm import sessionmaker @@ -222,6 +222,7 @@ class LogstoreWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository): return logstore_model + @override def save(self, execution: WorkflowNodeExecution) -> None: """ Save or update a NodeExecution domain entity to LogStore. @@ -271,6 +272,7 @@ class LogstoreWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository): logger.exception("Failed to dual-write node execution to SQL database: id=%s", execution.id) # Don't raise - LogStore write succeeded, SQL is just a backup + @override def save_execution_data(self, execution: WorkflowNodeExecution) -> None: """ Save or update the inputs, process_data, or outputs associated with a specific @@ -305,6 +307,7 @@ class LogstoreWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository): logger.exception("Failed to dual-write node execution data to SQL database: id=%s", execution.id) # Don't raise - LogStore write succeeded, SQL is just a backup + @override def get_by_workflow_execution( self, workflow_execution_id: str, diff --git a/api/extensions/otel/decorators/handlers/generate_handler.py b/api/extensions/otel/decorators/handlers/generate_handler.py index df5142c310..514dad5daf 100644 --- a/api/extensions/otel/decorators/handlers/generate_handler.py +++ b/api/extensions/otel/decorators/handlers/generate_handler.py @@ -1,5 +1,6 @@ import logging from collections.abc import Callable +from typing import override from opentelemetry.trace import SpanKind, Status, StatusCode, Tracer from opentelemetry.util.types import AttributeValue @@ -14,6 +15,7 @@ logger = logging.getLogger(__name__) class AppGenerateHandler(SpanHandler): """Span handler for ``AppGenerateService.generate``.""" + @override def wrapper[**P, R]( self, tracer: Tracer, diff --git a/api/extensions/otel/decorators/handlers/workflow_app_runner_handler.py b/api/extensions/otel/decorators/handlers/workflow_app_runner_handler.py index 6b2112ceb2..77d8e435d0 100644 --- a/api/extensions/otel/decorators/handlers/workflow_app_runner_handler.py +++ b/api/extensions/otel/decorators/handlers/workflow_app_runner_handler.py @@ -1,5 +1,6 @@ import logging from collections.abc import Callable +from typing import override from opentelemetry.trace import SpanKind, Status, StatusCode, Tracer from opentelemetry.util.types import AttributeValue @@ -13,6 +14,7 @@ logger = logging.getLogger(__name__) class WorkflowAppRunnerHandler(SpanHandler): """Span handler for ``WorkflowAppRunner.run``.""" + @override def wrapper[**P, R]( self, tracer: Tracer, diff --git a/api/extensions/otel/instrumentation.py b/api/extensions/otel/instrumentation.py index 0a70f6ebe9..1718f9eb65 100644 --- a/api/extensions/otel/instrumentation.py +++ b/api/extensions/otel/instrumentation.py @@ -1,7 +1,7 @@ import contextlib import logging from collections.abc import Callable -from typing import Protocol, cast +from typing import Protocol, cast, override import flask from opentelemetry.instrumentation.celery import CeleryInstrumentor @@ -63,7 +63,8 @@ class ExceptionLoggingHandler(logging.Handler): to maintain trace context consistency throughout the request lifecycle. """ - def emit(self, record: logging.LogRecord): + @override + def emit(self, record: logging.LogRecord) -> None: with contextlib.suppress(Exception): if not record.exc_info: return