diff --git a/backend/src/backend/schedule_schemas.py b/backend/src/backend/schedule_schemas.py index feb0fc4..f614db3 100644 --- a/backend/src/backend/schedule_schemas.py +++ b/backend/src/backend/schedule_schemas.py @@ -5,7 +5,7 @@ from typing import Any from pydantic import Field, field_validator, model_validator -from common.db.models.schedules import FailurePolicy, TriggerType +from common.db.models.schedules import FailurePolicy, PythonVersion, TriggerType from common.schemas import StrictModel @@ -114,6 +114,7 @@ class CreateScheduleNodeRequest(StrictModel): position_y: float = Field(default=0, ge=-100_000, le=100_000) arguments_json: dict[str, Any] = Field(default_factory=dict, max_length=100) env_refs_json: dict[str, str] = Field(default_factory=dict, max_length=100) + python_version: PythonVersion = Field(default="3.12") @field_validator("node_key", "node_name") @classmethod @@ -146,6 +147,7 @@ class UpdateScheduleNodeRequest(StrictModel): default=None, max_length=100, ) + python_version: PythonVersion | None = Field(default=None) @field_validator("node_name") @classmethod diff --git a/backend/src/backend/schedules.py b/backend/src/backend/schedules.py index 59f8925..e163232 100644 --- a/backend/src/backend/schedules.py +++ b/backend/src/backend/schedules.py @@ -151,6 +151,7 @@ def node_payload( "node_key": item.node_key, "node_name": item.node_name, "versions_id": item.versions_id, + "python_version": item.python_version, "timeout_seconds": item.timeout_seconds, "retry_count": item.retry_count, "retry_interval_sec": item.retry_interval_sec, @@ -787,6 +788,7 @@ async def create_schedule_node( node_key=payload.node_key, node_name=payload.node_name, versions_id=payload.versions_id, + python_version=payload.python_version, timeout_seconds=payload.timeout_seconds, retry_count=payload.retry_count, retry_interval_sec=payload.retry_interval_sec, @@ -837,6 +839,7 @@ async def update_schedule_node( mutable_fields = { "node_name", "versions_id", + "python_version", "timeout_seconds", "retry_count", "retry_interval_sec", diff --git a/common/src/common/db/models/schedules.py b/common/src/common/db/models/schedules.py index 2254fe0..a34a54b 100644 --- a/common/src/common/db/models/schedules.py +++ b/common/src/common/db/models/schedules.py @@ -11,6 +11,7 @@ from common.db.base import Base TriggerType = Literal["manual", "cron", "api"] FailurePolicy = Literal["stop", "continue"] +PythonVersion = Literal["3.8", "3.10", "3.12"] class Schedules(Base): @@ -140,6 +141,10 @@ class ScheduleNodes(Base): ) node_name: Mapped[str] = mapped_column(String(255), nullable=False) versions_id: Mapped[str] = mapped_column(CHAR(26), nullable=False) + python_version: Mapped[str] = mapped_column( + String(8), nullable=False, server_default=text("'3.12'"), + comment="节点执行 Python 版本（3.8/3.10/3.12）", + ) timeout_seconds: Mapped[int] = mapped_column( INTEGER, nullable=False, server_default=text("600") ) diff --git a/migrations/versions/d4e5f6a7b8c9_squashed_baseline.py b/migrations/versions/d4e5f6a7b8c9_squashed_baseline.py index 73458a4..3d48e54 100644 --- a/migrations/versions/d4e5f6a7b8c9_squashed_baseline.py +++ b/migrations/versions/d4e5f6a7b8c9_squashed_baseline.py @@ -179,6 +179,7 @@ def upgrade() -> None: sa.Column('node_key', sa.String(length=64), nullable=False, comment='画布内稳定标识'), sa.Column('node_name', sa.String(length=255), nullable=False), sa.Column('versions_id', mysql.CHAR(length=26), nullable=False), + sa.Column('python_version', mysql.String(length=8), nullable=False, server_default=sa.text("'3.12'"), comment='节点执行 Python 版本（3.8/3.10/3.12）'), sa.Column('timeout_seconds', mysql.INTEGER(), server_default=sa.text('600'), nullable=False), sa.Column('retry_count', mysql.INTEGER(), server_default=sa.text('0'), nullable=False), sa.Column('retry_interval_sec', mysql.INTEGER(), server_default=sa.text('5'), nullable=False), diff --git a/migrations/versions/g7h8i9j0k1l2_add_schedule_node_python_version.py b/migrations/versions/g7h8i9j0k1l2_add_schedule_node_python_version.py new file mode 100644 index 0000000..50652fd --- /dev/null +++ b/migrations/versions/g7h8i9j0k1l2_add_schedule_node_python_version.py @@ -0,0 +1,34 @@ +"""add schedule_nodes.python_version + +Revision ID: g7h8i9j0k1l2 +Revises: f6a7b8c9d0e1 +Create Date: 2026-08-11 +""" + +from collections.abc import Sequence + +from alembic import op +import sqlalchemy as sa + + +revision: str = "g7h8i9j0k1l2" +down_revision: str | Sequence[str] | None = "f6a7b8c9d0e1" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.add_column( + "schedule_nodes", + sa.Column( + "python_version", + sa.String(length=8), + nullable=False, + server_default=sa.text("'3.12'"), + comment="节点执行 Python 版本（3.8/3.10/3.12）", + ), + ) + + +def downgrade() -> None: + op.drop_column("schedule_nodes", "python_version") diff --git a/schedule/Dockerfile b/schedule/Dockerfile index 20b3379..5da85a2 100644 --- a/schedule/Dockerfile +++ b/schedule/Dockerfile @@ -7,13 +7,28 @@ COPY extras ./extras RUN UV_DEFAULT_INDEX="https://pypi.tuna.tsinghua.edu.cn/simple/" uv sync --frozen --no-dev --no-editable --package schedule RUN uv venv --python 3.12 /opt/venv/python3.12 && \ - UV_DEFAULT_INDEX="https://pypi.tuna.tsinghua.edu.cn/simple/" uv pip install --python /opt/venv/python3.12/bin/python -r extras/requirements-py312 - -RUN uv venv --python 3.8 /opt/venv/python3.8 && \ - UV_DEFAULT_INDEX="https://pypi.tuna.tsinghua.edu.cn/simple/" uv pip install --python /opt/venv/python3.8/bin/python -r extras/requirements-py38 + UV_DEFAULT_INDEX=https://pypi.tuna.tsinghua.edu.cn/simple/ \ + uv pip install --python /opt/venv/python3.12/bin/python -r extras/requirements-py312 && \ + /opt/venv/python3.12/bin/python -m ipykernel install \ + --prefix=/usr/local \ + --name=python312 \ + --display-name="Python 3.12" RUN uv venv --python 3.10 /opt/venv/python3.10 && \ - UV_DEFAULT_INDEX="https://pypi.tuna.tsinghua.edu.cn/simple/" uv pip install --python /opt/venv/python3.10/bin/python -r extras/requirements-py310 + UV_DEFAULT_INDEX=https://pypi.tuna.tsinghua.edu.cn/simple/ \ + uv pip install --python /opt/venv/python3.10/bin/python -r extras/requirements-py310 && \ + /opt/venv/python3.10/bin/python -m ipykernel install \ + --prefix=/usr/local \ + --name=python310 \ + --display-name="Python 3.10" + +RUN uv venv --python 3.8 /opt/venv/python3.8 && \ + UV_DEFAULT_INDEX=https://pypi.tuna.tsinghua.edu.cn/simple/ \ + uv pip install --python /opt/venv/python3.8/bin/python -r extras/requirements-py38 && \ + /opt/venv/python3.8/bin/python -m ipykernel install \ + --prefix=/usr/local \ + --name=python38 \ + --display-name="Python 3.8" EXPOSE 8000 CMD ["uv", "run", "--frozen", "--package", "schedule", "gunicorn", "--config", "schedule/gunicorn.conf.py", "schedule.main:app"] diff --git a/schedule/src/schedule/execution.py b/schedule/src/schedule/execution.py index d41a570..088e4e9 100644 --- a/schedule/src/schedule/execution.py +++ b/schedule/src/schedule/execution.py @@ -39,12 +39,14 @@ async def _execute_notebook( artifact_name: str, arguments: list[str], timeout_seconds: int, + python_version: str = "3.12", ) -> ExecutionResult: output = artifact.with_name(f"executed-{artifact_name}") logger.debug( - "notebook exec start: artifact={} timeout={}s args={}", + "notebook exec start: artifact={} timeout={}s version={} args={}", artifact_name, timeout_seconds, + python_version, len(arguments), ) process = await asyncio.create_subprocess_exec( @@ -59,6 +61,8 @@ async def _execute_notebook( str(max(1, timeout_seconds)), "--arguments-json", json.dumps(arguments, ensure_ascii=False), + "--python-version", + python_version, cwd=str(artifact.parent), stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, @@ -76,9 +80,10 @@ async def _execute_notebook( error_code = "NODE_TIMEOUT" error_message = f"notebook exceeded timeout of {timeout_seconds} seconds" logger.warning( - "notebook exec timed out: artifact={} timeout={}s", + "notebook exec timed out: artifact={} timeout={}s version={}", artifact_name, timeout_seconds, + python_version, ) else: exit_code = process.returncode @@ -114,9 +119,10 @@ async def _execute_notebook( error_message=error_message, ) logger.info( - "notebook exec done: artifact={} status={} exit_code={}", + "notebook exec done: artifact={} status={} version={} exit_code={}", artifact_name, result.status, + python_version, result.exit_code, ) return result @@ -127,11 +133,15 @@ async def _execute_python( *, arguments: list[str], timeout_seconds: int, + python_version: str = "3.12", ) -> ExecutionResult: + # NOTE: The worker process runs a single interpreter, so .py scripts + # continue to use sys.executable. Switching interpreters is future work. logger.debug( - "python exec start: artifact={} timeout={}s args={}", + "python exec start: artifact={} timeout={}s version={} args={}", artifact.name, timeout_seconds, + python_version, len(arguments), ) process = await asyncio.create_subprocess_exec( @@ -156,9 +166,10 @@ async def _execute_python( ensure_ascii=False, ).encode("utf-8") logger.warning( - "python exec timed out: artifact={} timeout={}s", + "python exec timed out: artifact={} timeout={}s version={}", artifact.name, timeout_seconds, + python_version, ) return ExecutionResult( status="timed_out", @@ -190,9 +201,10 @@ async def _execute_python( error_message=message, ) logger.info( - "python exec done: artifact={} status={} exit_code={}", + "python exec done: artifact={} status={} version={} exit_code={}", artifact.name, result.status, + python_version, result.exit_code, ) return result @@ -207,6 +219,7 @@ async def execute_artifact( artifact_path: str, arguments: list[str], timeout_seconds: int, + python_version: str = "3.12", ) -> ExecutionResult: logger.debug( "execute_artifact: run={} node={} script_type={}", @@ -230,12 +243,14 @@ async def execute_artifact( artifact_name=artifact_name, arguments=arguments, timeout_seconds=timeout_seconds, + python_version=python_version, ) if script_type == "python": return await _execute_python( artifact, arguments=arguments, timeout_seconds=timeout_seconds, + python_version=python_version, ) logger.error("unsupported script_type: {}", script_type) raise ValueError(f"unsupported script_type: {script_type}") diff --git a/schedule/src/schedule/notebook_runner.py b/schedule/src/schedule/notebook_runner.py index 5f86e01..317d8ff 100644 --- a/schedule/src/schedule/notebook_runner.py +++ b/schedule/src/schedule/notebook_runner.py @@ -1,14 +1,11 @@ -from __future__ import annotations - import argparse import json import sys import traceback from pathlib import Path -from loguru import logger - import nbformat +from loguru import logger from nbclient import NotebookClient @@ -27,8 +24,7 @@ def emit_outputs(notebook: object) -> None: ) elif output_type == "error": print( - f"{output.get('ename', 'Error')}: " - f"{output.get('evalue', '')}", + f"{output.get('ename', 'Error')}: {output.get('evalue', '')}", file=sys.stderr, flush=True, ) @@ -39,6 +35,11 @@ def main() -> None: parser.add_argument("--input", required=True) parser.add_argument("--output", required=True) parser.add_argument("--timeout", required=True, type=int) + parser.add_argument( + "--python-version", + choices=("3.8", "3.10", "3.12"), + default="3.12", + ) parser.add_argument("--arguments-json", default="[]") args = parser.parse_args() @@ -51,9 +52,10 @@ def main() -> None: raise ValueError("arguments-json must contain an array of strings") logger.info( - "notebook runner start: input={} timeout={}s args={}", + "notebook runner start: input={} timeout={}s python={} args={}", source.name, args.timeout, + args.python_version, len(arguments), ) notebook = nbformat.read(source, as_version=4) @@ -68,14 +70,16 @@ def main() -> None: ) exit_code = 0 try: + kernel_name = f"python{args.python_version.replace('.', '')}" client = NotebookClient( notebook, timeout=max(1, args.timeout), - kernel_name="python3", + kernel_name=kernel_name, allow_errors=False, ) logger.debug( - "notebook client created: kernel=python3 timeout={}s", + "notebook client created: kernel={} timeout={}s", + kernel_name, max(1, args.timeout), ) # No explicit cwd — the kernel inherits the parent's cwd, which the diff --git a/schedule/src/schedule/worker.py b/schedule/src/schedule/worker.py index b0cf1bb..e727f8a 100644 --- a/schedule/src/schedule/worker.py +++ b/schedule/src/schedule/worker.py @@ -28,6 +28,7 @@ from common.db import session_scope from common.db.models import ( ConsumerInbox, ScheduleNodeRuns, + ScheduleNodes, ScheduleRuns, Schedules, StorageObjects, @@ -86,6 +87,9 @@ class NodeExecutor: object_key=context["object_key"], content_hash=context["content_hash"], ) + python_version = await self._node_python_version( + payload["node_run_id"] + ) result = await execute_artifact( content, run_id=payload["run_id"], @@ -94,6 +98,7 @@ class NodeExecutor: artifact_path=payload["artifact_path"], arguments=[str(item) for item in payload.get("arguments", [])], timeout_seconds=int(payload["timeout_seconds"]), + python_version=python_version, ) except Exception as exc: trace = traceback.format_exc() @@ -250,6 +255,29 @@ class NodeExecutor: ) return True + async def _node_python_version( + self, + node_run_id: str, + ) -> str: + async with self.session_factory() as session: + row = ( + await session.execute( + select(ScheduleNodes.python_version) + .join( + ScheduleNodeRuns, + ScheduleNodeRuns.node_id == ScheduleNodes.node_id, + ) + .where(ScheduleNodeRuns.node_run_id == node_run_id) + ) + ).one_or_none() + if row is None: + logger.warning( + "node python_version not found, defaulting to 3.12: node_run={}", + node_run_id[-12:], + ) + return "3.12" + return row[0] + async def _execution_context( self, payload: dict[str, Any],