`pygx.coding`¶

Runtime compilation, execution and sandboxing of Python source code.

coding ¶

Runtime compilation, execution and sandboxing of Python source code.

pygx.coding provides a small toolkit for compiling and executing Python source code at runtime under controlled conditions. It is used internally by PyGX to materialize generated programs, but is exposed as a public sub-package because the building blocks are useful on their own:

parse / evaluate / run for compiling and executing Python source.
make_function for synthesizing a callable from a source string.
permission / get_permission for declaring which language features (assignment, function definition, import) the executed source may use.
sandbox_call / maybe_sandbox_call for executing a callable in a child process with a timeout.
CodeError / SerializationError for errors raised by the above APIs.

NO_TYPE_ANNOTATION `module-attribute` ¶

NO_TYPE_ANNOTATION = _NoTypeAnnotation()

Sentinel for make_function's return_type meaning "no annotation".

CodeError ¶

CodeError(code: str, cause: BaseException)

Bases: RuntimeError

Error raised when Python code fails to parse or execute.

Wraps the original exception (cause) together with the source code (code) and, when determinable, the 1-based line range of the offending code (lineno / end_lineno, None otherwise).

Source code in pygx/coding/_errors.py

def __init__(
    self,
    code: str,
    cause: BaseException,
):
    self.code = code
    self.cause = cause

    # Figure out the starting and ending line numbers of the erratic code.
    lineno = None
    end_lineno = None
    if isinstance(cause, SyntaxError):
        lineno = cause.lineno
        end_lineno = cause.end_lineno
    elif not isinstance(cause, TimeoutError):
        tb = sys.exc_info()[2]
        frames = traceback.extract_tb(tb, limit=5)
        for f in frames:
            if not f.filename or f.filename == '<string>':
                lineno = f.lineno
                end_lineno = lineno
                break
    self.lineno = lineno
    self.end_lineno = end_lineno

code_lines ¶

code_lines(start_line: int, end_line: int | None)

Returns the code lines from start_line (0-based) to end_line.

Source code in pygx/coding/_errors.py

def code_lines(self, start_line: int, end_line: int | None):
    """Returns the code lines from `start_line` (0-based) to `end_line`."""
    return '\n'.join(self.code.split('\n')[start_line:end_line])

format ¶

format(include_complete_code: bool = True)

Formats the error, optionally with the complete source appended.

Source code in pygx/coding/_errors.py

def format(self, include_complete_code: bool = True):
    """Formats the error, optionally with the complete source appended."""
    r = io.StringIO()
    error_message = str(self.cause).rstrip()
    if 'line' not in error_message and self.lineno is not None:
        error_message += f' (<unknown>, line {self.lineno})'
    r.write(
        formatting.colored(
            f'{self.cause.__class__.__name__}: {error_message}', 'magenta'
        )
    )

    if self.lineno is not None:
        r.write('\n\n')
        r.write(
            textwrap.indent(
                formatting.colored(
                    self.code_lines(self.lineno - 1, self.end_lineno),
                    'magenta',
                ),
                ' ' * 2,
            )
        )
        r.write('\n')

    if include_complete_code:
        r.write('\n')
        r.write(formatting.colored('[Code]', 'green', styles=['bold']))
        r.write('\n\n')
        r.write(formatting.colored('  ```python\n', 'green'))
        r.write(
            textwrap.indent(formatting.colored(self.code, 'green'), ' ' * 2)
        )
        r.write(formatting.colored('\n  ```\n', 'green'))
    return r.getvalue()

SerializationError ¶

SerializationError(message: str | None, cause: BaseException)

Bases: RuntimeError

Error raised when a value cannot be serialized or deserialized.

Raised by the sandboxed-execution APIs when a payload or result cannot cross the process boundary. Wraps the original exception as cause.

Source code in pygx/coding/_errors.py

def __init__(self, message: str | None, cause: BaseException):
    self.message = message
    self.cause = cause

CodePermission ¶

Bases: Flag

Flags for language constructs permitted during code execution.

Flags can be combined with |. Two combinations are predefined: CodePermission.BASIC (ASSIGN | CALL) and CodePermission.ALL.

ASSIGN `class-attribute` `instance-attribute` ¶

ASSIGN = enum.auto()

Allows assignment statements — plain (x = y), augmented (x += y) and annotated (x: int = y).

CONDITION `class-attribute` `instance-attribute` ¶

CONDITION = enum.auto()

Allows if and match statements.

LOOP `class-attribute` `instance-attribute` ¶

LOOP = enum.auto()

Allows for / while loops, plus with / async for / async with blocks.

CALL `class-attribute` `instance-attribute` ¶

CALL = enum.auto()

Allows function and method calls.

EXCEPTION `class-attribute` `instance-attribute` ¶

EXCEPTION = enum.auto()

Allows try, raise and assert statements.

CLASS_DEFINITION `class-attribute` `instance-attribute` ¶

CLASS_DEFINITION = enum.auto()

Allows class definitions.

FUNCTION_DEFINITION `class-attribute` `instance-attribute` ¶

FUNCTION_DEFINITION = enum.auto()

Allows def / lambda definitions (and return / yield).

IMPORT `class-attribute` `instance-attribute` ¶

IMPORT = enum.auto()

Allows import and from ... import statements.

context ¶

context(**kwargs)

Context manager to inject symbols for code execution.

Symbols injected here become part of the globals for pg.coding.evaluate and pg.coding.run calls made within the scope. Nested scopes add to (and may override) symbols from outer scopes.

Example:

with pg.coding.context(z=10):
  assert pg.coding.evaluate('z * 2') == 20

Source code in pygx/coding/_execution.py

@contextlib.contextmanager
def context(**kwargs):
    """Context manager to inject symbols for code execution.

    Symbols injected here become part of the globals for `pg.coding.evaluate`
    and `pg.coding.run` calls made within the scope. Nested scopes add to
    (and may override) symbols from outer scopes.

    **Example:**

    ```python
    with pg.coding.context(z=10):
      assert pg.coding.evaluate('z * 2') == 20
    ```
    """
    ctx = get_context()
    ctx.update(kwargs)
    concurrent.thread_local_push(_TLS_CODE_RUN_CONTEXT, ctx)

    try:
        yield ctx
    finally:
        concurrent.thread_local_pop(_TLS_CODE_RUN_CONTEXT)

evaluate ¶

evaluate(
    code: str,
    *,
    global_vars: dict[str, Any] | None = None,
    permission: CodePermission | None = None,
    returns_stdout: bool = False,
    outputs_intermediate: bool = False
) -> Any | dict[str, Any]

Executes Python code in the current process.

The code is parsed and validated against permission before execution, which limits the language constructs (assignment, loops, imports, etc.) the code may use. Note that permission control is not a security boundary: the code still runs with full access to the current process. Use pg.coding.run with sandbox=True to execute in a child process.

Example:

assert pg.coding.evaluate('x = 1\ny = x + 2\ny + 1') == 4

Intermediate variables can be exposed as well:

assert pg.coding.evaluate(
    'x = 1\nx + 2', outputs_intermediate=True
) == {'x': 1, '__result__': 3, '__stdout__': ''}

Parameters:

Name	Type	Description	Default
`code`	`str`	Python code to run.	required
`global_vars`	`dict[str, Any] \| None`	An optional dict as the globals that could be referenced by the code, merged over the symbols injected via `pg.coding.context`.	`None`
`permission`	`CodePermission \| None`	Permission for the Python code to run. If None, the permission from the current `pg.coding.permission` context is used; if there is none either, all constructs are allowed.	`None`
`returns_stdout`	`bool`	If True, the captured stdout (a str) will be returned.	`False`
`outputs_intermediate`	`bool`	Applicable when returns_stdout is False. If True, a dict will be returned containing the variables defined or updated by the code, with the last statement's value accessible by key 'result' and the captured stdout accessible by key 'stdout'. Otherwise the value of the last statement will be returned.	`False`

Returns:

Type	Description
`Any \| dict[str, Any]`	The value of the last statement of the code block if it is an
`Any \| dict[str, Any]`	expression or an assignment (the assigned value); otherwise the value
`Any \| dict[str, Any]`	of the last defined variable. Or the dict of intermediate variables
`Any \| dict[str, Any]`	described under `outputs_intermediate`. Or the captured stdout as a
`Any \| dict[str, Any]`	str when `returns_stdout` is True. Note (by design): when the code
`Any \| dict[str, Any]`	defines no variable at all and does not end in an expression, "the
`Any \| dict[str, Any]`	last defined variable" falls back to the last symbol in the
`Any \| dict[str, Any]`	execution globals — which can be a value injected via
`Any \| dict[str, Any]`	`pg.coding.context` / `global_vars` rather than something the code
`Any \| dict[str, Any]`	produced. End the code with an expression to make the result
`Any \| dict[str, Any]`	explicit.

Raises:

Type	Description
`CodeError`	If the code fails to parse, uses a construct not granted by `permission`, or raises during execution.

Source code in pygx/coding/_execution.py

def evaluate(
    code: str,
    *,
    global_vars: dict[str, Any] | None = None,
    permission: permissions.CodePermission | None = None,
    returns_stdout: bool = False,
    outputs_intermediate: bool = False,
) -> Any | dict[str, Any]:
    """Executes Python code in the current process.

    The code is parsed and validated against `permission` before execution,
    which limits the language constructs (assignment, loops, imports, etc.)
    the code may use. Note that permission control is not a security boundary:
    the code still runs with full access to the current process. Use
    `pg.coding.run` with `sandbox=True` to execute in a child process.

    **Example:**

    ```python
    assert pg.coding.evaluate('x = 1\\ny = x + 2\\ny + 1') == 4
    ```

    Intermediate variables can be exposed as well:

    ```python
    assert pg.coding.evaluate(
        'x = 1\\nx + 2', outputs_intermediate=True
    ) == {'x': 1, '__result__': 3, '__stdout__': ''}
    ```

    Args:
      code: Python code to run.
      global_vars: An optional dict as the globals that could be referenced by
        the code, merged over the symbols injected via `pg.coding.context`.
      permission: Permission for the Python code to run. If None, the
        permission from the current `pg.coding.permission` context is used;
        if there is none either, all constructs are allowed.
      returns_stdout: If True, the captured stdout (a str) will be returned.
      outputs_intermediate: Applicable when returns_stdout is False. If True,
        a dict will be returned containing the variables defined or updated by
        the code, with the last statement's value accessible by key
        '__result__' and the captured stdout accessible by key '__stdout__'.
        Otherwise the value of the last statement will be returned.

    Returns:
      The value of the last statement of the code block if it is an
      expression or an assignment (the assigned value); otherwise the value
      of the last defined variable. Or the dict of intermediate variables
      described under `outputs_intermediate`. Or the captured stdout as a
      str when `returns_stdout` is True. Note (by design): when the code
      defines no variable at all and does not end in an expression, "the
      last defined variable" falls back to the last symbol in the
      execution globals — which can be a value injected via
      `pg.coding.context` / `global_vars` rather than something the code
      produced. End the code with an expression to make the result
      explicit.

    Raises:
      CodeError: If the code fails to parse, uses a construct not granted by
        `permission`, or raises during execution.
    """
    # Set up the permission and context.
    permission = permission or permissions.get_permission()
    ctx = dict(get_context())
    if global_vars:
        ctx.update(global_vars)

    # Parse the code str.
    code_block = parsing.parse(code, permission)
    global_vars, orig_global_vars = ctx, ctx.copy()

    # No code.
    if not code_block.body:  # pytype: disable=attribute-error
        return {} if outputs_intermediate else None

    stdout = io.StringIO()
    with contextlib.redirect_stdout(stdout):
        if hasattr(
            code_block.body[-1], 'value'
        ):  # pytype: disable=attribute-error
            last_expr = code_block.body.pop()  # pytype: disable=attribute-error
            result_vars = [RESULT_KEY]

            if isinstance(last_expr, ast.Assign):
                for name_node in last_expr.targets:
                    if isinstance(name_node, ast.Name):
                        result_vars.append(name_node.id)

            last_expr = ast.Expression(getattr(last_expr, 'value'))

            try:
                # Execute the lines before the last expression.
                # NOTE(daiyip): Only a `globals` dict is specified here, which will also
                # be used to output intermediate values by `exec`. We do not specify a
                # separate `locals` dict here, for - "If exec gets two separate objects
                # as globals and locals, the code will be executed as if it were
                # embedded in a class definition." - as the Python document explains.
                # The outcome is that new functions defined in the code block could not
                # be called by other newly defined functions.
                # Refer to https://stackoverflow.com/questions/
                # 73940751/why-cant-i-call-a-function-from-another-function-using-exec
                # for more details.
                exec(compile(code_block, '', mode='exec'), global_vars)  # pylint: disable=exec-used

                # Evaluate the last expression.
                result = eval(  # pylint: disable=eval-used
                    compile(last_expr, '', mode='eval'), global_vars
                )
            except BaseException as e:
                raise errors.CodeError(code, e) from e

            for result_var in result_vars:
                global_vars[result_var] = result
        else:
            try:
                exec(compile(code_block, '', mode='exec'), global_vars)  # pylint: disable=exec-used
            except BaseException as e:
                raise errors.CodeError(code, e) from e
            global_vars[RESULT_KEY] = list(global_vars.values())[-1]

    if returns_stdout:
        return stdout.getvalue()
    if outputs_intermediate:
        outputs = {}
        for k, v in global_vars.items():
            if k == '__builtins__':
                continue
            if k not in orig_global_vars or v is not orig_global_vars[k]:
                outputs[k] = v
        # Add stdout to outputs.
        outputs[STDOUT_KEY] = stdout.getvalue()
        return outputs
    return global_vars[RESULT_KEY]

get_context ¶

get_context() -> dict[str, Any]

Gets the symbols injected by the current pg.coding.context scope.

Source code in pygx/coding/_execution.py

def get_context() -> dict[str, Any]:
    """Gets the symbols injected by the current `pg.coding.context` scope."""
    context_stack = concurrent.thread_local_get(_TLS_CODE_RUN_CONTEXT, None)
    return dict(context_stack[-1]) if context_stack else {}

maybe_sandbox_call ¶

maybe_sandbox_call(
    func: Callable[..., Any],
    *args: Any,
    sandbox: bool | None = None,
    timeout: float | None = None,
    **kwargs: Any
) -> Any

Calls a function, optionally in a sandbox process.

Parameters:

Name	Type	Description	Default
`func`	`Callable[..., Any]`	Function to call.	required
`*args`	`Any`	Positional args that will be passed to `func`.	`()`
`sandbox`	`bool \| None`	If True, calls `func` in a child process (see `pg.coding.sandbox_call`); if False, calls `func` in the current process. If None, tries the child process first, and falls back to calling `func` in the current process when the call (or its result) cannot be serialized across the process boundary.	`None`
`timeout`	`float \| None`	Execution timeout in seconds. If None, waits for `func` to complete.	`None`
`**kwargs`	`Any`	Keyword args that will be passed to `func`.	`{}`

Returns:

Type	Description
`Any`	The return value of `func`.

Raises:

Type	Description
`TimeoutError`	If the execution time exceeds the timeout.
`Exception`	Exception raised from `func`.

Source code in pygx/coding/_execution.py

def maybe_sandbox_call(
    func: Callable[..., Any],
    *args: Any,
    sandbox: bool | None = None,
    timeout: float | None = None,
    **kwargs: Any,
) -> Any:
    """Calls a function, optionally in a sandbox process.

    Args:
      func: Function to call.
      *args: Positional args that will be passed to `func`.
      sandbox: If True, calls `func` in a child process (see
        `pg.coding.sandbox_call`); if False, calls `func` in the current
        process. If None, tries the child process first, and falls back to
        calling `func` in the current process when the call (or its result)
        cannot be serialized across the process boundary.
      timeout: Execution timeout in seconds. If None, waits for `func` to
        complete.
      **kwargs: Keyword args that will be passed to `func`.

    Returns:
      The return value of `func`.

    Raises:
      TimeoutError: If the execution time exceeds the timeout.
      Exception: Exception raised from `func`.
    """
    if sandbox is None:
        try:
            return sandbox_call(func, *args, timeout=timeout, **kwargs)
        # NOTE(daiyip): output could be serialized across processes, giving it
        # already finishes on sandbox, so it should be much safer to run under
        # current process.
        except errors.SerializationError:
            return func(*args, **kwargs)
    elif sandbox:
        return sandbox_call(func, *args, timeout=timeout, **kwargs)
    else:
        return func(*args, **kwargs)

run ¶

run(
    code: str,
    *,
    global_vars: dict[str, Any] | None = None,
    permission: CodePermission | None = None,
    returns_stdout: bool = False,
    outputs_intermediate: bool = False,
    sandbox: bool | None = None,
    timeout: float | None = None
) -> Any | dict[str, Any]

Executes Python code, optionally in a sandbox process.

Same as pg.coding.evaluate, with the option to execute the code in a child process (sandbox) and to bound its execution time (timeout).

Example:

assert pg.coding.run('x = 1\nx + 2', sandbox=False) == 3

Parameters:

Name	Type	Description	Default
`code`	`str`	Python code to run.	required
`global_vars`	`dict[str, Any] \| None`	An optional dict as the globals that could be referenced by the code, merged over the symbols injected via `pg.coding.context`.	`None`
`permission`	`CodePermission \| None`	Permission for the Python code to run. If None, the permission from the current `pg.coding.permission` context is used; if there is none either, all constructs are allowed.	`None`
`returns_stdout`	`bool`	If True, the captured stdout (a str) will be returned.	`False`
`outputs_intermediate`	`bool`	Applicable when returns_stdout is False. If True, a dict will be returned containing the variables defined or updated by the code, with the last statement's value accessible by key 'result' and the captured stdout accessible by key 'stdout'. Otherwise the value of the last statement will be returned.	`False`
`sandbox`	`bool \| None`	If True, run code in a child process; if False, run code in the current process. If None, try the child process first and fall back to the current process when the code (or its result) cannot be serialized across the process boundary.	`None`
`timeout`	`float \| None`	Execution timeout in seconds. If None, waits for the code to complete.	`None`

Returns:

Type	Description
`Any \| dict[str, Any]`	The value of the last statement of the code block, the dict of
`Any \| dict[str, Any]`	intermediate variables, or the captured stdout — see
`Any \| dict[str, Any]`	`pg.coding.evaluate` for the exact shape per flag.

Raises:

Type	Description
`CodeError`	If the code fails to parse, uses a construct not granted by `permission`, or raises during execution.
`TimeoutError`	If the execution time exceeds the timeout.

Source code in pygx/coding/_execution.py

def run(
    code: str,
    *,
    global_vars: dict[str, Any] | None = None,
    permission: permissions.CodePermission | None = None,
    returns_stdout: bool = False,
    outputs_intermediate: bool = False,
    sandbox: bool | None = None,
    timeout: float | None = None,
) -> Any | dict[str, Any]:
    """Executes Python code, optionally in a sandbox process.

    Same as `pg.coding.evaluate`, with the option to execute the code in a
    child process (`sandbox`) and to bound its execution time (`timeout`).

    **Example:**

    ```python
    assert pg.coding.run('x = 1\\nx + 2', sandbox=False) == 3
    ```

    Args:
      code: Python code to run.
      global_vars: An optional dict as the globals that could be referenced by
        the code, merged over the symbols injected via `pg.coding.context`.
      permission: Permission for the Python code to run. If None, the
        permission from the current `pg.coding.permission` context is used;
        if there is none either, all constructs are allowed.
      returns_stdout: If True, the captured stdout (a str) will be returned.
      outputs_intermediate: Applicable when returns_stdout is False. If True,
        a dict will be returned containing the variables defined or updated by
        the code, with the last statement's value accessible by key
        '__result__' and the captured stdout accessible by key '__stdout__'.
        Otherwise the value of the last statement will be returned.
      sandbox: If True, run code in a child process; if False, run code in
        the current process. If None, try the child process first and fall
        back to the current process when the code (or its result) cannot be
        serialized across the process boundary.
      timeout: Execution timeout in seconds. If None, waits for the code to
        complete.

    Returns:
      The value of the last statement of the code block, the dict of
      intermediate variables, or the captured stdout — see
      `pg.coding.evaluate` for the exact shape per flag.

    Raises:
      CodeError: If the code fails to parse, uses a construct not granted by
        `permission`, or raises during execution.
      TimeoutError: If the execution time exceeds the timeout.
    """
    return maybe_sandbox_call(
        evaluate,
        code=code,
        global_vars=global_vars,
        permission=permission,
        returns_stdout=returns_stdout,
        outputs_intermediate=outputs_intermediate,
        sandbox=sandbox,
        timeout=timeout,
    )

sandbox_call ¶

sandbox_call(
    func: Callable[..., Any],
    *args: Any,
    timeout: float | None = None,
    **kwargs: Any
) -> Any

Calls a function in a child process, with an optional timeout.

The function, its arguments and its return value are shipped between the parent and the child process via pickling (falling back to the optional cloudpickle for lambdas, closures and locally-defined callables). On timeout, the child process is killed.

Parameters:

Name	Type	Description	Default
`func`	`Callable[..., Any]`	Function to call.	required
`*args`	`Any`	Positional arguments for `func`.	`()`
`timeout`	`float \| None`	Execution timeout in seconds. If None, waits for `func` to complete.	`None`
`**kwargs`	`Any`	Keyword arguments for `func`.	`{}`

Returns:

Type	Description
`Any`	Return value from `func`.

Raises:

Type	Description
`TimeoutError`	If the execution time exceeds the timeout.
`SerializationError`	If `func` (or its arguments) cannot be shipped to the sandbox subprocess, or its return value cannot be shipped back.
`Exception`	Exception raised from `func`.

Source code in pygx/coding/_execution.py

def sandbox_call(
    func: Callable[..., Any],
    *args: Any,
    timeout: float | None = None,
    **kwargs: Any,
) -> Any:
    """Calls a function in a child process, with an optional timeout.

    The function, its arguments and its return value are shipped between the
    parent and the child process via pickling (falling back to the optional
    ``cloudpickle`` for lambdas, closures and locally-defined callables). On
    timeout, the child process is killed.

    Args:
      func: Function to call.
      *args: Positional arguments for `func`.
      timeout: Execution timeout in seconds. If None, waits for `func` to
        complete.
      **kwargs: Keyword arguments for `func`.

    Returns:
      Return value from `func`.

    Raises:
      TimeoutError: If the execution time exceeds the timeout.
      SerializationError: If `func` (or its arguments) cannot be shipped to
        the sandbox subprocess, or its return value cannot be shipped back.
      Exception: Exception raised from `func`.
    """
    # Serialize the payload up front (in the parent) so a non-picklable target
    # surfaces as a `SerializationError` here rather than as an opaque child
    # crash, and so `maybe_sandbox_call`'s automatic fallback can catch it.
    payload = _dumps_payload((func, args, kwargs, get_context()))

    ctx = multiprocessing.get_context(sandbox_start_method())
    q = ctx.Queue()
    # `get_context(str)` is typed to return the loosely-typed `BaseContext`
    # (only literal method names resolve to a concrete context exposing
    # `Process`); the attribute exists on every concrete context at runtime.
    p = ctx.Process(  # pyright: ignore[reportAttributeAccessIssue]
        target=_sandbox_target, args=(payload, q)
    )

    def _wait_result() -> Any:
        # Poll instead of a single blocking `get`: a child that dies without
        # ever posting a result — most commonly the 'spawn'/'forkserver'
        # bootstrap re-importing an UNGUARDED `__main__` that calls back into
        # `run(sandbox=True)`, which the child aborts with a bootstrapping
        # error — used to hang this call until `timeout` elapsed (forever
        # when None) (#773 ruling 11).
        deadline = None if timeout is None else time.time() + timeout
        while True:
            wait = 0.2
            if deadline is not None:
                remaining = deadline - time.time()
                if remaining <= 0:
                    raise TimeoutError(
                        f'Execution time exceed {timeout} seconds.'
                    )
                wait = min(wait, remaining)
            try:
                return q.get(timeout=wait)
            except queue.Empty:
                if p.exitcode is not None:
                    # The child exited; drain once more in case the result
                    # raced the exit, then fail fast.
                    try:
                        return q.get_nowait()
                    except queue.Empty:
                        raise ChildProcessError(
                            f'Sandbox subprocess exited (exitcode='
                            f'{p.exitcode}) without returning a result. If '
                            f'this call runs at import time of a script, '
                            f"guard it with `if __name__ == '__main__':` — "
                            f'the {sandbox_start_method()!r} start method '
                            f're-imports `__main__` in the child.'
                        ) from None

    try:
        p.start()
        x = _wait_result()
    except TimeoutError:
        if p.is_alive():
            # We use `kill` instead of `terminate` to release process resources
            # right away.
            p.kill()
        raise
    finally:
        q.close()

    if isinstance(x, Exception):
        raise x
    try:
        return pickle.loads(x)
    except Exception as e:
        raise errors.SerializationError(
            'Cannot deserialize the output from sandbox.', e
        ) from e

make_function ¶

make_function(
    name: str,
    args: list[str],
    body: list[str | list],
    *,
    exec_globals: dict[str, Any] | None = None,
    exec_locals: dict[str, Any] | None = None,
    return_type: Any = NO_TYPE_ANNOTATION
)

Creates a function dynamically from source.

Example:

add = pg.coding.make_function(
    'add',
    args=['x: int', 'y: int'],
    body=['return x + y'],
    return_type=int,
)
assert add(1, 2) == 3

Parameters:

Name	Type	Description	Default
`name`	`str`	The function name.	required
`args`	`list[str]`	The argument declarations, e.g. `['x', 'y: int', 'z=1']`.	required
`body`	`list[str \| list]`	The source lines of the function body, without indentation. A nested list is one indent level deeper, so a block reads as a block: `['if x:', ['return 1'], 'return 0']`. Nesting may go any depth.	required
`exec_globals`	`dict[str, Any] \| None`	The globals available to the generated function.	`None`
`exec_locals`	`dict[str, Any] \| None`	Local symbols available to the function definition (e.g. types referenced by annotations).	`None`
`return_type`	`Any`	The return type annotation of the function, or `NO_TYPE_ANNOTATION` for none.	`NO_TYPE_ANNOTATION`

The created function object is returned.

Source code in pygx/coding/_function_generation.py

def make_function(
    name: str,
    args: list[str],
    body: 'list[str | list]',
    *,
    exec_globals: dict[str, Any] | None = None,
    exec_locals: dict[str, Any] | None = None,
    return_type: Any = NO_TYPE_ANNOTATION,
):
    """Creates a function dynamically from source.

    **Example:**

    ```python
    add = pg.coding.make_function(
        'add',
        args=['x: int', 'y: int'],
        body=['return x + y'],
        return_type=int,
    )
    assert add(1, 2) == 3
    ```

    Args:
      name: The function name.
      args: The argument declarations, e.g. ``['x', 'y: int', 'z=1']``.
      body: The source lines of the function body, without indentation. A
        nested list is one indent level deeper, so a block reads as a block:
        ``['if x:', ['return 1'], 'return 0']``. Nesting may go any depth.
      exec_globals: The globals available to the generated function.
      exec_locals: Local symbols available to the function definition (e.g.
        types referenced by annotations).
      return_type: The return type annotation of the function, or
        `NO_TYPE_ANNOTATION` for none.

    The created function object is returned.
    """
    if exec_locals is None:
        exec_locals = {}
    if return_type != NO_TYPE_ANNOTATION:
        exec_locals['_return_type'] = return_type
        return_annotation = '->_return_type'
    else:
        return_annotation = ''
    args_str = ', '.join(args)

    def _render(lines, depth: int) -> list[str]:
        """Flatten `body`, indenting one level per nesting level."""
        pad = '  ' * depth
        out = []
        for line in lines:
            if isinstance(line, list):
                out.extend(_render(line, depth + 1))
            else:
                out.append(f'{pad}{line}')
        return out

    body_str = '\n'.join(_render(body, 1))
    fn_def = f' def {name}({args_str}){return_annotation}:\n{body_str}'
    local_vars = ', '.join(exec_locals.keys())
    fn_def = f'def _make_fn({local_vars}):\n{fn_def}\n return {name}'
    ns: dict[str, Any] = {}
    exec(fn_def, exec_globals, ns)  # pylint: disable=exec-used
    return ns['_make_fn'](**exec_locals)

parse ¶

parse(code: str, permission: CodePermission | None = None) -> Module

Parses Python source code, optionally enforcing a permission.

Parameters:

Name	Type	Description	Default
`code`	`str`	Python source code.	required
`permission`	`CodePermission \| None`	If not None, the parsed AST is validated so that `code` only uses the language constructs granted by the permission.	`None`

Returns:

Type	Description
`Module`	The parsed `ast.Module`.

Raises:

Type	Description
`CodeError`	If `code` has a syntax error, or uses a language construct not granted by `permission`.

Source code in pygx/coding/_parsing.py

def parse(
    code: str,
    permission: permissions.CodePermission | None = None,
) -> ast.Module:
    """Parses Python source code, optionally enforcing a permission.

    Args:
      code: Python source code.
      permission: If not None, the parsed AST is validated so that `code`
        only uses the language constructs granted by the permission.

    Returns:
      The parsed `ast.Module`.

    Raises:
      CodeError: If `code` has a syntax error, or uses a language construct
        not granted by `permission`.
    """
    try:
        parsed_code = ast.parse(code, mode='exec')
        if permission is not None:
            _CodeValidator(code, permission).visit(parsed_code)
    except SyntaxError as e:
        raise errors.CodeError(code, e) from e
    return parsed_code

get_permission ¶

get_permission() -> CodePermission | None

Gets the current permission for code execution.

Returns None when no pg.coding.permission context is active, in which case code execution is unrestricted.

Source code in pygx/coding/_permissions.py

def get_permission() -> CodePermission | None:
    """Gets the current permission for code execution.

    Returns None when no `pg.coding.permission` context is active, in which
    case code execution is unrestricted.
    """
    return concurrent.thread_local_get(_TLS_CODE_RUN_PERMISSION, None)

permission ¶

permission(perm: CodePermission) -> Iterator[CodePermission]

Context manager for controlling the permission for code execution.

When the permission context manager is nested, the outermost permission will be used. This design allows users to control permission at the top level.

Example:

with pg.coding.permission(pg.coding.CodePermission.BASIC):
  assert pg.coding.evaluate('x = 1\nx + 1') == 2
  # Raises `pg.coding.CodeError`: `import` is not allowed.
  pg.coding.evaluate('import os')

Parameters:

Name	Type	Description	Default
`perm`	`CodePermission`	Code execution permission.	required

Yields:

Type	Description
`CodePermission`	Actual permission applied.

Source code in pygx/coding/_permissions.py

@contextlib.contextmanager
def permission(perm: CodePermission) -> Iterator[CodePermission]:
    """Context manager for controlling the permission for code execution.

    When the `permission` context manager is nested, the outermost permission
    will be used. This design allows users to control permission at the top
    level.

    **Example:**

    ```python
    with pg.coding.permission(pg.coding.CodePermission.BASIC):
      assert pg.coding.evaluate('x = 1\\nx + 1') == 2
      # Raises `pg.coding.CodeError`: `import` is not allowed.
      pg.coding.evaluate('import os')
    ```

    Args:
      perm: Code execution permission.

    Yields:
      Actual permission applied.
    """

    outter_perm = concurrent.thread_local_get(_TLS_CODE_RUN_PERMISSION, None)

    # Use the top-level permission as the actual permission
    if outter_perm is not None:
        perm = outter_perm

    concurrent.thread_local_set(_TLS_CODE_RUN_PERMISSION, perm)

    try:
        yield perm
    finally:
        if outter_perm is None:
            concurrent.thread_local_del(_TLS_CODE_RUN_PERMISSION)

options: show_root_heading: false show_root_toc_entry: false members_order: source heading_level: 2

pygx.coding¶

coding ¶

NO_TYPE_ANNOTATION module-attribute ¶

CodeError ¶

code_lines ¶

format ¶

SerializationError ¶

CodePermission ¶

ASSIGN class-attribute instance-attribute ¶

CONDITION class-attribute instance-attribute ¶

LOOP class-attribute instance-attribute ¶

CALL class-attribute instance-attribute ¶

EXCEPTION class-attribute instance-attribute ¶

CLASS_DEFINITION class-attribute instance-attribute ¶

FUNCTION_DEFINITION class-attribute instance-attribute ¶

IMPORT class-attribute instance-attribute ¶

context ¶

evaluate ¶

get_context ¶

maybe_sandbox_call ¶

run ¶

sandbox_call ¶

make_function ¶

parse ¶

get_permission ¶

permission ¶

`pygx.coding`¶

NO_TYPE_ANNOTATION `module-attribute` ¶

ASSIGN `class-attribute` `instance-attribute` ¶

CONDITION `class-attribute` `instance-attribute` ¶

LOOP `class-attribute` `instance-attribute` ¶

CALL `class-attribute` `instance-attribute` ¶

EXCEPTION `class-attribute` `instance-attribute` ¶

CLASS_DEFINITION `class-attribute` `instance-attribute` ¶

FUNCTION_DEFINITION `class-attribute` `instance-attribute` ¶

IMPORT `class-attribute` `instance-attribute` ¶