jyapayne/langflow
1
0
Fork 0
Code Issues Pull requests Projects Releases 1 Packages Wiki Activity Actions

ref: Add ruff rules for arguments (ARG) (#4123)

Browse source 
Add ruff rules for arguments (ARG)
This commit is contained in:
Christophe Bornet 2024-10-14 17:37:41 +02:00 • committed by GitHub
commit 9a666c1b50
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
49 changed files with 138 additions and 144 deletions
Download patch file Download diff file Expand all files Collapse all files

16
src/backend/base/langflow/__main__.py
View file

@ -95,12 +95,12 @@ def run(
),
log_level: str | None = typer.Option(None, help="Logging level.", show_default=False),
log_file: Path | None = typer.Option(None, help="Path to the log file.", show_default=False),
cache: str | None = typer.Option(
cache: str | None = typer.Option( # noqa: ARG001
None,
help="Type of cache to use. (InMemoryCache, SQLiteCache)",
show_default=False,
),
dev: bool | None = typer.Option(None, help="Run in development mode (may contain bugs)", show_default=False),
dev: bool | None = typer.Option(None, help="Run in development mode (may contain bugs)", show_default=False), # noqa: ARG001
frontend_path: str | None = typer.Option(
None,
help="Path to the frontend directory containing build files. This is for development purposes only.",
@ -111,7 +111,7 @@ def run(
help="Open the browser after starting the server.",
show_default=False,
),
remove_api_keys: bool | None = typer.Option(
remove_api_keys: bool | None = typer.Option( # noqa: ARG001
None,
help="Remove API keys from the projects saved in the database.",
show_default=False,
@ -121,27 +121,27 @@ def run(
help="Run only the backend server without the frontend.",
show_default=False,
),
store: bool | None = typer.Option(
store: bool | None = typer.Option( # noqa: ARG001
None,
help="Enables the store features.",
show_default=False,
),
auto_saving: bool | None = typer.Option(
auto_saving: bool | None = typer.Option( # noqa: ARG001
None,
help="Defines if the auto save is enabled.",
show_default=False,
),
auto_saving_interval: int | None = typer.Option(
auto_saving_interval: int | None = typer.Option( # noqa: ARG001
None,
help="Defines the debounce time for the auto save.",
show_default=False,
),
health_check_max_retries: bool | None = typer.Option(
health_check_max_retries: bool | None = typer.Option( # noqa: ARG001
None,
help="Defines the number of retries for the health check.",
show_default=False,
),
max_file_size_upload: int | None = typer.Option(
max_file_size_upload: int | None = typer.Option( # noqa: ARG001
None,
help="Defines the maximum file size for the upload in MB.",
show_default=False,

3
src/backend/base/langflow/api/v1/api_key.py
View file

@ -46,10 +46,9 @@ def create_api_key_route(
raise HTTPException(status_code=400, detail=str(e)) from e
@router.delete("/{api_key_id}")
@router.delete("/{api_key_id}", dependencies=[Depends(auth_utils.get_current_active_user)])
def delete_api_key_route(
api_key_id: UUID,
current_user=Depends(auth_utils.get_current_active_user),
db: Session = Depends(get_session),
):
try:

22
src/backend/base/langflow/api/v1/callback.py
View file

@ -6,6 +6,7 @@ from uuid import UUID
from langchain_core.agents import AgentAction, AgentFinish
from langchain_core.callbacks.base import AsyncCallbackHandler
from loguru import logger
from typing_extensions import override
from langflow.api.v1.schemas import ChatResponse, PromptResponse
from langflow.services.deps import get_chat_service, get_socket_service
@ -31,11 +32,13 @@ class AsyncStreamingLLMCallbackHandleSIO(AsyncCallbackHandler):
self.sid = session_id
# self.socketio_service = self.chat_service.active_connections[self.client_id]
async def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
@override
async def on_llm_new_token(self, token: str, **kwargs: Any) -> None: # type: ignore[misc]
resp = ChatResponse(message=token, type="stream", intermediate_steps="")
await self.socketio_service.emit_token(to=self.sid, data=resp.model_dump())
async def on_tool_start(self, serialized: dict[str, Any], input_str: str, **kwargs: Any) -> Any:
@override
async def on_tool_start(self, serialized: dict[str, Any], input_str: str, **kwargs: Any) -> Any: # type: ignore[misc]
"""Run when tool starts running."""
resp = ChatResponse(
message="",
@ -88,7 +91,10 @@ class AsyncStreamingLLMCallbackHandleSIO(AsyncCallbackHandler):
) -> None:
"""Run when tool errors."""
async def on_text(self, text: str, **kwargs: Any) -> Any:
@override
async def on_text( # type: ignore[misc]
self, text: str, **kwargs: Any
) -> Any:
"""Run on arbitrary text."""
# This runs when first sending the prompt
# to the LLM, adding it will send the final prompt
@ -101,7 +107,10 @@ class AsyncStreamingLLMCallbackHandleSIO(AsyncCallbackHandler):
)
await self.socketio_service.emit_message(to=self.sid, data=resp.model_dump())
async def on_agent_action(self, action: AgentAction, **kwargs: Any):
@override
async def on_agent_action( # type: ignore[misc]
self, action: AgentAction, **kwargs: Any
):
log = f"Thought: {action.log}"
# if there are line breaks, split them and send them
# as separate messages
@ -114,7 +123,10 @@ class AsyncStreamingLLMCallbackHandleSIO(AsyncCallbackHandler):
resp = ChatResponse(message="", type="stream", intermediate_steps=log)
await self.socketio_service.emit_token(to=self.sid, data=resp.model_dump())
async def on_agent_finish(self, finish: AgentFinish, **kwargs: Any) -> Any:
@override
async def on_agent_finish( # type: ignore[misc]
self, finish: AgentFinish, **kwargs: Any
) -> Any:
"""Run on agent end."""
resp = ChatResponse(
message="",

5
src/backend/base/langflow/api/v1/chat.py
View file

@ -40,13 +40,12 @@ from langflow.graph.utils import log_vertex_build
from langflow.schema.schema import OutputValue
from langflow.services.auth.utils import get_current_active_user
from langflow.services.chat.service import ChatService
from langflow.services.deps import get_chat_service, get_session, get_session_service, get_telemetry_service
from langflow.services.deps import get_chat_service, get_session, get_telemetry_service
from langflow.services.telemetry.schema import ComponentPayload, PlaygroundPayload
from langflow.services.telemetry.service import TelemetryService
if TYPE_CHECKING:
from langflow.graph.vertex.types import InterfaceVertex
from langflow.services.session.service import SessionService
router = APIRouter(tags=["Chat"])
@ -696,9 +695,7 @@ async def _stream_vertex(flow_id: str, vertex_id: str, chat_service: ChatService
async def build_vertex_stream(
flow_id: uuid.UUID,
vertex_id: str,
session_id: str | None = None,
chat_service: ChatService = Depends(get_chat_service),
session_service: SessionService = Depends(get_session_service),
):
"""Build a vertex instead of the entire graph.

25
src/backend/base/langflow/api/v1/endpoints.py
View file

@ -1,7 +1,6 @@
from __future__ import annotations
import time
from asyncio import Lock
from http import HTTPStatus
from typing import TYPE_CHECKING, Annotated
from uuid import UUID
@ -41,7 +40,6 @@ from langflow.services.database.models.flow.model import FlowRead
from langflow.services.database.models.flow.utils import get_all_webhook_components_in_flow
from langflow.services.database.models.user.model import User, UserRead
from langflow.services.deps import (
get_cache_service,
get_session,
get_session_service,
get_settings_service,
@ -49,13 +47,11 @@ from langflow.services.deps import (
get_telemetry_service,
)
from langflow.services.session.service import SessionService
from langflow.services.task.service import TaskService
from langflow.services.telemetry.schema import RunPayload
from langflow.services.telemetry.service import TelemetryService
from langflow.utils.version import get_version_info
if TYPE_CHECKING:
from langflow.services.cache.base import CacheService
from langflow.services.settings.service import SettingsService
router = APIRouter(tags=["Base"])
@ -64,16 +60,11 @@ router = APIRouter(tags=["Base"])
@router.get("/all", dependencies=[Depends(get_current_active_user)])
async def get_all(
settings_service=Depends(get_settings_service),
cache_service: CacheService = Depends(dependency=get_cache_service),
force_refresh: bool = False,
):
from langflow.interface.types import get_and_cache_all_types_dict
try:
async with Lock() as lock:
return await get_and_cache_all_types_dict(
settings_service=settings_service, cache_service=cache_service, force_refresh=force_refresh, lock=lock
)
return await get_and_cache_all_types_dict(settings_service=settings_service)
except Exception as exc:
raise HTTPException(status_code=500, detail=str(exc)) from exc
@ -496,19 +487,9 @@ async def experimental_run_flow(
@router.post(
"/process/{flow_id}",
response_model=ProcessResponse,
dependencies=[Depends(api_key_security)],
)
async def process(
session: Annotated[Session, Depends(get_session)],
flow_id: str,
inputs: list[dict] | dict | None = None,
tweaks: dict | None = None,
clear_cache: Annotated[bool, Body(embed=True)] = False,
session_id: Annotated[None | str, Body(embed=True)] = None,
task_service: TaskService = Depends(get_task_service),
api_key_user: UserRead = Depends(api_key_security),
sync: Annotated[bool, Body(embed=True)] = True,
session_service: SessionService = Depends(get_session_service),
):
async def process():
"""Endpoint to process an input with a given flow_id."""
# Raise a depreciation warning
logger.warning(

14
src/backend/base/langflow/api/v1/monitor.py
View file

@ -10,7 +10,6 @@ from langflow.services.auth.utils import get_current_active_user
from langflow.services.database.models.message.model import MessageRead, MessageTable, MessageUpdate
from langflow.services.database.models.transactions.crud import get_transactions_by_flow_id
from langflow.services.database.models.transactions.model import TransactionReadResponse
from langflow.services.database.models.user.model import User
from langflow.services.database.models.vertex_builds.crud import (
delete_vertex_builds_by_flow_id,
get_vertex_builds_by_flow_id,
@ -72,11 +71,10 @@ async def get_messages(
raise HTTPException(status_code=500, detail=str(e)) from e
@router.delete("/messages", status_code=204)
@router.delete("/messages", status_code=204, dependencies=[Depends(get_current_active_user)])
async def delete_messages(
message_ids: list[UUID],
session: Annotated[Session, Depends(get_session)],
current_user: Annotated[User, Depends(get_current_active_user)],
):
try:
session.exec(delete(MessageTable).where(MessageTable.id.in_(message_ids))) # type: ignore[attr-defined]
@ -85,12 +83,11 @@ async def delete_messages(
raise HTTPException(status_code=500, detail=str(e)) from e
@router.put("/messages/{message_id}", response_model=MessageRead)
@router.put("/messages/{message_id}", dependencies=[Depends(get_current_active_user)], response_model=MessageRead)
async def update_message(
message_id: UUID,
message: MessageUpdate,
session: Annotated[Session, Depends(get_session)],
user: Annotated[User, Depends(get_current_active_user)],
):
try:
db_message = session.get(MessageTable, message_id)
@ -112,12 +109,15 @@ async def update_message(
return db_message
@router.patch("/messages/session/{old_session_id}", response_model=list[MessageResponse])
@router.patch(
"/messages/session/{old_session_id}",
dependencies=[Depends(get_current_active_user)],
response_model=list[MessageResponse],
)
async def update_session_id(
old_session_id: str,
new_session_id: Annotated[str, Query(..., description="The new session ID to update to")],
session: Annotated[Session, Depends(get_session)],
current_user: Annotated[User, Depends(get_current_active_user)],
):
try:
# Get all messages with the old session ID

8
src/backend/base/langflow/api/v1/starter_projects.py
View file

@ -2,16 +2,12 @@ from fastapi import APIRouter, Depends, HTTPException
from langflow.graph.graph.schema import GraphDump
from langflow.services.auth.utils import get_current_active_user
from langflow.services.database.models.user.model import User
router = APIRouter(prefix="/starter-projects", tags=["Flows"])
@router.get("/", response_model=list[GraphDump], status_code=200)
def get_starter_projects(
*,
current_user: User = Depends(get_current_active_user),
):
@router.get("/", dependencies=[Depends(get_current_active_user)], response_model=list[GraphDump], status_code=200)
def get_starter_projects():
"""Get a list of starter projects."""
from langflow.initial_setup.load import get_starter_projects_dump

3
src/backend/base/langflow/api/v1/variable.py
View file

@ -7,7 +7,7 @@ from sqlmodel import Session
from langflow.services.auth.utils import get_current_active_user
from langflow.services.database.models.user.model import User
from langflow.services.database.models.variable import VariableCreate, VariableRead, VariableUpdate
from langflow.services.deps import get_session, get_settings_service, get_variable_service
from langflow.services.deps import get_session, get_variable_service
from langflow.services.variable.base import VariableService
from langflow.services.variable.constants import GENERIC_TYPE
from langflow.services.variable.service import DatabaseVariableService
@ -21,7 +21,6 @@ def create_variable(
session: Session = Depends(get_session),
variable: VariableCreate,
current_user: User = Depends(get_current_active_user),
settings_service=Depends(get_settings_service),
variable_service: DatabaseVariableService = Depends(get_variable_service),
):
"""Create a new variable."""

8
src/backend/base/langflow/base/agents/utils.py
View file

@ -68,9 +68,9 @@ def validate_and_create_openai_tools_agent(
llm: BaseLanguageModel,
tools: Sequence[BaseTool],
prompt: ChatPromptTemplate,
tools_renderer: Callable[[list[BaseTool]], str] = render_text_description,
_tools_renderer: Callable[[list[BaseTool]], str] = render_text_description,
*,
stop_sequence: bool | list[str] = True,
_stop_sequence: bool | list[str] = True,
):
return create_openai_tools_agent(
llm=llm,
@ -83,9 +83,9 @@ def validate_and_create_tool_calling_agent(
llm: BaseLanguageModel,
tools: Sequence[BaseTool],
prompt: ChatPromptTemplate,
tools_renderer: Callable[[list[BaseTool]], str] = render_text_description,
_tools_renderer: Callable[[list[BaseTool]], str] = render_text_description,
*,
stop_sequence: bool | list[str] = True,
_stop_sequence: bool | list[str] = True,
):
return create_tool_calling_agent(
llm=llm,

5
src/backend/base/langflow/base/flow_processing/utils.py
View file

@ -24,12 +24,11 @@ def build_data_from_run_outputs(run_outputs: RunOutputs) -> list[Data]:
return data
def build_data_from_result_data(result_data: ResultData, get_final_results_only: bool = True) -> list[Data]:
def build_data_from_result_data(result_data: ResultData) -> list[Data]:
"""Build a list of data from the given ResultData.
Args:
result_data (ResultData): The ResultData object containing the result data.
get_final_results_only (bool, optional): Whether to include only final results. Defaults to True.
Returns:
List[Data]: A list of data built from the ResultData.
@ -64,7 +63,7 @@ def build_data_from_result_data(result_data: ResultData, get_final_results_only:
if isinstance(result_data.results, dict):
for name, result in result_data.results.items():
dataobj: Data | Message | None = None
dataobj: Data | Message | None
dataobj = result if isinstance(result, Message) else Data(data=result, text_key=name)
data.append(dataobj)

10
src/backend/base/langflow/base/tools/flow_tool.py
View file

@ -4,6 +4,7 @@ from typing import TYPE_CHECKING, Any
from langchain_core.tools import BaseTool, ToolException
from loguru import logger
from typing_extensions import override
from langflow.base.flow_processing.utils import build_data_from_result_data, format_flow_output_data
from langflow.graph.graph.base import Graph # cannot be a part of TYPE_CHECKING # noqa: TCH001
@ -30,7 +31,10 @@ class FlowTool(BaseTool):
schema = self.get_input_schema()
return schema.schema()["properties"]
def get_input_schema(self, config: RunnableConfig | None = None) -> type[BaseModel]:
@override
def get_input_schema( # type: ignore[misc]
self, config: RunnableConfig | None = None
) -> type[BaseModel]:
"""The tool's input schema."""
if self.args_schema is not None:
return self.args_schema
@ -68,7 +72,7 @@ class FlowTool(BaseTool):
if run_output is not None:
for output in run_output.outputs:
if output:
data.extend(build_data_from_result_data(output, get_final_results_only=self.get_final_results_only))
data.extend(build_data_from_result_data(output))
return format_flow_output_data(data)
def validate_inputs(self, args_names: list[dict[str, str]], args: Any, kwargs: Any):
@ -118,5 +122,5 @@ class FlowTool(BaseTool):
if run_output is not None:
for output in run_output.outputs:
if output:
data.extend(build_data_from_result_data(output, get_final_results_only=self.get_final_results_only))
data.extend(build_data_from_result_data(output))
return format_flow_output_data(data)

4
src/backend/base/langflow/components/deactivated/SubFlow.py
View file

@ -104,7 +104,7 @@ class SubFlowComponent(CustomComponent):
},
}
async def build(self, flow_name: str, get_final_results_only: bool = True, **kwargs) -> list[Data]:
async def build(self, flow_name: str, **kwargs) -> list[Data]:
tweaks = {key: {"input_value": value} for key, value in kwargs.items()}
run_outputs: list[RunOutputs | None] = await self.run_flow(
tweaks=tweaks,
@ -118,7 +118,7 @@ class SubFlowComponent(CustomComponent):
if run_output is not None:
for output in run_output.outputs:
if output:
data.extend(build_data_from_result_data(output, get_final_results_only))
data.extend(build_data_from_result_data(output))
self.status = data
logger.debug(data)

6
src/backend/base/langflow/components/embeddings/GoogleGenerativeAIEmbeddings.py
View file

@ -77,7 +77,7 @@ class GoogleGenerativeAIEmbeddingsComponent(Component):
text=text,
task_type=task_type,
title=title,
output_dimensionality=1536,
output_dimensionality=output_dimensionality,
)
for text, title in zip(batch, titles_batch, strict=True)
]
@ -112,12 +112,12 @@ class GoogleGenerativeAIEmbeddingsComponent(Component):
Returns:
Embedding for the text.
"""
task_type = self.task_type or "RETRIEVAL_QUERY"
task_type = task_type or "RETRIEVAL_QUERY"
return self.embed_documents(
[text],
task_type=task_type,
titles=[title] if title else None,
output_dimensionality=1536,
output_dimensionality=output_dimensionality,
)[0]
return HotaGoogleGenerativeAIEmbeddings(model=self.model_name, google_api_key=self.api_key)

3
src/backend/base/langflow/components/helpers/IDGenerator.py
View file

@ -1,6 +1,8 @@
import uuid
from typing import Any
from typing_extensions import override
from langflow.custom import Component
from langflow.io import MessageTextInput, Output
from langflow.schema import dotdict
@ -26,6 +28,7 @@ class IDGeneratorComponent(Component):
Output(display_name="ID", name="id", method="generate_id"),
]
@override
def update_build_config(self, build_config: dotdict, field_value: Any, field_name: str | None = None):
if field_name == "unique_id":
build_config[field_name]["value"] = str(uuid.uuid4())

2
src/backend/base/langflow/components/models/GroqModel.py
View file

@ -1,6 +1,7 @@
import requests
from langchain_groq import ChatGroq
from pydantic.v1 import SecretStr
from typing_extensions import override
from langflow.base.models.model import LCModelComponent
from langflow.field_typing import LanguageModel
@ -75,6 +76,7 @@ class GroqModel(LCModelComponent):
self.status = f"Error fetching models: {e}"
return []
@override
def update_build_config(self, build_config: dict, field_value: str, field_name: str | None = None):
if field_name in {"groq_api_key", "groq_api_base", "model_name"}:
models = self.get_models()

5
src/backend/base/langflow/components/models/HuggingFaceModel.py
View file

@ -52,7 +52,6 @@ class HuggingFaceEndpointsComponent(LCModelComponent):
def create_huggingface_endpoint(
self,
model_id: str,
task: str | None,
huggingfacehub_api_token: str | None,
model_kwargs: dict[str, Any],
@ -76,7 +75,7 @@ class HuggingFaceEndpointsComponent(LCModelComponent):
max_new_tokens=max_new_tokens,
top_k=top_k,
top_p=top_p,
typical_p=self.typical_p,
typical_p=typical_p,
temperature=temperature,
repetition_penalty=repetition_penalty,
)
@ -84,7 +83,6 @@ class HuggingFaceEndpointsComponent(LCModelComponent):
return _attempt_create()
def build_model(self) -> LanguageModel:
model_id = self.model_id
task = self.task or None
huggingfacehub_api_token = self.huggingfacehub_api_token
model_kwargs = self.model_kwargs or {}
@ -97,7 +95,6 @@ class HuggingFaceEndpointsComponent(LCModelComponent):
try:
llm = self.create_huggingface_endpoint(
model_id=model_id,
task=task,
huggingfacehub_api_token=huggingfacehub_api_token,
model_kwargs=model_kwargs,

2
src/backend/base/langflow/components/prototypes/FlowTool.py
View file

@ -1,6 +1,7 @@
from typing import Any
from loguru import logger
from typing_extensions import override
from langflow.base.langchain_utilities.model import LCToolComponent
from langflow.base.tools.flow_tool import FlowTool
@ -39,6 +40,7 @@ class FlowToolComponent(LCToolComponent):
return flow_data
return None
@override
def update_build_config(self, build_config: dotdict, field_value: Any, field_name: str | None = None):
if field_name == "flow_name":
build_config["flow_name"]["options"] = self.get_flow_names()

3
src/backend/base/langflow/components/prototypes/RunFlow.py
View file

@ -1,5 +1,7 @@
from typing import TYPE_CHECKING, Any
from typing_extensions import override
from langflow.base.flow_processing.utils import build_data_from_run_outputs
from langflow.custom import Component
from langflow.io import DropdownInput, MessageTextInput, NestedDictInput, Output
@ -19,6 +21,7 @@ class RunFlowComponent(Component):
flow_data = self.list_flows()
return [flow_data.data["name"] for flow_data in flow_data]
@override
def update_build_config(self, build_config: dotdict, field_value: Any, field_name: str | None = None):
if field_name == "flow_name":
build_config["flow_name"]["options"] = self.get_flow_names()

2
src/backend/base/langflow/components/toolkits/ComposioAPI.py
View file

@ -4,6 +4,7 @@ from typing import Any
from composio_langchain import Action, App, ComposioToolSet
from langchain_core.tools import Tool
from loguru import logger
from typing_extensions import override
from langflow.base.langchain_utilities.model import LCToolComponent
from langflow.inputs import DropdownInput, MessageTextInput, MultiselectInput, SecretStrInput, StrInput
@ -142,6 +143,7 @@ class ComposioAPIComponent(LCToolComponent):
def _get_normalized_app_name(self) -> str:
return self.app_names.replace("_CONNECTED", "").replace("_connected", "")
@override
def update_build_config(self, build_config: dict, field_value: Any, field_name: str | None = None) -> dict:
if field_name == "api_key":
if hasattr(self, "api_key") and self.api_key != "":

2
src/backend/base/langflow/components/tools/PythonCodeStructuredTool.py
View file

@ -7,6 +7,7 @@ from langchain_core.tools import StructuredTool
from loguru import logger
from pydantic.v1 import Field, create_model
from pydantic.v1.fields import Undefined
from typing_extensions import override
from langflow.base.langchain_utilities.model import LCToolComponent
from langflow.inputs.inputs import BoolInput, DropdownInput, FieldTypes, HandleInput, MessageTextInput, MultilineInput
@ -83,6 +84,7 @@ class PythonCodeStructuredTool(LCToolComponent):
Output(display_name="Tool", name="result_tool", method="build_tool"),
]
@override
def update_build_config(self, build_config: dotdict, field_value: Any, field_name: str | None = None) -> dotdict:
if field_name is None:
return build_config

2
src/backend/base/langflow/custom/attributes.py
View file

@ -4,7 +4,7 @@ import emoji
from loguru import logger
def validate_icon(value: str, *args, **kwargs):
def validate_icon(value: str):
# we are going to use the emoji library to validate the emoji
# emojis can be defined using the :emoji_name: syntax

6
src/backend/base/langflow/custom/custom_component/component.py
View file

@ -29,7 +29,6 @@ from .custom_component import CustomComponent
if TYPE_CHECKING:
from collections.abc import Callable
from uuid import UUID
from langflow.events.event_manager import EventManager
from langflow.graph.edge.schema import EdgeData
@ -775,12 +774,9 @@ class Component(CustomComponent):
return str(self.repr_value)
return self.repr_value
def build_inputs(self, user_id: str | UUID | None = None):
def build_inputs(self):
"""Builds the inputs for the custom component.
Args:
user_id (Optional[Union[str, UUID]], optional): The user ID. Defaults to None.
Returns:
List[Input]: The list of inputs.
"""

3
src/backend/base/langflow/custom/utils.py
View file

@ -247,11 +247,10 @@ def get_field_dict(field: Input | dict):
def run_build_inputs(
custom_component: Component,
user_id: str | UUID | None = None,
):
"""Run the build inputs of a custom component."""
try:
return custom_component.build_inputs(user_id=user_id)
return custom_component.build_inputs()
# add_extra_fields(frontend_node, field_config, field_config.values())
except Exception as exc:
logger.exception("Error running build inputs")

4
src/backend/base/langflow/field_typing/range_spec.py
View file

@ -11,7 +11,7 @@ class RangeSpec(BaseModel):
@field_validator("max")
@classmethod
def max_must_be_greater_than_min(cls, v, values, **kwargs):
def max_must_be_greater_than_min(cls, v, values):
if "min" in values.data and v <= values.data["min"]:
msg = "Max must be greater than min"
raise ValueError(msg)
@ -19,7 +19,7 @@ class RangeSpec(BaseModel):
@field_validator("step")
@classmethod
def step_must_be_positive(cls, v, values, **kwargs):
def step_must_be_positive(cls, v, values):
if v <= 0:
msg = "Step must be positive"
raise ValueError(msg)

12
src/backend/base/langflow/graph/graph/base.py
View file

@ -1041,7 +1041,7 @@ class Graph:
# both graphs have the same vertices and edges
# but the data of the vertices might be different
def update_edges_from_vertex(self, vertex: Vertex, other_vertex: Vertex) -> None:
def update_edges_from_vertex(self, other_vertex: Vertex) -> None:
"""Updates the edges of a vertex in the Graph."""
new_edges = []
for edge in self.edges:
@ -1121,7 +1121,7 @@ class Graph:
vertex._data = other_vertex._data
vertex._parse_data()
# Now we update the edges of the vertex
self.update_edges_from_vertex(vertex, other_vertex)
self.update_edges_from_vertex(other_vertex)
vertex.params = {}
vertex._build_params()
vertex.graph = self
@ -1208,7 +1208,7 @@ class Graph:
# All vertices that do not have edges are invalid
return len(self.get_vertex_edges(vertex.id)) > 0
def get_vertex(self, vertex_id: str, silent: bool = False) -> Vertex:
def get_vertex(self, vertex_id: str) -> Vertex:
"""Returns a vertex by id."""
try:
return self.vertex_map[vertex_id]
@ -1289,7 +1289,7 @@ class Graph:
}
)
def _record_snapshot(self, vertex_id: str | None = None, start: bool = False):
def _record_snapshot(self, vertex_id: str | None = None):
self._snapshots.append(self.get_snapshot())
if vertex_id:
self._call_order.append(vertex_id)
@ -1480,7 +1480,7 @@ class Graph:
logger.debug("Graph processing complete")
return self
def find_next_runnable_vertices(self, vertex_id: str, vertex_successors_ids: list[str]) -> list[str]:
def find_next_runnable_vertices(self, vertex_successors_ids: list[str]) -> list[str]:
next_runnable_vertices = set()
for v_id in sorted(vertex_successors_ids):
if not self.is_vertex_runnable(v_id):
@ -1495,7 +1495,7 @@ class Graph:
v_successors_ids = vertex.successors_ids
async with lock:
self.run_manager.remove_vertex_from_runnables(v_id)
next_runnable_vertices = self.find_next_runnable_vertices(v_id, v_successors_ids)
next_runnable_vertices = self.find_next_runnable_vertices(v_successors_ids)
for next_v_id in set(next_runnable_vertices): # Use set to avoid duplicates
if next_v_id == v_id:

5
src/backend/base/langflow/graph/graph/utils.py
View file

@ -131,13 +131,12 @@ def update_template(template, g_nodes):
g_nodes[node_index]["data"]["node"]["template"][field]["display_name"] = display_name
def update_target_handle(new_edge, g_nodes, group_node_id):
def update_target_handle(new_edge, g_nodes):
"""Updates the target handle of a given edge if it is a proxy node.
Args:
new_edge (dict): The edge to update.
g_nodes (list): The list of nodes in the graph.
group_node_id (str): The ID of the group node.
Returns:
dict: The updated edge.
@ -222,7 +221,7 @@ def get_updated_edges(base_flow, g_nodes, g_edges, group_node_id):
for edge in base_flow["edges"]:
new_edge = copy.deepcopy(edge)
if new_edge["target"] == group_node_id:
new_edge = update_target_handle(new_edge, g_nodes, group_node_id)
new_edge = update_target_handle(new_edge, g_nodes)
if new_edge["source"] == group_node_id:
new_edge = update_source_handle(new_edge, g_nodes, g_edges)

2
src/backend/base/langflow/graph/state/model.py
View file

@ -126,7 +126,7 @@ def build_output_setter(method: Callable, validate: bool = True) -> Callable:
>>> print(component.get_output_by_method(component.set_message).value) # Prints "New message"
"""
def output_setter(self, value):
def output_setter(self, value): # noqa: ARG001
if validate:
__validate_method(method)
methods_class = method.__self__

12
src/backend/base/langflow/graph/vertex/base.py
View file

@ -480,7 +480,7 @@ class Vertex:
):
"""Initiate the build process."""
logger.debug(f"Building {self.display_name}")
await self._build_each_vertex_in_params_dict(user_id)
await self._build_each_vertex_in_params_dict()
if self.base_type is None:
msg = f"Base type for vertex {self.display_name} not found"
@ -562,7 +562,7 @@ class Vertex:
)
self.set_result(result_dict)
async def _build_each_vertex_in_params_dict(self, user_id=None):
async def _build_each_vertex_in_params_dict(self):
"""Iterates over each vertex in the params dictionary and builds it."""
for key, value in self._raw_params.items():
if self._is_vertex(value):
@ -622,7 +622,11 @@ class Vertex:
self.log_transaction_tasks.add(task)
task.add_done_callback(self.log_transaction_tasks.discard)
async def _get_result(self, requester: Vertex, target_handle_name: str | None = None) -> Any:
async def _get_result(
self,
requester: Vertex,
target_handle_name: str | None = None, # noqa: ARG002
) -> Any:
"""Retrieves the result of the built component.
If the component has not been built yet, a ValueError is raised.
@ -749,7 +753,7 @@ class Vertex:
msg = f"You are trying to stream to a {self.display_name}. Try using a Chat Output instead."
raise ValueError(msg)
def _reset(self, params_update: dict[str, Any] | None = None):
def _reset(self):
self._built = False
self._built_object = UnbuiltObject()
self._built_result = UnbuiltResult()

2
src/backend/base/langflow/helpers/flow.py
View file

@ -186,7 +186,7 @@ async def flow_function({func_args}):
if run_output is not None:
for output in run_output.outputs:
if output:
data.extend(build_data_from_result_data(output, get_final_results_only=True))
data.extend(build_data_from_result_data(output))
return format_flow_output_data(data)
except Exception as e:
raise ToolException(f'Error running flow: ' + e)

6
src/backend/base/langflow/interface/types.py
View file

@ -8,9 +8,6 @@ from loguru import logger
from langflow.custom.utils import abuild_custom_components, build_custom_components
if TYPE_CHECKING:
import asyncio
from langflow.services.cache.base import CacheService
from langflow.services.settings.service import SettingsService
@ -63,9 +60,6 @@ all_types_dict_cache = None
async def get_and_cache_all_types_dict(
settings_service: SettingsService,
cache_service: CacheService,
force_refresh: bool = False,
lock: asyncio.Lock | None = None,
):
global all_types_dict_cache # noqa: PLW0603
if all_types_dict_cache is None:

14
src/backend/base/langflow/main.py
View file

@ -29,7 +29,7 @@ from langflow.initial_setup.setup import (
from langflow.interface.types import get_and_cache_all_types_dict
from langflow.interface.utils import setup_llm_caching
from langflow.logging.logger import configure
from langflow.services.deps import get_cache_service, get_settings_service, get_telemetry_service
from langflow.services.deps import get_settings_service, get_telemetry_service
from langflow.services.utils import initialize_services, teardown_services
# Ignore Pydantic deprecation warnings from Langchain
@ -90,9 +90,9 @@ class JavaScriptMIMETypeMiddleware(BaseHTTPMiddleware):
telemetry_service_tasks = set()
def get_lifespan(fix_migration=False, socketio_server=None, version=None):
def get_lifespan(fix_migration=False, version=None):
@asynccontextmanager
async def lifespan(app: FastAPI):
async def lifespan(_app: FastAPI):
nest_asyncio.apply()
# Startup message
if version:
@ -100,10 +100,10 @@ def get_lifespan(fix_migration=False, socketio_server=None, version=None):
else:
rprint("[bold green]Starting Langflow...[/bold green]")
try:
initialize_services(fix_migration=fix_migration, socketio_server=socketio_server)
initialize_services(fix_migration=fix_migration)
setup_llm_caching()
initialize_super_user_if_needed()
task = asyncio.create_task(get_and_cache_all_types_dict(get_settings_service(), get_cache_service()))
task = asyncio.create_task(get_and_cache_all_types_dict(get_settings_service()))
await create_or_update_starter_projects(task)
telemetry_service_task = asyncio.create_task(get_telemetry_service().start())
telemetry_service_tasks.add(telemetry_service_task)
@ -206,7 +206,7 @@ def create_app():
app.include_router(log_router)
@app.exception_handler(Exception)
async def exception_handler(request: Request, exc: Exception):
async def exception_handler(_request: Request, exc: Exception):
if isinstance(exc, HTTPException):
logger.error(f"HTTPException: {exc}", exc_info=exc)
return JSONResponse(
@ -252,7 +252,7 @@ def setup_static_files(app: FastAPI, static_files_dir: Path):
)
@app.exception_handler(404)
async def custom_404_handler(request, __):
async def custom_404_handler(_request, _exc):
path = static_files_dir / "index.html"
if not path.exists():

2
src/backend/base/langflow/processing/base.py
View file

@ -11,7 +11,7 @@ if TYPE_CHECKING:
from langfuse.callback import CallbackHandler
def setup_callbacks(sync, trace_id, **kwargs):
def setup_callbacks(trace_id):
"""Setup callbacks for langchain object."""
callbacks = []
plugin_service = get_plugins_service()

6
src/backend/base/langflow/services/cache/service.py vendored
View file

@ -6,6 +6,7 @@ from collections import OrderedDict
from typing import Generic, Union
from loguru import logger
from typing_extensions import override
from langflow.services.cache.base import AsyncBaseCacheService, AsyncLockType, CacheService, LockType
from langflow.services.cache.utils import CACHE_MISS
@ -225,12 +226,14 @@ class RedisCache(AsyncBaseCacheService, Generic[LockType]):
return False
return True
@override
async def get(self, key, lock=None):
if key is None:
return None
value = self._client.get(str(key))
return pickle.loads(value) if value else None
@override
async def set(self, key, value, lock=None):
try:
if pickled := pickle.dumps(value):
@ -242,6 +245,7 @@ class RedisCache(AsyncBaseCacheService, Generic[LockType]):
msg = "RedisCache only accepts values that can be pickled. "
raise TypeError(msg) from exc
@override
async def upsert(self, key, value, lock=None):
"""Inserts or updates a value in the cache.
@ -261,9 +265,11 @@ class RedisCache(AsyncBaseCacheService, Generic[LockType]):
await self.set(key, value)
@override
async def delete(self, key, lock=None):
self._client.delete(key)
@override
async def clear(self, lock=None):
"""Clear all items from the cache."""
self._client.flushdb()

2
src/backend/base/langflow/services/database/service.py
View file

@ -82,7 +82,7 @@ class DatabaseService(Service):
msg = "Error creating database engine"
raise RuntimeError(msg) from exc
def on_connection(self, dbapi_connection, connection_record):
def on_connection(self, dbapi_connection, _connection_record):
from sqlite3 import Connection as sqliteConnection
if isinstance(dbapi_connection, sqliteConnection):

9
src/backend/base/langflow/services/plugins/factory.py
View file

@ -1,17 +1,12 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from langflow.services.factory import ServiceFactory
from langflow.services.plugins.service import PluginService
if TYPE_CHECKING:
from langflow.services.settings.service import SettingsService
class PluginServiceFactory(ServiceFactory):
def __init__(self):
super().__init__(PluginService)
def create(self, settings_service: SettingsService):
return PluginService(settings_service)
def create(self):
return PluginService()

7
src/backend/base/langflow/services/plugins/service.py
View file

@ -3,23 +3,18 @@ from __future__ import annotations
import importlib
import inspect
from pathlib import Path
from typing import TYPE_CHECKING
from loguru import logger
from langflow.services.base import Service
from langflow.services.plugins.base import BasePlugin, CallbackPlugin
if TYPE_CHECKING:
from langflow.services.settings.service import SettingsService
class PluginService(Service):
name = "plugin_service"
def __init__(self, settings_service: SettingsService):
def __init__(self):
self.plugins: dict[str, BasePlugin] = {}
# plugin_dir = settings_service.settings.PLUGIN_DIR
self.plugin_dir = Path(__file__).parent
self.plugins_base_module = "langflow.services.plugins"
self.load_plugins()

4
src/backend/base/langflow/services/settings/base.py
View file

@ -11,6 +11,7 @@ from loguru import logger
from pydantic import field_validator
from pydantic.fields import FieldInfo
from pydantic_settings import BaseSettings, EnvSettingsSource, PydanticBaseSettingsSource, SettingsConfigDict
from typing_extensions import override
from langflow.services.settings.constants import VARIABLES_TO_GET_FROM_ENVIRONMENT
@ -352,7 +353,8 @@ class Settings(BaseSettings):
logger.debug(f"{key}: {getattr(self, key)}")
@classmethod
def settings_customise_sources(
@override
def settings_customise_sources( # type: ignore[misc]
cls,
settings_cls: type[BaseSettings],
init_settings: PydanticBaseSettingsSource,

4
src/backend/base/langflow/services/socket/service.py
View file

@ -53,14 +53,12 @@ class SocketIOService(Service):
async def on_get_vertices(self, sid, flow_id):
await get_vertices(self.sio, sid, flow_id, get_chat_service())
async def on_build_vertex(self, sid, flow_id, vertex_id, tweaks, inputs):
async def on_build_vertex(self, sid, flow_id, vertex_id):
await build_vertex(
sio=self.sio,
sid=sid,
flow_id=flow_id,
vertex_id=vertex_id,
tweaks=tweaks,
inputs=inputs,
get_cache=self.get_cache,
set_cache=self.set_cache,
)

2
src/backend/base/langflow/services/socket/utils.py
View file

@ -48,8 +48,6 @@ async def build_vertex(
vertex_id: str,
get_cache: Callable,
set_cache: Callable,
tweaks=None,
inputs=None,
):
try:
cache = get_cache(flow_id)

2
src/backend/base/langflow/services/telemetry/opentelemetry.py
View file

@ -50,7 +50,7 @@ class ObservableGaugeWrapper:
name=name, description=description, unit=unit, callbacks=[self._callback]
)
def _callback(self, options: CallbackOptions):
def _callback(self, _options: CallbackOptions):
return [Observation(value, attributes=dict(labels)) for labels, value in self._values.items()]
# return [Observation(self._value)]

4
src/backend/base/langflow/services/tracing/langfuse.py
View file

@ -5,6 +5,7 @@ from datetime import datetime, timezone
from typing import TYPE_CHECKING, Any
from loguru import logger
from typing_extensions import override
from langflow.services.tracing.base import BaseTracer
@ -63,6 +64,7 @@ class LangFuseTracer(BaseTracer):
return True
@override
def add_trace(
self,
trace_id: str,
@ -93,6 +95,7 @@ class LangFuseTracer(BaseTracer):
self.last_span = span
self.spans[trace_id] = span
@override
def end_trace(
self,
trace_id: str,
@ -114,6 +117,7 @@ class LangFuseTracer(BaseTracer):
content = {"output": _output, "end_time": end_time}
span.update(**content)
@override
def end(
self,
inputs: dict[str, Any],

3
src/backend/base/langflow/services/tracing/langsmith.py
View file

@ -7,6 +7,7 @@ from datetime import datetime, timezone
from typing import TYPE_CHECKING, Any
from loguru import logger
from typing_extensions import override
from langflow.schema.data import Data
from langflow.services.tracing.base import BaseTracer
@ -62,6 +63,7 @@ class LangSmithTracer(BaseTracer):
os.environ["LANGCHAIN_TRACING_V2"] = "true"
return True
@override
def add_trace(
self,
trace_id: str,
@ -115,6 +117,7 @@ class LangSmithTracer(BaseTracer):
value = str(value)
return value
@override
def end_trace(
self,
trace_id: str,

3
src/backend/base/langflow/services/tracing/langwatch.py
View file

@ -4,6 +4,7 @@ from typing import TYPE_CHECKING, Any, cast
import nanoid
from loguru import logger
from typing_extensions import override
from langflow.schema.data import Data
from langflow.services.tracing.base import BaseTracer
@ -64,6 +65,7 @@ class LangWatchTracer(BaseTracer):
return False
return True
@override
def add_trace(
self,
trace_id: str,
@ -99,6 +101,7 @@ class LangWatchTracer(BaseTracer):
self.trace.set_current_span(span)
self.spans[trace_id] = span
@override
def end_trace(
self,
trace_id: str,

2
src/backend/base/langflow/services/utils.py
View file

@ -149,7 +149,7 @@ def initialize_session_service():
)
def initialize_services(fix_migration: bool = False, socketio_server=None):
def initialize_services(fix_migration: bool = False):
"""Initialize all the services needed."""
# Test cache connection
get_service(ServiceType.CACHE_SERVICE, default=CacheServiceFactory())

2
src/backend/base/langflow/services/variable/kubernetes.py
View file

@ -4,6 +4,7 @@ import os
from typing import TYPE_CHECKING
from loguru import logger
from typing_extensions import override
from langflow.services.auth import utils as auth_utils
from langflow.services.base import Service
@ -26,6 +27,7 @@ class KubernetesSecretService(VariableService, Service):
# TODO: settings_service to set kubernetes namespace
self.kubernetes_secrets = KubernetesSecretManager()
@override
def initialize_user_variables(self, user_id: UUID | str, session: Session):
# Check for environment variables that should be stored in the database
should_or_should_not = "Should" if self.settings_service.settings.store_environment_variables else "Should not"

2
src/backend/base/langflow/services/variable/kubernetes_secrets.py
View file

@ -42,7 +42,7 @@ class KubernetesSecretManager:
return self.core_api.create_namespaced_secret(self.namespace, secret)
def upsert_secret(self, secret_name: str, data: dict, secret_type: str = "Opaque"):
def upsert_secret(self, secret_name: str, data: dict):
"""Upsert a secret in the specified namespace.
If the secret doesn't exist, it will be created.

8
src/backend/base/langflow/worker.py
View file

@ -29,10 +29,10 @@ def build_vertex(self, vertex: Vertex) -> Vertex:
@celery_app.task(acks_late=True)
def process_graph_cached_task(
data_graph: dict[str, Any],
inputs: dict | list[dict] | None = None,
clear_cache=False,
session_id=None,
data_graph: dict[str, Any], # noqa: ARG001
inputs: dict | list[dict] | None = None, # noqa: ARG001
clear_cache=False, # noqa: ARG001
session_id=None, # noqa: ARG001
) -> dict[str, Any]:
msg = "This task is not implemented yet"
raise NotImplementedError(msg)

1
src/backend/base/pyproject.toml
View file

@ -57,7 +57,6 @@ ignore = [
# Rules that are TODOs
"ANN",
"ARG",
"D1", # Missing docstrings
"FBT",
"N",

3
src/backend/tests/unit/graph/test_graph.py
View file

@ -224,8 +224,7 @@ def test_update_target_handle_proxy():
}
}
g_nodes = [{"id": "some_id", "data": {"node": {"flow": None}}}]
group_node_id = "group_id"
updated_edge = update_target_handle(new_edge, g_nodes, group_node_id)
updated_edge = update_target_handle(new_edge, g_nodes)
assert updated_edge["data"]["targetHandle"] == new_edge["data"]["targetHandle"]