use rustpython_parser::ast;
use crate::{
error::{EvalError, EvalResult, InterpreterError},
eval::{eval_expr, functions::resolve_proxy},
security::validator,
state::InterpreterState,
tools::Tools,
value::{ExceptionValue, Value, shared_list},
};
pub fn eval_name(state: &InterpreterState, node: &ast::ExprName, tools: &Tools) -> EvalResult {
let name = node.id.as_str();
validator::validate_name(validator::NameContext::Access, name)?;
if let Some(val) = state.get_variable(name) {
return Ok(val.clone());
}
if tools.contains_key(name) {
return Ok(Value::ToolName(name.to_string()));
}
if name == "NotImplemented" {
return Ok(Value::NotImplemented);
}
if name == "Ellipsis" {
return Ok(Value::Ellipsis);
}
let builtin_functions = [
"print",
"len",
"range",
"str",
"int",
"float",
"complex",
"bool",
"type",
"isinstance",
"issubclass",
"super",
"hasattr",
"getattr",
"setattr",
"delattr",
"vars",
"callable",
"abs",
"round",
"min",
"max",
"sum",
"all",
"any",
"sorted",
"enumerate",
"zip",
"reversed",
"chr",
"ord",
"list",
"tuple",
"dict",
"set",
"frozenset",
"iter",
"next",
"filter",
"map",
"repr",
"ascii",
"slice",
"memoryview",
"hash",
"id",
"input",
"pow",
"divmod",
"format",
"object",
"bin",
"oct",
"hex",
"bytes",
"bytearray",
"__import__",
"dir",
"property",
];
if builtin_functions.contains(&name) {
return Ok(Value::BuiltinName(name.to_string()));
}
if crate::eval::functions::is_exception_type_name(name) {
return Ok(Value::ExceptionType(name.to_string()));
}
if crate::eval::modules::is_auto_imported(name) {
return Ok(Value::Module(name.to_string()));
}
Err(InterpreterError::name_not_defined(name).into())
}
pub async fn eval_named_expr(
state: &mut InterpreterState,
node: &ast::ExprNamedExpr,
tools: &Tools,
) -> EvalResult {
state.enter_expr().map_err(EvalError::from)?;
let out = eval_named_expr_inner(state, node, tools).await;
state.exit_expr();
out
}
async fn eval_named_expr_inner(
state: &mut InterpreterState,
node: &ast::ExprNamedExpr,
tools: &Tools,
) -> EvalResult {
let value = eval_expr(state, &node.value, tools).await?;
match node.target.as_ref() {
ast::Expr::Name(name_node) => {
let name = name_node.id.as_str();
validator::validate_name(validator::NameContext::Assignment, name)?;
state.set_variable(name, value.clone()).map_err(EvalError::Interpreter)?;
Ok(value)
}
other => Err(InterpreterError::Runtime(format!(
"walrus assignment target must be a name, not {:?}",
std::mem::discriminant(other)
))
.into()),
}
}
pub(crate) fn resolve_object_attr(obj: &Value, name: &str) -> Option<Value> {
match name {
"__class__" => Some(crate::eval::functions::type_object_of(obj)),
_ => None,
}
}
pub(crate) fn is_object_attr(name: &str) -> bool {
matches!(name, "__class__")
}
pub async fn eval_attribute(
state: &mut InterpreterState,
node: &ast::ExprAttribute,
tools: &Tools,
) -> EvalResult {
state.enter_expr().map_err(EvalError::from)?;
let out = eval_attribute_inner(state, node, tools).await;
state.exit_expr();
out
}
async fn eval_attribute_inner(
state: &mut InterpreterState,
node: &ast::ExprAttribute,
tools: &Tools,
) -> EvalResult {
let attr_name = node.attr.as_str();
if !is_object_attr(attr_name) {
validator::validate_attribute(attr_name)?;
}
let place_opt = crate::eval::place::eval_place(state, &node.value, tools).await?;
let usable_place: Option<&crate::eval::place::Place> = match &place_opt {
Some(p) if p.is_navigable() && state.variables.contains_key(&p.root) => Some(p),
_ => None,
};
let nav_value: Option<Value> = match usable_place {
Some(place) => {
let mut root = state
.get_variable(&place.root)
.cloned()
.ok_or_else(|| EvalError::from(InterpreterError::name_not_defined(&place.root)))?;
crate::eval::place::with_navigate_mut(&mut root, &place.steps, |target| target.clone())
.ok()
}
None => None,
};
let (obj, place_for_upgrade) = match nav_value {
Some(v) => (v, usable_place),
None => (eval_expr(state, &node.value, tools).await?, None),
};
let obj = resolve_proxy(&obj).await?;
if let Value::Super { defining_class, instance } = &obj {
return crate::eval::classes::super_attribute(
state,
defining_class,
instance,
attr_name,
tools,
)
.await;
}
getattr_on_value(state, obj, attr_name, tools, place_for_upgrade).await
}
pub(crate) async fn getattr_on_value(
state: &mut InterpreterState,
obj: Value,
attr_name: &str,
tools: &Tools,
place_for_upgrade: Option<&crate::eval::place::Place>,
) -> EvalResult {
if let Some(resolved) = resolve_object_attr(&obj, attr_name) {
return Ok(resolved);
}
if let Value::Instance(inst) = &obj {
if let Some((_, method)) =
crate::eval::classes::lookup_method_in_mro(state, &inst.class_name, "__getattribute__")
{
validator::validate_attribute(attr_name)?;
let attr_arg = Value::String(attr_name.into());
let call = crate::eval::functions::CallArgs {
positional: std::slice::from_ref(&attr_arg),
keyword: &indexmap::IndexMap::new(),
};
let (returned, _self) =
crate::eval::classes::call_method(state, &method, obj.clone(), call, tools).await?;
return Ok(returned);
}
}
getattr_normal_lookup(state, obj, attr_name, tools, place_for_upgrade).await
}
pub(crate) async fn getattr_normal_lookup(
state: &mut InterpreterState,
obj: Value,
attr_name: &str,
tools: &Tools,
place_for_upgrade: Option<&crate::eval::place::Place>,
) -> EvalResult {
validator::validate_attribute(attr_name)?;
if let Value::Instance(inst) = &obj {
if let Some(prop) =
crate::eval::classes::lookup_property(state, &inst.class_name, attr_name)
{
let cache_key = prop.cached.then_some(attr_name);
return crate::eval::classes::invoke_property_getter(
state,
&prop.getter,
obj.clone(),
cache_key,
tools,
)
.await;
}
if let Some(desc) =
crate::eval::classes::lookup_class_attr_instance(state, &inst.class_name, attr_name)
{
let has_get =
crate::eval::classes::lookup_method_in_mro(state, &desc.class_name, "__get__")
.is_some();
if has_get {
let is_data =
crate::eval::classes::lookup_method_in_mro(state, &desc.class_name, "__set__")
.is_some()
|| crate::eval::classes::lookup_method_in_mro(
state,
&desc.class_name,
"__delete__",
)
.is_some();
if !is_data {
if let Some(v) = inst.fields.lock().get(attr_name) {
return Ok(v.clone());
}
}
if let Some((_, get_method)) =
crate::eval::classes::lookup_method_in_mro(state, &desc.class_name, "__get__")
{
let owner = Value::Class(inst.class_name.clone());
let call = crate::eval::functions::CallArgs {
positional: &[obj.clone(), owner],
keyword: &indexmap::IndexMap::new(),
};
let (returned, _) = crate::eval::classes::call_method(
state,
&get_method,
Value::Instance(desc),
call,
tools,
)
.await?;
return Ok(returned);
}
}
}
}
if let Value::BuiltinName(type_name) = &obj {
if attr_name == "__name__" || attr_name == "__qualname__" {
return Ok(Value::String(type_name.as_str().into()));
}
if crate::types::builtin_type_attr_present(type_name, attr_name) {
return Ok(Value::BuiltinTypeMethod {
type_name: type_name.clone(),
method: attr_name.to_string(),
});
}
return Err(InterpreterError::AttributeError(format!(
"type object '{type_name}' has no attribute '{attr_name}'"
))
.into());
}
if let Some(val) = crate::types::dispatch_getattr_opt(&obj, attr_name)? {
return Ok(upgrade_bound_method_place(val, place_for_upgrade));
}
if let Value::Instance(inst) = &obj {
let has_field = inst.fields.lock().get(attr_name).is_some();
if !has_field
&& crate::eval::classes::lookup_method_in_mro(state, &inst.class_name, attr_name)
.is_some()
{
return Ok(Value::BoundMethod {
receiver: crate::value::BoundMethodReceiver::Snapshot(Box::new(obj.clone())),
method: attr_name.to_string(),
});
}
}
if matches!(obj, Value::Generator { .. } | Value::Lazy { .. } | Value::BuiltinIter { .. })
&& crate::eval::functions::is_generator_method(attr_name)
{
return Ok(Value::BoundMethod {
receiver: crate::value::BoundMethodReceiver::Snapshot(Box::new(obj.clone())),
method: attr_name.to_string(),
});
}
match legacy_attribute(state, &obj, attr_name) {
Ok(v) => Ok(v),
Err(err) => {
let is_attribute_error =
matches!(err, EvalError::Interpreter(InterpreterError::AttributeError(_)));
if !is_attribute_error {
return Err(err);
}
if let Value::Instance(inst) = &obj {
if let Some((_, method)) = crate::eval::classes::lookup_method_in_mro(
state,
&inst.class_name,
"__getattr__",
) {
let attr_arg = Value::String(attr_name.into());
let call = crate::eval::functions::CallArgs {
positional: std::slice::from_ref(&attr_arg),
keyword: &indexmap::IndexMap::new(),
};
let (returned, _self) =
crate::eval::classes::call_method(state, &method, obj.clone(), call, tools)
.await?;
return Ok(returned);
}
}
Err(err)
}
}
}
fn upgrade_bound_method_place(
val: Value,
place_for_upgrade: Option<&crate::eval::place::Place>,
) -> Value {
let (
Value::BoundMethod { receiver: crate::value::BoundMethodReceiver::Snapshot(_), method },
Some(place),
) = (&val, place_for_upgrade)
else {
return val;
};
let bm_steps: Vec<crate::value::BoundMethodStep> = place
.steps
.iter()
.filter_map(|s| match s {
crate::eval::place::PlaceStep::Index(v) => {
Some(crate::value::BoundMethodStep::Index(v.clone()))
}
crate::eval::place::PlaceStep::Attr(n) => {
Some(crate::value::BoundMethodStep::Attr(n.clone()))
}
crate::eval::place::PlaceStep::Slice(_) => None,
})
.collect();
Value::BoundMethod {
receiver: crate::value::BoundMethodReceiver::Place {
root: place.root.clone(),
steps: bm_steps,
},
method: method.clone(),
}
}
fn legacy_attribute(state: &InterpreterState, obj: &Value, attr_name: &str) -> EvalResult {
match obj {
Value::Exception(exc) => exception_attribute(exc, attr_name),
Value::Array { typecode, .. } => match attr_name {
"typecode" => Ok(Value::String(typecode.to_string().into())),
"itemsize" => {
Ok(Value::Int(crate::eval::modules::array_mod::itemsize(*typecode) as i64))
}
_ => Err(attribute_error("array.array", attr_name)),
},
Value::Instance(inst) => crate::eval::classes::instance_attribute(state, inst, attr_name),
Value::Class(class_name) => {
crate::eval::classes::class_attribute(state, class_name, attr_name)
}
Value::Property { class_name, name } => {
let prop = state.classes.get(class_name).and_then(|c| c.properties.get(name));
let Some(prop) = prop else {
return Err(attribute_error("property", attr_name));
};
let as_func =
|fd: &crate::value::FunctionDef| Value::Function(std::sync::Arc::new(fd.clone()));
match attr_name {
"fget" => Ok(as_func(&prop.getter)),
"fset" => Ok(prop.setter.as_ref().map_or(Value::None, as_func)),
"fdel" => Ok(prop.deleter.as_ref().map_or(Value::None, as_func)),
"__doc__" => Ok(prop
.getter
.docstring
.clone()
.map_or(Value::None, |d| Value::String(d.into()))),
"__isabstractmethod__" => Ok(Value::Bool(false)),
_ => Err(attribute_error("property", attr_name)),
}
}
Value::Type(type_name) | Value::ExceptionType(type_name) => {
if attr_name == "__name__" || attr_name == "__qualname__" {
Ok(Value::String(Value::short_type_name(type_name).to_string().into()))
} else {
Err(attribute_error("type", attr_name))
}
}
Value::Function(func_def) => {
if let Some(v) =
state.function_attrs.get(func_def.body_cache_key()).and_then(|m| m.get(attr_name))
{
return Ok(v.clone());
}
if attr_name == "__qualname__" {
let reported = func_def
.wraps_name
.clone()
.unwrap_or_else(|| func_def.display_qualname().to_string());
Ok(Value::String(reported.into()))
} else if attr_name == "__name__" {
let reported = func_def.wraps_name.clone().unwrap_or_else(|| {
let full = func_def.display_qualname();
full.rsplit('.').next().unwrap_or(full).to_string()
});
Ok(Value::String(reported.into()))
} else if attr_name == "__doc__" {
Ok(func_def.docstring.clone().map_or(Value::None, |d| Value::String(d.into())))
} else if attr_name == "__call__" {
Ok(Value::Function(func_def.clone()))
} else if attr_name == "__annotations__" {
let mut map: indexmap::IndexMap<crate::value::ValueKey, Value> =
indexmap::IndexMap::new();
for (k, v) in &func_def.annotations {
map.insert(
crate::eval::literals::value_to_key(&Value::String(k.as_str().into()))?,
v.clone(),
);
}
Ok(Value::Dict(crate::value::shared_dict(map)))
} else if attr_name == "__defaults__" {
let dv = &func_def.params.default_values;
if dv.is_empty() { Ok(Value::None) } else { Ok(Value::Tuple(dv.clone())) }
} else if attr_name == "__kwdefaults__" {
let mut map: indexmap::IndexMap<crate::value::ValueKey, Value> =
indexmap::IndexMap::new();
for (i, p) in func_def.params.kwonlyargs.iter().enumerate() {
if let Some(Some(v)) = func_def.params.kw_default_values.get(i) {
map.insert(
crate::eval::literals::value_to_key(&Value::String(
p.name.as_str().into(),
))?,
v.clone(),
);
}
}
if map.is_empty() {
Ok(Value::None)
} else {
Ok(Value::Dict(crate::value::shared_dict(map)))
}
} else {
Err(attribute_error("function", attr_name))
}
}
Value::Lambda(lambda_def) => {
if attr_name == "__name__" {
Ok(Value::String("<lambda>".into()))
} else if attr_name == "__qualname__" {
let reported = if lambda_def.qualname.is_empty() {
"<lambda>"
} else {
lambda_def.qualname.as_str()
};
Ok(Value::String(reported.into()))
} else if attr_name == "__call__" {
Ok(Value::Lambda(lambda_def.clone()))
} else if attr_name == "__doc__" {
Ok(Value::None)
} else {
Err(attribute_error("function", attr_name))
}
}
Value::BoundMethod { receiver, method } => match attr_name {
"__name__" => Ok(Value::String(method.clone().into())),
"__self__" | "__qualname__" => {
let self_value = match receiver {
crate::value::BoundMethodReceiver::Snapshot(v) => (**v).clone(),
crate::value::BoundMethodReceiver::Place { root, steps } => {
let mut root_clone = state
.variables
.get(root)
.ok_or_else(|| {
EvalError::from(InterpreterError::name_not_defined(root))
})?
.clone();
let pl_steps: Vec<crate::eval::place::PlaceStep> = steps
.iter()
.map(|s| match s {
crate::value::BoundMethodStep::Index(v) => {
crate::eval::place::PlaceStep::Index(v.clone())
}
crate::value::BoundMethodStep::Attr(n) => {
crate::eval::place::PlaceStep::Attr(n.clone())
}
})
.collect();
crate::eval::place::with_navigate_mut(&mut root_clone, &pl_steps, |t| {
t.clone()
})?
}
};
if attr_name == "__self__" {
Ok(self_value)
} else {
Ok(Value::String(format!("{}.{method}", self_value.python_type_name()).into()))
}
}
_ => Err(attribute_error("builtin_function_or_method", attr_name)),
},
Value::LruCache(data) => match attr_name {
"__name__" | "__qualname__" | "__doc__" => {
legacy_attribute(state, &data.func, attr_name)
}
"__wrapped__" => Ok(data.func.clone()),
_ => Err(attribute_error("functools._lru_cache_wrapper", attr_name)),
},
Value::SingleDispatch(sd) => match attr_name {
"register" => Ok(Value::Partial(Box::new(crate::value::PartialData {
func: Value::ModuleFunction {
module: "functools".into(),
name: "_sd_register".into(),
},
args: vec![obj.clone()],
keywords: indexmap::IndexMap::new(),
}))),
"__name__" | "__qualname__" => Ok(Value::String(sd.name.clone().into())),
"__doc__" => legacy_attribute(state, &sd.default, attr_name),
"__wrapped__" => Ok(sd.default.clone()),
_ => Err(attribute_error("function", attr_name)),
},
Value::BuiltinTypeMethod { type_name, method } => match attr_name {
"__name__" => Ok(Value::String(method.clone().into())),
"__qualname__" => Ok(Value::String(format!("{type_name}.{method}").into())),
_ => Err(attribute_error("method_descriptor", attr_name)),
},
Value::Module(module) => crate::eval::modules::module_member(module, attr_name),
Value::Slice(slice) => match attr_name {
"start" => Ok(slice.start.clone()),
"stop" => Ok(slice.stop.clone()),
"step" => Ok(slice.step.clone()),
_ => Err(attribute_error("slice", attr_name)),
},
Value::RePattern(pattern) => {
if attr_name == "pattern" {
Ok(Value::String((**pattern).clone().into()))
} else {
Err(attribute_error("re.Pattern", attr_name))
}
}
Value::Template(t) => {
if attr_name == "template" {
Ok(Value::String(t.clone()))
} else {
Err(attribute_error("string.Template", attr_name))
}
}
Value::ModuleFunction { module, name } => {
if let Some(value) = crate::eval::modules::type_attribute(module, name, attr_name) {
Ok(value)
} else if let Some(func) =
crate::eval::modules::type_classmethod(module, name, attr_name)
{
Ok(Value::ModuleFunction { module: module.clone(), name: func.into() })
} else {
Err(attribute_error(obj.type_name(), attr_name))
}
}
_ => Err(attribute_error(obj.type_name(), attr_name)),
}
}
fn attribute_error(type_name: &str, attr_name: &str) -> EvalError {
InterpreterError::AttributeError(format!("'{type_name}' object has no attribute '{attr_name}'"))
.into()
}
fn exception_attribute(exc: &ExceptionValue, attr_name: &str) -> EvalResult {
match attr_name {
"exceptions" => {
let items = exc
.exceptions
.as_ref()
.map(|xs| xs.iter().cloned().map(|e| Value::Exception(Box::new(e))).collect())
.unwrap_or_default();
Ok(Value::Tuple(items))
}
"subgroup" | "split" | "with_traceback" => Ok(Value::ExceptionMethod {
method: attr_name.to_string(),
exception: Box::new(exc.clone()),
}),
"__traceback__" => Ok(Value::None),
"args" => Ok(Value::Tuple(exc.args.clone())),
"value" if exc.type_name == "StopIteration" || exc.type_name == "StopAsyncIteration" => {
Ok(exc.args.first().cloned().unwrap_or(Value::None))
}
"code" if exc.type_name == "SystemExit" && !exc.fields.contains_key("code") => {
Ok(match exc.args.len() {
0 => Value::None,
1 => exc.args[0].clone(),
_ => Value::Tuple(exc.args.clone()),
})
}
"__cause__" => {
Ok(exc.cause.as_ref().map_or(Value::None, |cause| Value::Exception(cause.clone())))
}
"__context__" => Ok(exc.fields.get("__context__").cloned().unwrap_or(Value::None)),
"__suppress_context__" => {
Ok(exc.fields.get("__suppress_context__").cloned().unwrap_or(Value::Bool(false)))
}
"errno" | "strerror" | "filename" | "filename2"
if !exc.fields.contains_key(attr_name)
&& crate::eval::exceptions::builtin_exception_issubclass(
&exc.type_name,
"OSError",
) =>
{
let n = exc.args.len();
let arg = |i: usize| exc.args.get(i).cloned().unwrap_or(Value::None);
let two_to_five = (2..=5).contains(&n);
Ok(match attr_name {
"errno" if two_to_five => arg(0),
"strerror" if two_to_five => arg(1),
"filename" if n >= 3 => arg(2),
"filename2" if n >= 5 => arg(4),
_ => Value::None,
})
}
_ => exc
.fields
.get(attr_name)
.cloned()
.map_or_else(|| Err(attribute_error(&exc.type_name, attr_name)), Ok),
}
}
pub async fn eval_subscript(
state: &mut InterpreterState,
node: &ast::ExprSubscript,
tools: &Tools,
) -> EvalResult {
state.enter_expr().map_err(EvalError::from)?;
let out = eval_subscript_inner(state, node, tools).await;
state.exit_expr();
out
}
async fn eval_subscript_inner(
state: &mut InterpreterState,
node: &ast::ExprSubscript,
tools: &Tools,
) -> EvalResult {
if let ast::Expr::Name(name_node) = node.value.as_ref() {
if !matches!(node.slice.as_ref(), ast::Expr::Slice(_)) {
let container_name = name_node.id.as_str();
let slice_is_static = match node.slice.as_ref() {
ast::Expr::Constant(_) => true,
ast::Expr::Name(slice_name) => slice_name.id.as_str() != container_name,
_ => false,
};
if slice_is_static {
let index = match crate::eval::try_eval_expr_sync(state, &node.slice, tools) {
Some(r) => r?,
None => eval_expr(state, &node.slice, tools).await?,
};
if let Some(container) = state.variables.get(container_name) {
let take_fast_path = matches!(
container,
Value::Dict(_)
| Value::List(_)
| Value::Tuple(_)
| Value::String(_)
| Value::Range { .. }
| Value::Bytes(_)
);
if take_fast_path && !matches!(index, Value::Instance(_) | Value::Slice(_)) {
return crate::types::dispatch_getitem(container, &index);
}
}
}
}
}
let obj = eval_expr(state, &node.value, tools).await?;
let obj = resolve_proxy(&obj).await?;
if let ast::Expr::Slice(slice_node) = node.slice.as_ref() {
return eval_subscript_slice(state, &obj, slice_node, tools).await;
}
let index = eval_expr(state, &node.slice, tools).await?;
if let Value::Slice(slice) = &index {
if matches!(obj, Value::Instance(_)) {
return crate::eval::op::getitem(state, &obj, &index, tools).await;
}
return apply_value_slice(&obj, Some(&slice.start), Some(&slice.stop), Some(&slice.step));
}
if let Value::Instance(inst) = &obj {
if let Some(Value::Tuple(field_names)) =
state.classes.get(&inst.class_name).and_then(|c| c.class_attrs.get("_fields"))
{
if let Value::Int(i) = &index {
let len = field_names.len();
let idx = if *i < 0 {
usize::try_from(i64::try_from(len).unwrap_or(i64::MAX) + *i).ok()
} else {
usize::try_from(*i).ok()
};
if let Some(idx) = idx.filter(|&n| n < len) {
if let Value::String(field_name) = &field_names[idx] {
return Ok(inst
.fields
.lock()
.get(field_name.as_str())
.cloned()
.unwrap_or(Value::None));
}
}
return Err(InterpreterError::Runtime(format!(
"tuple index out of range: {i} (len {len})"
))
.into());
}
}
}
if let Value::DefaultDict(data) = &obj {
let key = crate::eval::literals::value_to_key(&index)?;
if let Some(value) = data.items.get(&key) {
return Ok(value.clone());
}
let synthesised = invoke_factory(state, &data.factory, tools).await?;
if let ast::Expr::Name(name_node) = node.value.as_ref() {
let name = name_node.id.as_str().to_string();
if let Some(Value::DefaultDict(data)) = state.variables.get(&name).cloned() {
let mut new_data = *data;
new_data.items.insert(key, synthesised.clone());
state
.set_variable(&name, Value::DefaultDict(Box::new(new_data)))
.map_err(EvalError::Interpreter)?;
}
}
return Ok(synthesised);
}
if let Value::Type(name) = &obj {
if name.starts_with("typing.") {
return Ok(Value::Type(format!("{name}[{}]", typing_arg_repr(&index))));
}
}
if let Value::BuiltinName(name) = &obj {
const GENERIC_BUILTINS: &[&str] = &["list", "dict", "tuple", "set", "frozenset", "type"];
if GENERIC_BUILTINS.contains(&name.as_str()) {
return Ok(Value::Type(format!("{name}[{}]", typing_arg_repr(&index))));
}
}
if let Value::Class(class_name) = &obj {
if let Some((_, method)) =
crate::eval::classes::lookup_method_in_mro(state, class_name, "__class_getitem__")
{
let call = crate::eval::functions::CallArgs {
positional: std::slice::from_ref(&index),
keyword: &indexmap::IndexMap::new(),
};
let (returned, _self) =
crate::eval::classes::call_method(state, &method, obj.clone(), call, tools).await?;
return Ok(returned);
}
if state.classes.get(class_name).is_some_and(|c| c.is_generic) {
return Ok(obj.clone());
}
}
crate::eval::op::getitem(state, &obj, &index, tools).await
}
fn typing_arg_repr(v: &Value) -> String {
match v {
Value::Type(n) | Value::Class(n) | Value::BuiltinName(n) | Value::ExceptionType(n) => {
n.clone()
}
Value::Ellipsis => "...".to_string(),
Value::None => "NoneType".to_string(),
Value::Tuple(items) => items.iter().map(typing_arg_repr).collect::<Vec<_>>().join(", "),
Value::List(items) => {
let inner = items.lock().iter().map(typing_arg_repr).collect::<Vec<_>>().join(", ");
format!("[{inner}]")
}
other => other.repr(),
}
}
pub async fn invoke_factory_pub(
state: &mut InterpreterState,
factory: &Value,
tools: &Tools,
) -> EvalResult {
invoke_factory(state, factory, tools).await
}
async fn invoke_factory(
state: &mut InterpreterState,
factory: &Value,
tools: &Tools,
) -> EvalResult {
let kwargs: indexmap::IndexMap<String, Value> = indexmap::IndexMap::new();
let empty: [Value; 0] = [];
match factory {
Value::Function(def) => {
crate::eval::functions::call_user_function(state, def, &empty, &kwargs, tools).await
}
Value::Lambda(def) => {
crate::eval::functions::call_lambda(state, def, &empty, &kwargs, tools).await
}
Value::Class(name) => {
crate::eval::classes::instantiate(state, name, &empty, &kwargs, tools).await
}
Value::None => Ok(Value::None),
Value::BuiltinName(builtin) => {
match builtin.as_str() {
"int" => Ok(Value::Int(0)),
"float" => Ok(Value::Float(0.0)),
"bool" => Ok(Value::Bool(false)),
"str" => Ok(Value::String("".into())),
"bytes" => Ok(Value::Bytes(Vec::new())),
"list" => Ok(Value::List(shared_list(Vec::new()))),
"tuple" => Ok(Value::Tuple(Vec::new())),
"dict" => Ok(Value::Dict(crate::value::shared_dict(indexmap::IndexMap::new()))),
"set" => Ok(Value::new_set(Vec::new())),
"frozenset" => Ok(Value::new_frozenset(Vec::new())),
_ => Err(InterpreterError::TypeError(format!(
"defaultdict factory builtin '{builtin}' is not zero-arg constructable"
))
.into()),
}
}
other => Err(InterpreterError::TypeError(format!(
"defaultdict factory must be callable (got '{}')",
other.type_name()
))
.into()),
}
}
async fn eval_subscript_slice(
state: &mut InterpreterState,
obj: &Value,
slice_node: &ast::ExprSlice,
tools: &Tools,
) -> EvalResult {
let lower = match &slice_node.lower {
Some(expr) => {
let v = eval_expr(state, expr, tools).await?;
Some(crate::eval::op::coerce_index(state, v, tools).await?)
}
None => None,
};
let upper = match &slice_node.upper {
Some(expr) => {
let v = eval_expr(state, expr, tools).await?;
Some(crate::eval::op::coerce_index(state, v, tools).await?)
}
None => None,
};
let step_expr = match &slice_node.step {
Some(expr) => {
let v = eval_expr(state, expr, tools).await?;
Some(crate::eval::op::coerce_index(state, v, tools).await?)
}
None => None,
};
if matches!(obj, Value::Instance(_)) {
let slice_val = Value::Slice(Box::new(crate::value::SliceValue {
start: lower.unwrap_or(Value::None),
stop: upper.unwrap_or(Value::None),
step: step_expr.unwrap_or(Value::None),
}));
return crate::eval::op::getitem(state, obj, &slice_val, tools).await;
}
apply_value_slice(obj, lower.as_ref(), upper.as_ref(), step_expr.as_ref())
}
pub(crate) fn apply_value_slice(
obj: &Value,
lower: Option<&Value>,
upper: Option<&Value>,
step_expr: Option<&Value>,
) -> EvalResult {
let lower = lower.filter(|v| !matches!(v, Value::None));
let upper = upper.filter(|v| !matches!(v, Value::None));
let stride = match step_expr {
Some(Value::Int(s)) => *s,
Some(Value::Bool(b)) => i64::from(*b),
None | Some(Value::None) => 1,
Some(_) => {
return Err(InterpreterError::TypeError(
"slice indices must be integers or None or have an __index__ method".to_string(),
)
.into());
}
};
if stride == 0 {
return Err(InterpreterError::ValueError("slice step cannot be zero".into()).into());
}
match obj {
Value::List(items) => {
let snapshot = items.lock().clone();
let sliced = slice_sequence(&snapshot, lower, upper, stride)?;
Ok(Value::List(shared_list(sliced)))
}
Value::Array { typecode, items } => {
let snapshot = items.lock().clone();
let sliced = slice_sequence(&snapshot, lower, upper, stride)?;
Ok(Value::Array { typecode: *typecode, items: shared_list(sliced) })
}
Value::Tuple(items) => {
let sliced = slice_sequence(items, lower, upper, stride)?;
Ok(Value::Tuple(sliced))
}
Value::String(s) => {
let chars: Vec<Value> =
s.chars().map(|c| Value::String(c.to_string().into())).collect();
let sliced = slice_sequence(&chars, lower, upper, stride)?;
let result: String = sliced
.into_iter()
.map(|v| match v {
Value::String(s) => s.into(),
_ => String::new(),
})
.collect();
Ok(Value::String(result.into()))
}
Value::Bytes(_) | Value::ByteArray(_) | Value::MemoryView(_) => {
let raw = crate::types::memoryview_bytes(obj);
let elems: Vec<Value> = raw.iter().map(|&n| Value::Int(i64::from(n))).collect();
let sliced = slice_sequence(&elems, lower, upper, stride)?;
let bytes: Vec<u8> = sliced
.into_iter()
.filter_map(|v| match v {
Value::Int(n) => u8::try_from(n & 0xFF).ok(),
_ => None,
})
.collect();
match obj {
Value::ByteArray(_) => Ok(Value::ByteArray(crate::value::shared_bytes(bytes))),
Value::MemoryView(_) => Ok(Value::MemoryView(Box::new(Value::Bytes(bytes)))),
_ => Ok(Value::Bytes(bytes)),
}
}
Value::Range { start, stop, step } => {
let len =
i64::try_from(crate::types::range_length(*start, *stop, *step)).map_err(|_| {
EvalError::from(InterpreterError::Runtime("range length overflow".into()))
})?;
let resolve = |v: Option<&Value>, default: i64| -> Result<i64, EvalError> {
match v {
None | Some(Value::None) => Ok(default),
Some(Value::Int(i)) => Ok(*i),
Some(Value::Bool(b)) => Ok(i64::from(*b)),
Some(_) => Err(InterpreterError::TypeError(
"slice indices must be integers or None or have an __index__ method"
.to_string(),
)
.into()),
}
};
let (begin, end) = if stride > 0 {
(
clamp_slice_index(resolve(lower, 0)?, len),
clamp_slice_index(resolve(upper, len)?, len),
)
} else {
(
clamp_slice_index_neg(resolve(lower, len - 1)?, len),
clamp_slice_index_neg(resolve(upper, -(len + 1))?, len),
)
};
Ok(Value::Range {
start: start + begin * step,
stop: start + end * step,
step: step * stride,
})
}
_ => Err(InterpreterError::TypeError(format!(
"'{}' object is not subscriptable",
obj.type_name()
))
.into()),
}
}
fn slice_sequence(
items: &[Value],
lower: Option<&Value>,
upper: Option<&Value>,
stride: i64,
) -> Result<Vec<Value>, EvalError> {
let len = i64::try_from(items.len()).map_err(|_| {
InterpreterError::Runtime("sequence length overflows i64 for slicing".into())
})?;
let resolve_index = |val: Option<&Value>, default: i64| -> Result<i64, EvalError> {
match val {
None | Some(Value::None) => Ok(default),
Some(Value::Int(i)) => Ok(*i),
Some(Value::Bool(b)) => Ok(i64::from(*b)),
Some(_) => Err(InterpreterError::TypeError(
"slice indices must be integers or None or have an __index__ method".to_string(),
)
.into()),
}
};
let to_index = |i: i64| -> Result<usize, EvalError> {
usize::try_from(i).map_err(|_| {
InterpreterError::Runtime("slice index overflow (internal invariant)".into()).into()
})
};
if stride > 0 {
let raw_start = resolve_index(lower, 0)?;
let raw_stop = resolve_index(upper, len)?;
let begin = clamp_slice_index(raw_start, len);
let end = clamp_slice_index(raw_stop, len);
let mut result = Vec::new();
let mut i = begin;
while i < end {
result.push(items[to_index(i)?].clone());
i += stride;
}
Ok(result)
} else {
let raw_start = resolve_index(lower, len - 1)?;
let raw_stop = resolve_index(upper, -(len + 1))?;
let begin = clamp_slice_index_neg(raw_start, len);
let end = clamp_slice_index_neg(raw_stop, len);
let mut result = Vec::new();
let mut i = begin;
while i > end {
result.push(items[to_index(i)?].clone());
i += stride;
}
Ok(result)
}
}
pub(crate) fn clamp_slice_index(idx: i64, len: i64) -> i64 {
let adjusted = if idx < 0 { idx + len } else { idx };
adjusted.max(0).min(len)
}
pub(crate) fn clamp_slice_index_neg(idx: i64, len: i64) -> i64 {
let adjusted = if idx < 0 { idx + len } else { idx };
adjusted.max(-1).min(len - 1)
}
pub async fn eval_slice(
state: &mut InterpreterState,
node: &ast::ExprSlice,
tools: &Tools,
) -> EvalResult {
let lower = if let Some(ref expr) = node.lower {
eval_expr(state, expr, tools).await?
} else {
Value::None
};
let upper = if let Some(ref expr) = node.upper {
eval_expr(state, expr, tools).await?
} else {
Value::None
};
let stride = if let Some(ref expr) = node.step {
eval_expr(state, expr, tools).await?
} else {
Value::None
};
Ok(Value::Tuple(vec![lower, upper, stride]))
}