"""`RUN ...` - execute a command and capture the resulting image."""
from __future__ import annotations
import json
import shlex
from dataclasses import dataclass, field
from typing import ClassVar
from contree_client.models import InstanceResult
from contree_sdk.exceptions import DockerBuildError
from contree_sdk.execution import RunRequest
from contree_sdk.session.asyncio import ContreeAsyncSession
from contree_sdk.session.base import exit_code_of, or_none
from contree_sdk.session.sync import ContreeSession
from .context import AsyncBuildContext, BuildContext
from .keyword import DockerKeyword, parse_command_form
DISPLAY_TITLE_MAX_LEN = 200
[docs]
@dataclass(frozen=True, repr=False)
class RunKeyword(DockerKeyword):
NAME: ClassVar[str] = "RUN"
parts: tuple[str, ...] = field(default_factory=tuple)
shell_form: bool = True
def __repr__(self) -> str:
if self.shell_form:
return f"RUN {self.parts[0] if self.parts else ''}"
return f"RUN {json.dumps(list(self.parts))}"
[docs]
@classmethod
def parse(cls, args_text: str) -> RunKeyword:
raw = args_text.strip()
if not raw:
raise ValueError("RUN requires a command")
parts, shell_form = parse_command_form(raw)
return cls(parts=tuple(parts), shell_form=shell_form)
[docs]
def serialize(self) -> str:
if self.shell_form:
return f"RUN {self.parts[0]}"
return f"RUN {json.dumps(list(self.parts))}"
[docs]
def execute(self, ctx: BuildContext) -> None:
# RUN is not one of Docker's ${VAR}-substituted instructions - the remote shell
# (or, for exec-form, whatever the argv itself invokes) expands $VAR at runtime
# using the process environment, so `self.parts` is sent verbatim.
contribution = f"RUN shell={self.shell_form} parts={json.dumps(list(self.parts))}"
chain = ctx.chain(contribution)
branch_name = f"layer:{chain}"
if ctx.try_cache_hit(branch_name) is not None:
ctx.parent_hash = chain
ctx.pending.clear()
return
require_session(ctx.session)
result = ctx.run_operation(self.build_request(ctx), branch=branch_name)
check_success(result, self.parts, self.shell_form)
ctx.parent_hash = chain
ctx.pending.clear()
[docs]
async def execute_async(self, ctx: AsyncBuildContext) -> None:
contribution = f"RUN shell={self.shell_form} parts={json.dumps(list(self.parts))}"
chain = ctx.chain(contribution)
branch_name = f"layer:{chain}"
if await ctx.try_cache_hit(branch_name) is not None:
ctx.parent_hash = chain
ctx.pending.clear()
return
require_session_async(ctx.session)
result = await ctx.run_operation(self.build_request(ctx), branch=branch_name)
check_success(result, self.parts, self.shell_form)
ctx.parent_hash = chain
ctx.pending.clear()
[docs]
def build_request(self, ctx: BuildContext | AsyncBuildContext) -> RunRequest:
"""Prepare the command for the context's operation execution hook.
Returns:
A non-disposable request with the current build state and attachments.
"""
command, args, shell = build_command(self.parts, self.shell_form, ctx.user)
return RunRequest(
command=None if shell else command,
shell=command if shell else None,
args=tuple(args),
env={**ctx.arg_values(), **ctx.env} or None,
cwd=ctx.workdir if ctx.workdir != "/" else None,
files=ctx.pending_files_payload() if ctx.pending else None,
disposable=False,
hostname="linuxkit",
truncate_output_at=65536,
timeout=ctx.timeout if ctx.timeout else None,
)
def check_success(result: InstanceResult, parts: tuple[str, ...], shell_form: bool) -> None:
exit_code = exit_code_of(result)
if exit_code is not None and exit_code != 0:
title = display_title(parts, shell_form)
stdout = or_none(result.stdout)
stderr = or_none(result.stderr)
raise DockerBuildError(
f"RUN {title!r} exited with code {exit_code}\n"
f"stdout: {stdout.as_text() if stdout is not None else ''}\n"
f"stderr: {stderr.as_text() if stderr is not None else ''}"
)
def require_session(session: ContreeSession | None) -> ContreeSession:
if session is None:
raise DockerBuildError("no active FROM: cannot RUN without a base image")
return session
def require_session_async(session: ContreeAsyncSession | None) -> ContreeAsyncSession:
if session is None:
raise DockerBuildError("no active FROM: cannot RUN without a base image")
return session
def build_command(parts: tuple[str, ...], shell_form: bool, user: str) -> tuple[str, list[str], bool]:
# maps parsed RUN parts plus an optional active USER into a (command, args, shell) triple
if shell_form:
expr = parts[0]
if user:
return wrap_with_user(expr, user), [], True
return expr, [], True
command = parts[0]
args = list(parts[1:])
if user:
joined = shlex.join([command, *args])
return wrap_with_user(joined, user), [], True
return command, args, False
def wrap_with_user(expr: str, user: str) -> str:
return f"su -s /bin/sh -c {shlex.quote(expr)} {shlex.quote(user)}"
def display_title(parts: tuple[str, ...], shell_form: bool) -> str:
if shell_form:
return f"RUN {parts[0]}"[:DISPLAY_TITLE_MAX_LEN]
return f"RUN {json.dumps(list(parts))}"[:DISPLAY_TITLE_MAX_LEN]