// Return True if the current Python executable supports JIT compilation,

// and False otherwise.

#[pyfunction]

const fn is_available() -> bool {

false // RustPython has no JIT

}

// Return True if JIT compilation is enabled for the current Python process,

// and False otherwise.

#[pyfunction]

const fn is_enabled() -> bool {

false // RustPython has no JIT

}

// Return True if the topmost Python frame is currently executing JIT code,

// and False otherwise.

#[pyfunction]

const fn is_active() -> bool {

false // RustPython has no JIT

}

use crate::{

AsObject, Py, PyObject, PyObjectRef, PyPayload, PyRef, PyRefExact, PyResult,

builtins::{

PyBaseExceptionRef, PyDictRef, PyFrozenSet, PyNamespace, PyStr, PyStrRef, PyTuple,

PyTupleRef, PyTypeRef, PyUtf8StrRef,

},

common::{

ascii,

hash::{PyHash, PyUHash},

},

convert::ToPyObject,

frame::FrameObjectRef,

function::{FuncArgs, KwArgs, OptionalArg, PosArgs},

stdlib::{_warnings::warn, builtins},

types::PyStructSequence,

version,

vm::{Settings, VirtualMachine},

};

use core::ffi::CStr;

use core::sync::atomic::Ordering;

use num_traits::ToPrimitive;

use std::{

env,

io::{IsTerminal, Read, Write},

};

// Rust target triple (e.g., "x86_64-unknown-linux-gnu")

pub(crate) const RUST_MULTIARCH: &str = env!("RUSTPYTHON_TARGET_TRIPLE");

/// Convert Rust target triple to CPython-style multiarch

/// e.g., "x86_64-unknown-linux-gnu" -> "x86_64-linux-gnu"

pub(crate) fn multiarch() -> String {

RUST_MULTIARCH.replace("-unknown", "")

}

#[pymodule(name = "monitoring", with(super::monitoring::sys_monitoring))]

pub(super) mod monitoring {}

#[pyclass(no_attr, name = "_BootstrapStderr")]

#[derive(Debug, PyPayload)]

pub(super) struct BootstrapStderr;

#[pyclass(with(Py))]

impl BootstrapStderr {}

#[pyclass]

impl Py<BootstrapStderr> {

#[pymethod]

fn write(&self, s: PyStrRef) -> usize {

let bytes = s.as_bytes();

let _ = std::io::stderr().write_all(bytes);

bytes.len()

}

#[pymethod]

fn flush(&self) {

let _ = std::io::stderr().flush();

}

}

// Lightweight stdio wrapper for sandbox mode (no host_env).

// Directly uses Rust's std::io for stdin/stdout/stderr without FileIO.

#[pyclass(no_attr, name = "_SandboxStdio")]

#[derive(Debug, PyPayload)]

pub struct SandboxStdio {

pub fd: i32,

pub name: String,

pub mode: String,

}

#[pyclass(with(Py))]

impl SandboxStdio {}

#[pyclass]

impl Py<SandboxStdio> {

#[pymethod]

fn write(&self, s: PyStrRef, vm: &VirtualMachine) -> PyResult<usize> {

if self.fd == 0 {

return Err(vm.new_os_error("not writable"));

}

let bytes = s.as_bytes();

if self.fd == 2 {

std::io::stderr()

.write_all(bytes)

.map_err(|e| vm.new_os_error(e.to_string()))?;

} else {

std::io::stdout()

.write_all(bytes)

.map_err(|e| vm.new_os_error(e.to_string()))?;

}

Ok(bytes.len())

}

#[pymethod]

fn readline(&self, size: OptionalArg<isize>, vm: &VirtualMachine) -> PyResult<String> {

if self.fd != 0 {

return Err(vm.new_os_error("not readable"));

}

let size = size.unwrap_or(-1);

if size == 0 {

return Ok(String::new());

}

let mut line = String::new();

std::io::stdin()

.read_line(&mut line)

.map_err(|e| vm.new_os_error(e.to_string()))?;

if size > 0 {

line.truncate(size as usize);

}

Ok(line)

}

#[pymethod]

fn flush(&self, vm: &VirtualMachine) -> PyResult<()> {

match self.fd {

1 => {

std::io::stdout()

.flush()

.map_err(|e| vm.new_os_error(e.to_string()))?;

}

2 => {

std::io::stderr()

.flush()

.map_err(|e| vm.new_os_error(e.to_string()))?;

}

_ => {}

}

Ok(())

}

#[pymethod]

fn fileno(&self) -> i32 {

self.fd

}

#[pymethod]

fn isatty(&self) -> bool {

match self.fd {

0 => std::io::stdin().is_terminal(),

1 => std::io::stdout().is_terminal(),

2 => std::io::stderr().is_terminal(),

_ => false,

}

}

#[pymethod]

fn readable(&self) -> bool {

self.fd == 0

}

#[pymethod]

fn writable(&self) -> bool {

self.fd == 1 || self.fd == 2

}

#[pygetset]

fn closed(&self) -> bool {

false

}

#[pygetset]

fn encoding(&self) -> String {

"utf-8".to_owned()

}

#[pygetset]

fn errors(&self) -> String {

if self.fd == 2 {

"backslashreplace"

} else {

"strict"

}

.to_owned()

}

#[pygetset(name = "name")]

fn name_prop(&self) -> String {

self.name.clone()

}

#[pygetset(name = "mode")]

fn mode_prop(&self) -> String {

self.mode.clone()

}

}

#[pyattr(name = "_rustpython_debugbuild")]

const RUSTPYTHON_DEBUGBUILD: bool = cfg!(debug_assertions);

#[cfg(not(windows))]

#[pyattr(name = "abiflags")]

const ABIFLAGS_ATTR: &str = "t"; // 't' for free-threaded (no GIL)

// Internal constant used for sysconfigdata_name

pub const ABIFLAGS: &str = "t";

#[pyattr(name = "api_version")]

const API_VERSION: u32 = 0x0; // what C api?

#[pyattr(name = "copyright")]

pub const COPYRIGHT: &CStr = c"Copyright (c) 2019 RustPython Team";

#[pyattr(name = "float_repr_style")]

const FLOAT_REPR_STYLE: &str = "short";

#[pyattr(name = "_framework")]

const FRAMEWORK: &str = "";

#[pyattr(name = "hexversion")]

const HEXVERSION: usize = version::VERSION_HEX;

#[pyattr(name = "maxsize")]

pub(crate) const MAXSIZE: isize = isize::MAX;

#[pyattr(name = "maxunicode")]

const MAXUNICODE: u32 = core::char::MAX as u32;

#[pyattr(name = "platform")]

pub const PLATFORM: &CStr = cfg_select! {

target_os = "linux" => c"linux",

target_os = "android" => c"android",

target_os = "macos" => c"darwin",

target_os = "ios" => c"ios",

windows => c"win32",

target_os = "wasi" => c"wasi",

_ => c"unknown"

};

#[pyattr(name = "ps1")]

const PS1: &str = ">>> ";

#[pyattr(name = "ps2")]

const PS2: &str = "... ";

#[cfg(windows)]

#[pyattr(name = "_vpath")]

const VPATH: Option<&'static str> = None; // TODO: actual VPATH value

#[cfg(windows)]

#[pyattr(name = "dllhandle")]

const DLLHANDLE: usize = 0;

#[pyattr]

fn prefix(vm: &VirtualMachine) -> String {

vm.state.config.paths.prefix.clone()

}

#[pyattr]

fn base_prefix(vm: &VirtualMachine) -> String {

vm.state.config.paths.base_prefix.clone()

}

#[pyattr]

fn exec_prefix(vm: &VirtualMachine) -> String {

vm.state.config.paths.exec_prefix.clone()

}

#[pyattr]

fn base_exec_prefix(vm: &VirtualMachine) -> String {

vm.state.config.paths.base_exec_prefix.clone()

}

#[pyattr]

fn platlibdir(_vm: &VirtualMachine) -> &'static str {

option_env!("RUSTPYTHON_PLATLIBDIR").unwrap_or("lib")

}

#[pyattr]

fn _stdlib_dir(vm: &VirtualMachine) -> PyObjectRef {

vm.state.config.paths.stdlib_dir.clone().to_pyobject(vm)

}

// alphabetical order with segments of pyattr and others

#[pyattr]

fn argv(vm: &VirtualMachine) -> Vec<PyObjectRef> {

vm.state

.config

.settings

.argv

.iter()

.map(|arg| vm.ctx.new_str(arg.clone()).into())

.collect()

}

#[pyattr]

fn builtin_module_names(vm: &VirtualMachine) -> PyTupleRef {

let mut module_names: Vec<String> =

vm.state.module_defs.keys().map(|&s| s.to_owned()).collect();

module_names.push("sys".to_owned());

module_names.push("builtins".to_owned());

module_names.sort();

vm.ctx.new_tuple(

module_names

.into_iter()

.map(|n| vm.ctx.new_utf8_str(n).into())

.collect(),

)

}

// List from cpython/Python/stdlib_module_names.h

const STDLIB_MODULE_NAMES: &[&str] = &[

"__future__",

"_abc",

"_aix_support",

"_android_support",

"_apple_support",

"_ast",

"_asyncio",

"_bisect",

"_blake2",

"_bz2",

"_codecs",

"_codecs_cn",

"_codecs_hk",

"_codecs_iso2022",

"_codecs_jp",

"_codecs_kr",

"_codecs_tw",

"_collections",

"_collections_abc",

"_colorize",

"_compat_pickle",

"_compression",

"_contextvars",

"_csv",

"_ctypes",

"_curses",

"_curses_panel",

"_datetime",

"_dbm",

"_decimal",

"_elementtree",

"_frozen_importlib",

"_frozen_importlib_external",

"_functools",

"_gdbm",

"_hashlib",

"_heapq",

"_imp",

"_interpchannels",

"_interpqueues",

"_interpreters",

"_io",

"_ios_support",

"_json",

"_locale",

"_lsprof",

"_lzma",

"_markupbase",

"_md5",

"_multibytecodec",

"_multiprocessing",

"_opcode",

"_opcode_metadata",

"_operator",

"_osx_support",

"_overlapped",

"_pickle",

"_posixshmem",

"_posixsubprocess",

"_py_abc",

"_pydatetime",

"_pydecimal",

"_pyio",

"_pylong",

"_pyrepl",

"_queue",

"_random",

"_scproxy",

"_sha1",

"_sha2",

"_sha3",

"_signal",

"_sitebuiltins",

"_socket",

"_sqlite3",

"_sre",

"_ssl",

"_stat",

"_statistics",

"_string",

"_strptime",

"_struct",

"_suggestions",

"_symtable",

"_sysconfig",

"_testinternalcapi",

"_thread",

"_threading_local",

"_tkinter",

"_tokenize",

"_tracemalloc",

"_typing",

"_uuid",

"_warnings",

"_weakref",

"_weakrefset",

"_winapi",

"_wmi",

"_zoneinfo",

"abc",

"antigravity",

"argparse",

"array",

"ast",

"asyncio",

"atexit",

"base64",

"bdb",

"binascii",

"bisect",

"builtins",

"bz2",

"cProfile",

"calendar",

"cmath",

"cmd",

"code",

"codecs",

"codeop",

"collections",

"colorsys",

"compileall",

"concurrent",

"configparser",

"contextlib",

"contextvars",

"copy",

"copyreg",

"csv",

"ctypes",

"curses",

"dataclasses",

"datetime",

"dbm",

"decimal",

"difflib",

"dis",

"doctest",

"email",

"encodings",

"ensurepip",

"enum",

"errno",

"faulthandler",

"fcntl",

"filecmp",

"fileinput",

"fnmatch",

"fractions",

"ftplib",

"functools",

"gc",

"genericpath",

"getopt",

"getpass",

"gettext",

"glob",

"graphlib",

"grp",

"gzip",

"hashlib",

"heapq",

"hmac",

"html",

"http",

"idlelib",

"imaplib",

"importlib",

"inspect",

"io",

"ipaddress",

"itertools",

"json",

"keyword",

"linecache",

"locale",

"logging",

"lzma",

"mailbox",

"marshal",

"math",

"mimetypes",

"mmap",

"modulefinder",

"msvcrt",

"multiprocessing",

"netrc",

"nt",

"ntpath",

"nturl2path",

"numbers",

"opcode",

"operator",

"optparse",

"os",

"pathlib",

"pdb",

"pickle",

"pickletools",

"pkgutil",

"platform",

"plistlib",

"poplib",

"posix",

"posixpath",

"pprint",

"profile",

"pstats",

"pty",

"pwd",

"py_compile",

"pyclbr",

"pydoc",

"pydoc_data",

"pyexpat",

"queue",

"quopri",

"random",

"re",

"readline",

"reprlib",

"resource",

"rlcompleter",

"runpy",

"sched",

"secrets",

"select",

"selectors",

"shelve",

"shlex",

"shutil",

"signal",

"site",

"smtplib",

"socket",

"socketserver",

"sqlite3",

"sre_compile",

"sre_constants",

"sre_parse",

"ssl",

"stat",

"statistics",

"string",

"stringprep",

"struct",

"subprocess",

"symtable",

"sys",

"sysconfig",

"syslog",

"tabnanny",

"tarfile",

"tempfile",

"termios",

"textwrap",

"this",

"threading",

"time",

"timeit",

"tkinter",

"token",

"tokenize",

"tomllib",

"trace",

"traceback",

"tracemalloc",

"tty",

"turtle",

"turtledemo",

"types",

"typing",

"unicodedata",

"unittest",

"urllib",

"uuid",

"venv",

"warnings",

"wave",

"weakref",

"webbrowser",

"winreg",

"winsound",

"wsgiref",

"xml",

"xmlrpc",

"zipapp",

"zipfile",

"zipimport",

"zlib",

"zoneinfo",

];

#[pyattr(once)]

fn stdlib_module_names(vm: &VirtualMachine) -> PyObjectRef {

let names = STDLIB_MODULE_NAMES

.iter()

.map(|&n| vm.ctx.new_str(n).into());

PyFrozenSet::from_iter(vm, names)

.expect("Creating stdlib_module_names frozen set must succeed")

.to_pyobject(vm)

}

#[pyattr]

fn byteorder(vm: &VirtualMachine) -> PyStrRef {

// https://doc.rust-lang.org/reference/conditional-compilation.html#target_endian

vm.ctx

.intern_str(if cfg!(target_endian = "little") {

"little"

} else if cfg!(target_endian = "big") {

"big"

} else {

"unknown"

})

.to_owned()

}

#[pyattr]

fn _base_executable(vm: &VirtualMachine) -> String {

vm.state.config.paths.base_executable.clone()

}

#[pyattr]

fn dont_write_bytecode(vm: &VirtualMachine) -> bool {

!vm.state.config.settings.write_bytecode

}

#[pyattr]

fn executable(vm: &VirtualMachine) -> String {

vm.state.config.paths.executable.clone()

}

#[pyattr]

fn _git(vm: &VirtualMachine) -> PyTupleRef {

vm.new_tuple((

ascii!("RustPython"),

version::GIT_IDENTIFIER,

version::GIT_REVISION,

))

}

#[pyattr]

fn implementation(vm: &VirtualMachine) -> PyRef<PyNamespace> {

const NAME: &str = "rustpython";

// Keep implementation-specific bytecode caches separate.

let cache_tag = format!("{NAME}-{}{}", version::MAJOR, version::MINOR);

let ctx = &vm.ctx;

py_namespace!(vm, {

"name" => ctx.new_str(NAME),

"cache_tag" => ctx.new_str(cache_tag),

"_multiarch" => ctx.new_str(multiarch()),

"version" => PyVersionInfo::from_data(VersionInfoData::IMPLEMENTATION, vm),

"hexversion" => ctx.new_int(version::VERSION_HEX_IMPL),

"supports_isolated_interpreters" =>

ctx.new_bool(crate::vm::runtime::SUPPORTS_ISOLATED_INTERPRETERS),

})

}

#[pyattr]

const fn meta_path(_vm: &VirtualMachine) -> Vec<PyObjectRef> {

Vec::new()

}

#[pyattr]

fn orig_argv(vm: &VirtualMachine) -> Vec<PyObjectRef> {

env::args().map(|arg| vm.ctx.new_str(arg).into()).collect()

}

#[pyattr]

fn path(vm: &VirtualMachine) -> Vec<PyObjectRef> {

vm.state

.config

.paths

.module_search_paths

.iter()

.map(|path| vm.ctx.new_str(path.clone()).into())

.collect()

}

#[pyattr]

const fn path_hooks(_vm: &VirtualMachine) -> Vec<PyObjectRef> {

Vec::new()

}

#[pyattr]

fn path_importer_cache(vm: &VirtualMachine) -> PyDictRef {

vm.ctx.new_dict()

}

#[pyattr]

fn pycache_prefix(vm: &VirtualMachine) -> PyObjectRef {

vm.ctx.none()

}

#[pyattr(name = "version")]

const VERSION: &str = version::RUSTPYTHON_VERSION;

// Note: This is Python DLL version in CPython, but we arbitrary fill it for compatibility

#[cfg(windows)]

#[pyattr(name = "winver")]

const WINVER: &str = version::WINVER;

#[pyattr]

fn _xoptions(vm: &VirtualMachine) -> PyDictRef {

let ctx = &vm.ctx;

let xopts = ctx.new_dict();

for (key, value) in &vm.state.config.settings.xoptions {

let value = value.as_ref().map_or_else(

|| ctx.new_bool(true).into(),

|s| ctx.new_str(s.clone()).into(),

);

xopts.set_item(&**key, value, vm).unwrap();

}

xopts

}

#[pyattr]

fn warnoptions(vm: &VirtualMachine) -> Vec<PyObjectRef> {

vm.state

.config

.settings

.warnoptions

.iter()

.map(|s| vm.ctx.new_str(s.clone()).into())

.collect()

}

#[cfg(feature = "rustpython-compiler")]

#[pyfunction]

fn _baserepl(vm: &VirtualMachine) -> PyResult<()> {

// read stdin to end

let stdin = std::io::stdin();

let mut handle = stdin.lock();

let mut source = String::new();

handle

.read_to_string(&mut source)

.map_err(|e| vm.new_os_error(format!("Error reading from stdin: {e}")))?;

vm.compile(&source, crate::compiler::Mode::Single, "<stdin>")

.map_err(|e| e.into_pyexception(vm, Some(&source)))?;

Ok(())

}

#[pyfunction]

const fn _is_gil_enabled() -> bool {

false // RustPython has no GIL (like free-threaded Python)

}

#[pyfunction]

const fn is_remote_debug_enabled() -> bool {

false // RustPython does not support remote debugging

}

#[derive(FromArgs)]

struct ExitArgs {

#[pyarg(positional, optional)]

status: Option<PyObjectRef>,

}

#[derive(FromArgs)]

struct GetFrameArgs {

#[pyarg(positional, default)]

depth: usize,

}

#[derive(FromArgs)]

struct GetFrameModuleNameArgs {

#[pyarg(any, default)]

depth: usize,

}

#[derive(FromArgs)]

struct SetMaxDigitsArgs {

#[pyarg(any)]

maxdigits: usize,

}

#[derive(FromArgs)]

struct SetDepthArgs {

#[pyarg(any)]

depth: i32,

}

#[derive(FromArgs)]

struct AuditHookArgs {

#[pyarg(any)]

hook: PyObjectRef,

}

#[pyfunction]

fn exit(args: ExitArgs, vm: &VirtualMachine) -> PyResult {

let status = args.status.unwrap_or_else(|| vm.ctx.none());

let args = if let Some(status_tuple) = status.downcast_ref::<PyTuple>() {

status_tuple.as_slice().to_vec()

} else {

vec![status]

};

Err(vm.new_system_exit(args.into()))

}

#[pyfunction]

fn call_tracing(func: PyObjectRef, args: PyTupleRef, vm: &VirtualMachine) -> PyResult {

// CPython temporarily enables tracing state around this call.

// RustPython does not currently model the full C-level tracing toggles,

// but call semantics (func(*args)) are matched.

func.call(PosArgs::new(args.as_slice().to_vec()), vm)

}

#[pyfunction]

fn exception(vm: &VirtualMachine) -> Option<PyBaseExceptionRef> {

vm.topmost_exception()

}

#[pyfunction(name = "__displayhook__")]

#[pyfunction]

fn displayhook(object: PyObjectRef, vm: &VirtualMachine) -> PyResult<()> {

// Save non-None values as "_"

if vm.is_none(&object) {

return Ok(());

}

// set to none to avoid recursion while printing

vm.builtins.set_attr("_", vm.ctx.none(), vm)?;

// TODO: catch encoding errors

let repr = object.repr(vm)?.into();

builtins::print(PosArgs::new(vec![repr]), Default::default(), vm)?;

vm.builtins.set_attr("_", object, vm)?;

Ok(())

}

#[pyfunction(name = "__excepthook__")]

#[pyfunction]

fn excepthook(

exctype: PyObjectRef,

value: PyObjectRef,

traceback: PyObjectRef,

vm: &VirtualMachine,

) -> PyResult<()> {

let stderr = super::get_stderr(vm)?;

// Keep runtime SyntaxErrors on the normal traceback path.

let has_traceback = !vm.is_none(&traceback);

match vm.normalize_exception(exctype, value.clone(), traceback) {

Ok(exc) => {

let native_syntax_error_display = !has_traceback

&& exc.fast_isinstance(vm.ctx.exceptions.syntax_error)

&& exc

.as_object()

.get_attr("msg", vm)

.ok()

.and_then(|msg| msg.downcast::<PyStr>().ok())

.is_some_and(|msg| msg.to_string_lossy() == "unexpected EOF while parsing")

&& exc

.as_object()

.get_attr("text", vm)

.ok()

.and_then(|text| text.downcast::<PyStr>().ok())

.is_some_and(|text| text.to_string_lossy().trim_end() == "\\");

if native_syntax_error_display {

return vm.write_exception(&mut crate::py_io::PyWriter(stderr, vm), &exc);

}

// PyErr_Display: try traceback._print_exception_bltin first

if let Ok(tb_mod) = vm.import("traceback", 0)

&& let Ok(print_exc_builtin) = tb_mod.get_attr("_print_exception_bltin", vm)

&& print_exc_builtin

.call((exc.as_object().to_owned(),), vm)

.is_ok()

{

return Ok(());

}

// Fallback to Rust-level exception printing

vm.write_exception(&mut crate::py_io::PyWriter(stderr, vm), &exc)

}

Err(_) => {

let type_name = value.class().name();

let msg = format!(

"TypeError: print_exception(): Exception expected for value, {type_name} found\n"

);

use crate::py_io::Write;

write!(&mut crate::py_io::PyWriter(stderr, vm), "{msg}")?;

Ok(())

}

}

}

#[pyfunction(name = "__breakpointhook__")]

#[pyfunction]

pub fn breakpointhook(args: FuncArgs, vm: &VirtualMachine) -> PyResult {

#[cfg(feature = "host_env")]

let env_var = crate::host_env::os::var("PYTHONBREAKPOINT")

.and_then(|env_var| {

if env_var.is_empty() {

Err(std::env::VarError::NotPresent)

} else {

Ok(env_var)

}

})

.unwrap_or_else(|_| "pdb.set_trace".to_owned());

#[cfg(not(feature = "host_env"))]

let env_var = "pdb.set_trace".to_owned();

if env_var.eq("0") {

return Ok(vm.ctx.none());

};

let print_unimportable_module_warn = || {

warn(

vm.ctx.exceptions.runtime_warning,

format!("Ignoring unimportable $PYTHONBREAKPOINT: \"{env_var}\"",),

0,

vm,

)?;

Ok(vm.ctx.none())

};

let last = match env_var.rsplit_once('.') {

Some((_, last)) => last,

None if !env_var.is_empty() => env_var.as_str(),

_ => return print_unimportable_module_warn(),

};

let (module_path, attr_name) = if last == env_var {

("builtins", env_var.as_str())

} else {

(&env_var[..(env_var.len() - last.len() - 1)], last)