Run Code in a Subinterpreter and Check the Result

A subinterpreter is a second Python interpreter inside the same process. It has its own imported modules and its own global variables, so code running in it cannot see yours. Python 3.14 added the concurrent.interpreters module to create one, run code in it, and pass values back through a queue:

from concurrent import interpreters

queue = interpreters.create_queue()
worker = interpreters.create()
try:
    worker.prepare_main(q=queue, x=7)  # set globals in the worker
    worker.exec("q.put(x * x)")        # run code there, on this thread
    print(queue.get(timeout=5))        # 49
finally:
    worker.close()

exec runs the code on your thread and returns when it finishes. The next article runs several interpreters in parallel. Interpreters share one process, so they are no sandbox. Run only code you trust in them.

The code is a string, so your type checker never sees what it puts on the queue. You will wrap the queue so that get() checks each value as it arrives.

Write the queue wrapper

TypedQueue[T] wraps a raw interpreters.Queue so that get() returns a T or raises.

• TypedQueue(item_type, raw) stores both. item_type is a plain class such as int or str. A generic type such as list[int] does not work with isinstance.
• The read-only raw property returns the same queue object you passed in.
• put(item) passes the item to the raw queue, with no runtime check.
• get(timeout=5) calls the raw queue's get with that timeout. Return the item if isinstance(item, item_type) is true, and raise TypeError if not. Let queue errors, such as a timeout, propagate.

isinstance accepts subclasses, so a TypedQueue of int lets True through. It checks only the outer class, not the items inside a list.

Run one program

Write run_isolated(channel, code, **shared). It takes these steps in order:

1. Create an interpreter with interpreters.create().
2. Call prepare_main to bind channel.raw as q, plus every name in shared.
3. Run code with exec.
4. Return one checked result from channel.get().
5. Close the interpreter in a finally block, so it closes even when step 2, 3 or 4 raises.

Read the result before you close the interpreter. When an interpreter closes, an unread item it put on a queue turns into a placeholder called UNBOUND. A return inside the try runs before the finally, so return channel.get() reads the result before the interpreter closes.

You can assume that the code puts exactly one result on an otherwise unused queue, and that shared never contains q. prepare_main takes simple values such as numbers and strings, and raises NotShareableError for a list. The timeout in get cannot stop code that never finishes.

Write the report

Write solve(*, code, values, failing_code):

1. Record len(interpreters.list_all()), the number of live interpreters.
2. Make a TypedQueue[int] around interpreters.create_queue(). For each integer in values, call run_isolated with x bound to it. Sort the results into a tuple, keeping duplicates. Empty values gives ().
3. Call run_isolated with failing_code and the same queue. That code raises before it puts anything, so exec raises interpreters.ExecutionFailed. Catch it and record exc.excinfo.type.__name__, the original exception's class name. If nothing raises, record "".
4. In your own interpreter, put {"copied": "by pickle"} on a second raw queue and get it back. Record whether the result equals the original. The queue pickles the dictionary, so you get a copy.
5. Return (sorted_results, change, failure_name, dict_ok). change is the live interpreter count now minus the count from step 1. It is 0 when you closed every interpreter.

solve(code="q.put(x * x)", values=[3, 1, 3], failing_code="raise ValueError('boom')")
# ((1, 9, 9), 0, 'ValueError', True)