use std::{collections::BTreeMap, sync::Arc};
use indexmap::IndexMap;
use rustpython_parser::ast::{self, Expr, Stmt};
use crate::{
error::{EvalError, EvalResult, InterpreterError},
eval::{
eval_expr,
functions::{
CallArgs, bind_params_named, build_function_params, call_lambda, call_user_function,
execute_body, extract_function_source,
},
},
state::InterpreterState,
tools::Tools,
value::{ClassValue, FunctionDef, InstanceValue, PropertyDef, Value, shared_list},
};
pub async fn eval_class_def(
state: &mut InterpreterState,
node: &ast::StmtClassDef,
tools: &Tools,
) -> EvalResult {
let class_name = node.name.as_str();
crate::security::validator::validate_name(
crate::security::validator::NameContext::Assignment,
class_name,
)?;
let mut metaclass_name: Option<String> = None;
let mut init_subclass_kwargs: IndexMap<String, Value> = IndexMap::new();
for kw in &node.keywords {
let key = kw.arg.as_ref().map(|a| a.as_str());
let val = eval_expr(state, &kw.value, tools).await?;
if key == Some("metaclass") {
match val {
Value::BuiltinName(n) | Value::Type(n) if n == "type" => {}
Value::Type(n) if n == "abc.ABCMeta" || n == "abc.ABC" => {}
Value::Class(n) => metaclass_name = Some(n),
other => {
return Err(InterpreterError::TypeError(format!(
"metaclass must be a type, not '{}'",
other.type_name()
))
.into());
}
}
} else if let Some(name) = key {
init_subclass_kwargs.insert(name.to_string(), val);
}
}
let mut bases: Vec<String> = Vec::new();
let mut enum_kind: Option<crate::value::EnumKind> = None;
let mut is_typing_namedtuple = false;
let mut is_generic = false;
for base in &node.bases {
let (base_name, resolved): (String, Option<Value>) = match base {
Expr::Name(name_node) => {
let base_name = name_node.id.as_str().to_string();
let resolved = state.variables.get(&base_name).cloned();
(base_name, resolved)
}
other => {
let val = eval_expr(state, other, tools).await?;
match &val {
Value::Class(n) | Value::ExceptionType(n) => (n.clone(), Some(val)),
Value::Type(n) => (n.clone(), Some(val)),
other_v => {
return Err(InterpreterError::TypeError(format!(
"bases must be types, not '{}'",
other_v.type_name()
))
.into());
}
}
}
};
if base_name == "object" {
continue;
}
if let Some(Value::Type(type_name)) = &resolved {
match type_name.as_str() {
"enum.Enum" => enum_kind = Some(crate::value::EnumKind::Plain),
"enum.IntEnum" => enum_kind = Some(crate::value::EnumKind::Int),
"enum.Flag" => enum_kind = Some(crate::value::EnumKind::Flag),
"enum.IntFlag" => enum_kind = Some(crate::value::EnumKind::IntFlag),
"enum.StrEnum" => enum_kind = Some(crate::value::EnumKind::Str),
"typing.NamedTuple" => is_typing_namedtuple = true,
"abc.ABC" | "abc.ABCMeta" => {}
n if n == "typing.Generic" || n.starts_with("typing.Generic[") => {
is_generic = true;
}
_ => {}
}
continue;
}
if matches!(resolved, Some(Value::ModuleFunction { .. })) {
continue;
}
let is_exception_base = crate::eval::functions::is_exception_type_name(&base_name)
|| matches!(resolved, Some(Value::ExceptionType(_)));
let is_type_base =
base_name == "type" && resolved.is_none() && !state.classes.contains_key(&base_name);
if !state.classes.contains_key(&base_name) && !is_exception_base && !is_type_base {
return Err(InterpreterError::name_not_defined(&base_name).into());
}
bases.push(base_name);
}
if metaclass_name.is_none() {
for base in &bases {
if let Some(mc) = state.classes.get(base).and_then(|c| c.metaclass.clone()) {
metaclass_name = Some(mc);
break;
}
}
}
let mut methods: BTreeMap<String, FunctionDef> = BTreeMap::new();
let mut class_attrs: BTreeMap<String, Value> = BTreeMap::new();
let mut properties: BTreeMap<String, PropertyDef> = BTreeMap::new();
let mut static_methods: BTreeMap<String, FunctionDef> = BTreeMap::new();
let mut class_methods: BTreeMap<String, FunctionDef> = BTreeMap::new();
let mut annotations: Vec<String> = Vec::new();
let mut initvar_fields: Vec<String> = Vec::new();
let mut enum_members: Vec<String> = Vec::new();
let mut enum_auto_next: i64 = 1;
let mut abstract_here: Vec<String> = Vec::new();
if let Some(ref meta) = metaclass_name {
if let Some(prepared) =
invoke_metaclass_prepare(state, meta, class_name, &bases, tools).await?
{
for (k, v) in prepared {
if let crate::value::ValueKey::String(s) = k {
class_attrs.insert(s.to_string(), v);
}
}
}
}
let qualname = match state.qualname_stack.last() {
Some(parent) => format!("{parent}.{class_name}"),
None => class_name.to_string(),
};
state.qualname_stack.push(qualname.clone());
let class_scope_saved = state.variables.clone();
let body_result: Result<(), EvalError> = async {
for stmt in &node.body {
for (name, value) in &class_attrs {
state.variables.insert(name.clone(), value.clone());
}
match stmt {
Stmt::FunctionDef(_) | Stmt::AsyncFunctionDef(_) => {
let async_converted;
let (method, method_is_async): (&ast::StmtFunctionDef, bool) = match stmt {
Stmt::FunctionDef(m) => (m, false),
Stmt::AsyncFunctionDef(m) => {
async_converted = ast::StmtFunctionDef {
name: m.name.clone(),
args: m.args.clone(),
body: m.body.clone(),
decorator_list: m.decorator_list.clone(),
returns: m.returns.clone(),
type_comment: m.type_comment.clone(),
type_params: m.type_params.clone(),
range: m.range,
};
(&async_converted, true)
}
_ => unreachable!(),
};
let mut params = build_function_params(&method.args)?;
crate::eval::functions::evaluate_param_defaults(state, &mut params, tools)
.await?;
let source = extract_function_source(&state.current_source, method);
let method_name = method.name.as_str().to_string();
let has_abstract =
method.decorator_list.iter().any(is_abstractmethod_decorator);
if has_abstract {
abstract_here.push(method_name.clone());
}
let decorators: Vec<Expr> = method
.decorator_list
.iter()
.filter(|d| !is_abstractmethod_decorator(d))
.cloned()
.collect();
if decorators.len() == 1
&& !is_builtin_method_decorator(&decorators[0], &properties)
{
let key = format!("{class_name}.{method_name}");
let qualname = state.qualname_for(&method_name);
let func_def = build_plain_method_def(
state,
key,
params,
method.body.clone(),
source,
qualname,
method_is_async,
);
let func_value = Value::Function(Arc::new(func_def));
let decorator_val = eval_expr(state, &decorators[0], tools).await?;
let empty_kwargs = indexmap::IndexMap::new();
let decorated = crate::eval::functions::call_value_as_function(
state,
&decorator_val,
std::slice::from_ref(&func_value),
&empty_kwargs,
tools,
)
.await?;
class_attrs.insert(method_name, decorated);
continue;
}
let patch_name = method_name.clone();
classify_decorated_method(
state,
&decorators,
class_name,
method_name,
params,
method.body.clone(),
source,
&mut methods,
&mut properties,
&mut static_methods,
&mut class_methods,
)?;
if method_is_async {
if let Some(fd) = methods
.get_mut(&patch_name)
.or_else(|| static_methods.get_mut(&patch_name))
.or_else(|| class_methods.get_mut(&patch_name))
{
fd.is_async = true;
}
}
}
Stmt::Assign(assign) => {
let value = eval_expr(state, &assign.value, tools).await?;
for target in &assign.targets {
bind_class_target(
target,
&value,
enum_kind,
class_name,
&mut class_attrs,
&mut enum_auto_next,
)?;
if enum_kind.is_some() {
if let Expr::Name(n) = target {
let member = n.id.as_str();
if matches!(class_attrs.get(member), Some(Value::EnumMember { .. }))
&& !enum_members.iter().any(|m| m == member)
{
enum_members.push(member.to_string());
}
}
}
}
}
Stmt::AnnAssign(ann) => {
if let Expr::Name(target) = ann.target.as_ref() {
let attr_name = target.id.as_str().to_string();
let head = annotation_head_name(ann.annotation.as_ref());
let is_classvar = head == Some("ClassVar");
if head == Some("InitVar") && !initvar_fields.contains(&attr_name) {
initvar_fields.push(attr_name.clone());
}
if !is_classvar && !annotations.contains(&attr_name) {
annotations.push(attr_name.clone());
}
if let Some(value_expr) = &ann.value {
let value = eval_expr(state, value_expr, tools).await?;
let wrapped = wrap_enum_member(
enum_kind,
class_name,
&attr_name,
value,
&mut enum_auto_next,
);
class_attrs.insert(attr_name, wrapped);
}
}
}
Stmt::If(_) | Stmt::For(_) | Stmt::While(_) | Stmt::ClassDef(_) => {
crate::eval::eval_stmt(state, stmt, tools).await?;
}
_ => {}
}
let mut changed: Vec<(String, Value)> = Vec::new();
for (name, value) in &state.variables {
if class_attrs.get(name) != Some(value)
&& class_scope_saved.get(name) != Some(value)
{
changed.push((name.clone(), value.clone()));
}
}
for (name, value) in changed {
let wrapped =
wrap_enum_member(enum_kind, class_name, &name, value, &mut enum_auto_next);
class_attrs.insert(name, wrapped);
}
}
Ok(())
}
.await;
state.variables = class_scope_saved;
state.qualname_stack.pop();
body_result?;
let mro = build_mro(class_name, &bases, &state.classes)?;
let bases_for_hook = bases.clone();
let attrs_for_set_name: Vec<(String, Value)> =
class_attrs.iter().map(|(k, v)| (k.clone(), v.clone())).collect();
let (mut slots, mut slot_names) = parse_slots_attr(class_attrs.get("__slots__"));
for base in &bases {
if let Some(b) = state.classes.get(base) {
if b.slots {
slots = true;
for n in &b.slot_names {
if !slot_names.iter().any(|s| s == n) {
slot_names.push(n.clone());
}
}
}
}
}
let abstract_methods: Vec<String> = {
let mut unresolved: std::collections::BTreeSet<String> =
abstract_here.iter().cloned().collect();
for base in &bases {
if let Some(bc) = state.classes.get(base) {
unresolved.extend(bc.abstract_methods.iter().cloned());
}
}
unresolved.retain(|n| {
abstract_here.iter().any(|a| a == n)
|| !(methods.contains_key(n)
|| properties.contains_key(n)
|| static_methods.contains_key(n)
|| class_methods.contains_key(n)
|| class_attrs.contains_key(n))
});
unresolved.into_iter().collect()
};
state.classes.insert(class_name.to_string(), {
let mut cv = ClassValue::new(class_name);
cv.methods = methods;
cv.class_attrs = class_attrs;
cv.bases = bases;
cv.mro = mro;
cv.properties = properties;
cv.static_methods = static_methods;
cv.class_methods = class_methods;
cv.enum_kind = enum_kind;
cv.enum_members = enum_members;
cv.annotations = annotations;
cv.initvar_fields = initvar_fields;
cv.slots = slots;
cv.slot_names = slot_names;
cv.abstract_methods = abstract_methods;
cv.metaclass = metaclass_name.clone();
cv.qualname = qualname;
cv.is_generic = is_generic;
cv
});
state
.set_variable(class_name, Value::Class(class_name.to_string()))
.map_err(EvalError::Interpreter)?;
if is_typing_namedtuple {
finalize_typing_namedtuple(state, class_name)?;
}
if let Some(meta) = metaclass_name {
let saved_methods = state.classes.get(class_name).map(|c| {
(
c.methods.clone(),
c.properties.clone(),
c.static_methods.clone(),
c.class_methods.clone(),
)
});
invoke_metaclass_new(state, class_name, &meta, tools).await?;
if let Some((methods, properties, static_methods, class_methods)) = saved_methods {
if let Some(cv) = state.classes.get_mut(class_name) {
if cv.methods.is_empty() && !methods.is_empty() {
cv.methods = methods;
cv.properties = properties;
cv.static_methods = static_methods;
cv.class_methods = class_methods;
}
}
}
invoke_metaclass_init(state, class_name, &meta, tools).await?;
}
invoke_set_name(state, class_name, &attrs_for_set_name, tools).await?;
invoke_init_subclass(state, class_name, &bases_for_hook, &init_subclass_kwargs, tools).await?;
if !node.decorator_list.is_empty() {
let mut result = Value::Class(class_name.to_string());
for decorator in node.decorator_list.iter().rev() {
let dec_val = eval_expr(state, decorator, tools).await?;
result = apply_decorator(state, &dec_val, result, tools).await?;
}
state.set_variable(class_name, result).map_err(EvalError::Interpreter)?;
}
Ok(Value::None)
}
fn annotation_head_name(expr: &Expr) -> Option<&str> {
match expr {
Expr::Name(n) => Some(n.id.as_str()),
Expr::Attribute(a) => Some(a.attr.as_str()),
Expr::Subscript(s) => annotation_head_name(s.value.as_ref()),
_ => None,
}
}
fn is_abstractmethod_decorator(decorator: &Expr) -> bool {
let name = match decorator {
Expr::Name(n) => n.id.as_str(),
Expr::Attribute(a) => a.attr.as_str(),
_ => return false,
};
crate::eval::modules::abc_mod::ABSTRACT_DECORATORS.contains(&name)
}
fn is_builtin_method_decorator(dec: &Expr, properties: &BTreeMap<String, PropertyDef>) -> bool {
match dec {
Expr::Name(n) => {
matches!(n.id.as_str(), "property" | "cached_property" | "staticmethod" | "classmethod")
}
Expr::Attribute(a) if a.attr.as_str() == "cached_property" => true,
Expr::Attribute(a) if matches!(a.attr.as_str(), "setter" | "deleter") => {
matches!(a.value.as_ref(), Expr::Name(base) if properties.contains_key(base.id.as_str()))
}
_ => false,
}
}
fn build_plain_method_def(
state: &mut InterpreterState,
key: String,
params: crate::value::FunctionParams,
body: Vec<Stmt>,
source: String,
qualname: String,
is_async: bool,
) -> FunctionDef {
let (mut assigned_names, global_names) = crate::eval::functions::collect_assigned_names(&body);
assigned_names.retain(|n| !global_names.contains(n));
let is_generator = crate::eval::functions::contains_yield_stmts(&body);
let docstring = crate::eval::functions::extract_docstring(&body);
state.function_bodies.insert(key.clone(), Arc::new(body));
FunctionDef {
name: key,
body_key: String::new(),
wraps_name: None,
params,
closure: BTreeMap::new(),
source,
nonlocal_names: Vec::new(),
is_generator,
nonlocal_cell_id: None,
assigned_names,
global_names,
is_module_level: false,
docstring,
cell_refreshes: Vec::new(),
qualname,
annotations: Vec::new(),
is_async,
}
}
#[expect(
clippy::too_many_arguments,
reason = "the buckets travel as four separate `&mut BTreeMap`s by design; bundling them into a struct would just reify the same fan-out at the call site"
)]
fn classify_decorated_method(
state: &mut InterpreterState,
decorators: &[Expr],
class_name: &str,
method_name: String,
params: crate::value::FunctionParams,
body: Vec<Stmt>,
source: String,
methods: &mut BTreeMap<String, FunctionDef>,
properties: &mut BTreeMap<String, PropertyDef>,
static_methods: &mut BTreeMap<String, FunctionDef>,
class_methods: &mut BTreeMap<String, FunctionDef>,
) -> Result<(), EvalError> {
let method_qualname = state.qualname_for(&method_name);
let mut register = |key: String, body: Vec<Stmt>, source: String| -> FunctionDef {
let (mut assigned_names, global_names) =
crate::eval::functions::collect_assigned_names(&body);
assigned_names.retain(|n| !global_names.contains(n));
let is_generator = crate::eval::functions::contains_yield_stmts(&body);
let docstring = crate::eval::functions::extract_docstring(&body);
state.function_bodies.insert(key.clone(), Arc::new(body));
FunctionDef {
name: key,
body_key: String::new(),
wraps_name: None,
params: params.clone(),
closure: BTreeMap::new(),
source,
nonlocal_names: Vec::new(),
is_generator,
nonlocal_cell_id: None,
assigned_names,
global_names,
is_module_level: false,
docstring,
cell_refreshes: Vec::new(),
qualname: method_qualname.clone(),
annotations: Vec::new(),
is_async: false,
}
};
if decorators.is_empty() {
let key = format!("{class_name}.{method_name}");
let func = register(key, body, source);
methods.insert(method_name, func);
return Ok(());
}
if decorators.len() == 1 {
match &decorators[0] {
Expr::Name(n) if n.id.as_str() == "property" => {
let key = format!("{class_name}.{method_name}__get");
let func = register(key, body, source);
properties.insert(
method_name,
PropertyDef { getter: func, setter: None, deleter: None, cached: false },
);
return Ok(());
}
Expr::Name(n) if n.id.as_str() == "cached_property" => {
let key = format!("{class_name}.{method_name}__get");
let func = register(key, body, source);
properties.insert(
method_name,
PropertyDef { getter: func, setter: None, deleter: None, cached: true },
);
return Ok(());
}
Expr::Attribute(a) if a.attr.as_str() == "cached_property" => {
let key = format!("{class_name}.{method_name}__get");
let func = register(key, body, source);
properties.insert(
method_name,
PropertyDef { getter: func, setter: None, deleter: None, cached: true },
);
return Ok(());
}
Expr::Name(n) if n.id.as_str() == "staticmethod" => {
let key = format!("{class_name}.{method_name}__static");
let func = register(key, body, source);
static_methods.insert(method_name, func);
return Ok(());
}
Expr::Name(n) if n.id.as_str() == "classmethod" => {
let key = format!("{class_name}.{method_name}__class");
let func = register(key, body, source);
class_methods.insert(method_name, func);
return Ok(());
}
Expr::Attribute(attr) => {
if let Expr::Name(prop_name) = attr.value.as_ref() {
let prop_key = prop_name.id.as_str().to_string();
let kind = attr.attr.as_str();
if properties.contains_key(&prop_key) {
let suffix = match kind {
"setter" => "__set",
"deleter" => "__del",
_ => "",
};
if !suffix.is_empty() {
let key = format!("{class_name}.{prop_key}{suffix}");
let func = register(key, body, source);
if let Some(prop) = properties.get_mut(&prop_key) {
match kind {
"setter" => {
prop.setter = Some(func);
return Ok(());
}
"deleter" => {
prop.deleter = Some(func);
return Ok(());
}
_ => {}
}
}
}
}
}
}
_ => {}
}
}
Err(InterpreterError::Security(format!(
"method-level decorator stack on '{method_name}' is not one of the supported \
shapes (@property, @<name>.setter, @<name>.deleter, @staticmethod, @classmethod). \
See CONFORMANCE.md#unsupported-language-features.",
))
.into())
}
pub(crate) async fn apply_decorator(
state: &mut InterpreterState,
decorator: &Value,
target: Value,
tools: &Tools,
) -> EvalResult {
let kwargs: IndexMap<String, Value> = IndexMap::new();
match decorator {
Value::Function(def) => {
let positional = [target];
call_user_function(state, def, &positional, &kwargs, tools).await
}
Value::Lambda(def) => {
let positional = [target];
call_lambda(state, def, &positional, &kwargs, tools).await
}
Value::ModuleFunction { module, name }
if module == "dataclasses" && name == "dataclass" =>
{
let class_name = match &target {
Value::Class(n) => n.clone(),
other => {
return Err(InterpreterError::TypeError(format!(
"@dataclass requires a class target (got '{}')",
other.type_name()
))
.into());
}
};
crate::eval::modules::dataclasses::apply_dataclass(state, &class_name, &kwargs)?;
Ok(Value::Class(class_name))
}
Value::ModuleFunction { module, name }
if module == "functools" && (name == "lru_cache" || name == "cache") =>
{
let maxsize = if name == "cache" { None } else { Some(128) };
Ok(crate::eval::modules::functools::make_lru_cache_pub(target, maxsize))
}
Value::ModuleFunction { module, name } => {
crate::eval::modules::call_function(
state,
module,
name,
std::slice::from_ref(&target),
&kwargs,
tools,
)
.await
}
Value::Partial(data) => {
if let Value::ModuleFunction { module, name } = &data.func {
if module == "dataclasses" && name == "dataclass" {
let class_name = match &target {
Value::Class(n) => n.clone(),
other => {
return Err(InterpreterError::TypeError(format!(
"@dataclass requires a class target (got '{}')",
other.type_name()
))
.into());
}
};
crate::eval::modules::dataclasses::apply_dataclass(
state,
&class_name,
&data.keywords,
)?;
return Ok(Value::Class(class_name));
}
}
let mut combined = data.args.clone();
combined.push(target);
crate::eval::functions::call_value_as_function(
state,
&data.func,
&combined,
&indexmap::IndexMap::new(),
tools,
)
.await
}
other => Err(InterpreterError::TypeError(format!(
"decorator is not callable (got '{}')",
other.type_name()
))
.into()),
}
}
async fn invoke_set_name(
state: &mut InterpreterState,
class_name: &str,
attrs: &[(String, Value)],
tools: &Tools,
) -> Result<(), EvalError> {
let owner = Value::Class(class_name.to_string());
let empty_kwargs = IndexMap::new();
for (name, value) in attrs {
let Value::Instance(inst) = value else {
continue;
};
let Some((_, def)) = lookup_method_in_mro(state, &inst.class_name, "__set_name__") else {
continue;
};
let name_val = Value::String(name.as_str().into());
let call = CallArgs { positional: &[owner.clone(), name_val], keyword: &empty_kwargs };
let (_ret, _self) = call_method(state, &def, value.clone(), call, tools).await?;
}
Ok(())
}
fn finalize_typing_namedtuple(
state: &mut InterpreterState,
class_name: &str,
) -> Result<(), EvalError> {
let (field_vals, default_values, user_methods) = {
let class = state
.classes
.get(class_name)
.ok_or_else(|| EvalError::from(InterpreterError::name_not_defined(class_name)))?;
let fields = class.annotations.clone();
let default_values: Vec<Value> =
fields.iter().filter_map(|f| class.class_attrs.get(f).cloned()).collect();
let field_vals: Vec<Value> =
fields.iter().map(|f| Value::String(f.as_str().into())).collect();
(field_vals, default_values, class.methods.clone())
};
crate::eval::modules::collections::call_namedtuple_with_state(
state,
&[Value::String(class_name.into()), Value::Tuple(field_vals)],
&IndexMap::new(),
)?;
if let Some(class) = state.classes.get_mut(class_name) {
for (name, def) in user_methods {
class.methods.entry(name).or_insert(def);
}
if let Some(init) = class.methods.get_mut("__init__") {
init.params.defaults = vec!["None".to_string(); default_values.len()];
init.params.default_values = default_values;
}
}
Ok(())
}
async fn invoke_init_subclass(
state: &mut InterpreterState,
class_name: &str,
bases: &[String],
kwargs: &IndexMap<String, Value>,
tools: &Tools,
) -> Result<(), EvalError> {
let new_cls = Value::Class(class_name.to_string());
for base in bases {
let method = lookup_class_method(state, base, "__init_subclass__")
.or_else(|| lookup_method_in_mro(state, base, "__init_subclass__").map(|(_, d)| d));
let Some(def) = method else {
continue;
};
let call = CallArgs { positional: &[], keyword: kwargs };
let (_ret, _self) = call_method(state, &def, new_cls.clone(), call, tools).await?;
break;
}
Ok(())
}
fn bind_class_target(
target: &Expr,
value: &Value,
enum_kind: Option<crate::value::EnumKind>,
class_name: &str,
class_attrs: &mut BTreeMap<String, Value>,
auto_next: &mut i64,
) -> Result<(), EvalError> {
match target {
Expr::Name(name) => {
let attr = name.id.as_str().to_string();
let wrapped = wrap_enum_member(enum_kind, class_name, &attr, value.clone(), auto_next);
class_attrs.insert(attr, wrapped);
Ok(())
}
Expr::Tuple(t) => {
unpack_class_targets(&t.elts, value, enum_kind, class_name, class_attrs, auto_next)
}
Expr::List(l) => {
unpack_class_targets(&l.elts, value, enum_kind, class_name, class_attrs, auto_next)
}
_ => Ok(()),
}
}
fn unpack_class_targets(
elts: &[Expr],
value: &Value,
enum_kind: Option<crate::value::EnumKind>,
class_name: &str,
class_attrs: &mut BTreeMap<String, Value>,
auto_next: &mut i64,
) -> Result<(), EvalError> {
let items: Vec<Value> = match value {
Value::Tuple(items) => items.clone(),
Value::List(items) => items.lock().clone(),
Value::String(s) => s.chars().map(|c| Value::String(c.to_string().into())).collect(),
_ => {
return Err(InterpreterError::TypeError(format!(
"cannot unpack non-iterable {} object",
value.type_name()
))
.into());
}
};
if items.len() != elts.len() {
return Err(InterpreterError::ValueError(if items.len() > elts.len() {
format!("too many values to unpack (expected {})", elts.len())
} else {
format!("not enough values to unpack (expected {}, got {})", elts.len(), items.len())
})
.into());
}
for (elt, item) in elts.iter().zip(items) {
bind_class_target(elt, &item, enum_kind, class_name, class_attrs, auto_next)?;
}
Ok(())
}
fn wrap_enum_member(
enum_kind: Option<crate::value::EnumKind>,
class_name: &str,
member_name: &str,
value: Value,
auto_next: &mut i64,
) -> Value {
let Some(kind) = enum_kind else { return value };
if matches!(value, Value::Function(_) | Value::Lambda(_)) {
return value;
}
if matches!(value, Value::EnumMember { .. }) || member_name.starts_with('_') {
return value;
}
let value = if crate::eval::modules::enum_mod::is_auto_sentinel(&value) {
if matches!(kind, crate::value::EnumKind::Str) {
Value::String(member_name.to_lowercase().into())
} else if kind.is_flag() {
let n = u64::try_from((*auto_next).max(1)).unwrap_or(1).next_power_of_two();
*auto_next = i64::try_from(n.saturating_mul(2)).unwrap_or(i64::MAX);
Value::Int(i64::try_from(n).unwrap_or(i64::MAX))
} else {
let n = *auto_next;
*auto_next = n + 1;
Value::Int(n)
}
} else {
if let Value::Int(i) = &value {
*auto_next = (*auto_next).max(i.saturating_add(1));
}
value
};
Value::EnumMember {
class_name: class_name.to_string(),
member_name: member_name.to_string(),
value: Box::new(value),
kind,
}
}
fn build_mro(
class_name: &str,
bases: &[String],
registry: &rustc_hash::FxHashMap<String, ClassValue>,
) -> Result<Vec<String>, EvalError> {
let mut sequences: Vec<Vec<String>> = bases
.iter()
.map(|b| registry.get(b).map_or_else(|| vec![b.clone()], |cls| cls.mro.clone()))
.collect();
sequences.push(bases.to_vec());
let mut result: Vec<String> = vec![class_name.to_string()];
while !sequences.iter().all(Vec::is_empty) {
let candidate = sequences
.iter()
.filter_map(|seq| seq.first())
.find(|head| sequences.iter().all(|seq| !seq.iter().skip(1).any(|n| n == *head)))
.cloned();
let Some(head) = candidate else {
return Err(InterpreterError::TypeError(format!(
"Cannot create a consistent method resolution\norder (MRO) for bases of class '{class_name}'"
))
.into());
};
result.push(head.clone());
for seq in &mut sequences {
if seq.first() == Some(&head) {
seq.remove(0);
}
}
sequences.retain(|seq| !seq.is_empty());
}
Ok(result)
}
fn class_mro_is_exception(class: &ClassValue) -> bool {
class.mro.iter().any(|b| {
b == "Exception"
|| b == "BaseException"
|| crate::eval::functions::is_exception_type_name(b)
})
}
fn exception_message_from_args(args: &[Value]) -> String {
match args {
[] => String::new(),
[single] => format!("{single}"),
_ => format!("{}", Value::Tuple(args.to_vec())),
}
}
fn instance_to_exception(class_name: &str, inst: &InstanceValue) -> crate::value::ExceptionValue {
let fields = inst.fields.lock();
let args: Vec<Value> = match fields.get("args") {
Some(Value::Tuple(items)) => items.clone(),
Some(Value::List(items)) => items.lock().clone(),
_ => Vec::new(),
};
let mut exc = crate::value::ExceptionValue::new(
class_name.to_string(),
exception_message_from_args(&args),
)
.with_args(args);
for (name, value) in fields.iter() {
if name != "args" {
exc.fields.insert(name.clone(), value.clone());
}
}
exc
}
pub async fn instantiate(
state: &mut InterpreterState,
class_name: &str,
args: &[Value],
kwargs: &IndexMap<String, Value>,
tools: &Tools,
) -> EvalResult {
if let Some(meta) = state.classes.get(class_name).and_then(|c| c.metaclass.clone()) {
if let Some((_, call_def)) = lookup_method_in_mro(state, &meta, "__call__") {
let call = CallArgs { positional: args, keyword: kwargs };
let (returned, _) =
call_method(state, &call_def, Value::Class(class_name.to_string()), call, tools)
.await?;
return Ok(returned);
}
}
instantiate_default(state, class_name, args, kwargs, tools).await
}
async fn instantiate_default(
state: &mut InterpreterState,
class_name: &str,
args: &[Value],
kwargs: &IndexMap<String, Value>,
tools: &Tools,
) -> EvalResult {
if let Some(class) = state.classes.get(class_name) {
if !class.abstract_methods.is_empty() {
let names = class
.abstract_methods
.iter()
.map(|m| format!("'{m}'"))
.collect::<Vec<_>>()
.join(", ");
let noun = if class.abstract_methods.len() == 1 { "method" } else { "methods" };
return Err(crate::error::EvalError::Exception(crate::value::ExceptionValue::new(
"TypeError",
format!(
"Can't instantiate abstract class {class_name} without an implementation for abstract {noun} {names}"
),
)));
}
}
if let Some(class) = state.classes.get(class_name) {
if class.enum_kind.is_some() && args.len() == 1 && kwargs.is_empty() {
let needle = &args[0];
for (member_name, member_value) in &class.class_attrs {
if let Value::EnumMember { value, .. } = member_value {
if crate::eval::operations::values_equal_pub(value, needle) {
return Ok(member_value.clone());
}
let _ = member_name;
}
}
return Err(crate::error::EvalError::Exception(crate::value::ExceptionValue::new(
"ValueError",
format!("{needle} is not a valid {class_name}"),
)));
}
let is_exception_subclass = class_mro_is_exception(class);
if is_exception_subclass {
let Some((_defining_class, init_def)) =
lookup_method_in_mro(state, class_name, "__init__")
else {
return Ok(Value::Exception(Box::new(
crate::value::ExceptionValue::new(
class_name.to_string(),
exception_message_from_args(args),
)
.with_args(args.to_vec()),
)));
};
let mut seed = BTreeMap::new();
seed.insert("args".to_string(), Value::Tuple(args.to_vec()));
let instance = Value::Instance(InstanceValue {
class_name: class_name.to_string(),
fields: crate::value::shared_fields(seed),
});
let call = CallArgs { positional: args, keyword: kwargs };
let (_returned, configured_self) =
call_method(state, &init_def, instance, call, tools).await?;
let Value::Instance(inst) = configured_self else {
return Ok(Value::Exception(Box::new(
crate::value::ExceptionValue::new(
class_name.to_string(),
exception_message_from_args(args),
)
.with_args(args.to_vec()),
)));
};
return Ok(Value::Exception(Box::new(instance_to_exception(class_name, &inst))));
}
}
if let Some((_, new_def)) = lookup_method_in_mro(state, class_name, "__new__") {
let call = CallArgs { positional: args, keyword: kwargs };
let (obj, _) =
call_method(state, &new_def, Value::Class(class_name.to_string()), call, tools).await?;
if let Value::Instance(inst) = &obj {
let returns_cls = inst.class_name == class_name
|| state
.classes
.get(&inst.class_name)
.is_some_and(|c| c.mro.iter().any(|a| a == class_name));
if returns_cls {
if let Some((_, init_def)) = lookup_method_in_mro(state, class_name, "__init__") {
let call = CallArgs { positional: args, keyword: kwargs };
let (_r, configured) =
call_method(state, &init_def, obj.clone(), call, tools).await?;
return Ok(configured);
}
}
}
return Ok(obj);
}
let instance = Value::Instance(InstanceValue {
class_name: class_name.to_string(),
fields: crate::value::shared_fields(BTreeMap::new()),
});
let has_user_init =
state.classes.get(class_name).is_some_and(|c| c.methods.contains_key("__init__"));
if !has_user_init {
if let Some(fields) = state.classes.get(class_name).and_then(|c| c.dataclass_fields.clone())
{
return dataclass_instantiate(state, class_name, &fields, args, kwargs, tools).await;
}
}
let init = lookup_method_in_mro(state, class_name, "__init__");
let Some((_defining_class, init_def)) = init else {
if !args.is_empty() || !kwargs.is_empty() {
return Err(
InterpreterError::TypeError(format!("{class_name}() takes no arguments")).into()
);
}
return Ok(instance);
};
let call = CallArgs { positional: args, keyword: kwargs };
let (_returned, configured_self) = call_method(state, &init_def, instance, call, tools).await?;
Ok(configured_self)
}
async fn dataclass_instantiate(
state: &mut InterpreterState,
class_name: &str,
fields: &[crate::value::DataclassField],
args: &[Value],
kwargs: &IndexMap<String, Value>,
tools: &Tools,
) -> EvalResult {
let init_fields: Vec<&crate::value::DataclassField> =
fields.iter().filter(|f| f.init).collect();
if args.len() > init_fields.len() {
return Err(InterpreterError::TypeError(format!(
"{class_name}() takes {} positional arguments but {} were given",
init_fields.len(),
args.len()
))
.into());
}
for key in kwargs.keys() {
if !init_fields.iter().any(|f| &f.name == key) {
return Err(InterpreterError::TypeError(format!(
"{class_name}() got an unexpected keyword argument '{key}'"
))
.into());
}
}
let mut instance_fields: BTreeMap<String, Value> = BTreeMap::new();
let mut initvar_values: Vec<Value> = Vec::new();
for (index, field) in init_fields.iter().enumerate() {
let positional = args.get(index).cloned();
let keyword = kwargs.get(&field.name).cloned();
let value = match (positional, keyword) {
(Some(_), Some(_)) => {
return Err(InterpreterError::TypeError(format!(
"{class_name}() got multiple values for argument '{}'",
field.name
))
.into());
}
(Some(v), None) | (None, Some(v)) => v,
(None, None) => {
if let Some(default) = field.default.clone() {
default
} else if let Some(factory) = field.default_factory.clone() {
invoke_default_factory(state, &factory, tools).await?
} else {
return Err(InterpreterError::TypeError(format!(
"{class_name}() missing required argument: '{}'",
field.name
))
.into());
}
}
};
if field.init_only {
initvar_values.push(value);
} else {
instance_fields.insert(field.name.clone(), value);
}
}
for field in fields.iter().filter(|f| !f.init) {
let value = if let Some(default) = field.default.clone() {
default
} else if let Some(factory) = field.default_factory.clone() {
invoke_default_factory(state, &factory, tools).await?
} else {
continue;
};
instance_fields.insert(field.name.clone(), value);
}
let instance = Value::Instance(InstanceValue {
class_name: class_name.to_string(),
fields: crate::value::shared_fields(instance_fields),
});
if let Some((_, post)) = lookup_method_in_mro(state, class_name, "__post_init__") {
let call = CallArgs { positional: &initvar_values, keyword: &IndexMap::new() };
let (_returned, updated) = call_method(state, &post, instance, call, tools).await?;
return Ok(updated);
}
Ok(instance)
}
fn empty_for_builtin_factory(name: &str) -> EvalResult {
match name {
"list" => Ok(Value::List(shared_list(Vec::new()))),
"dict" => Ok(Value::Dict(crate::value::shared_dict(IndexMap::new()))),
"set" => Ok(Value::new_set(Vec::new())),
"frozenset" => Ok(Value::new_frozenset(Vec::new())),
"tuple" => Ok(Value::Tuple(Vec::new())),
"str" => Ok(Value::String("".into())),
other => {
Err(InterpreterError::TypeError(format!("default_factory '{other}' is not callable"))
.into())
}
}
}
async fn invoke_default_factory(
state: &mut InterpreterState,
factory: &Value,
tools: &Tools,
) -> EvalResult {
let empty_kwargs: IndexMap<String, Value> = IndexMap::new();
match factory {
Value::Function(def) => call_user_function(state, def, &[], &empty_kwargs, tools).await,
Value::Lambda(def) => call_lambda(state, def, &[], &empty_kwargs, tools).await,
Value::Class(name) => Box::pin(instantiate(state, name, &[], &empty_kwargs, tools)).await,
Value::ModuleFunction { module, name } => {
crate::eval::modules::call_function(state, module, name, &[], &empty_kwargs, tools)
.await
}
Value::Type(t) => empty_for_builtin_factory(t.as_str()),
Value::BuiltinName(name) => empty_for_builtin_factory(name.as_str()),
other => Err(InterpreterError::TypeError(format!(
"default_factory is not callable (got '{}')",
other.type_name()
))
.into()),
}
}
pub async fn instance_method_call(
state: &mut InterpreterState,
instance: Value,
method_name: &str,
call: CallArgs<'_>,
tools: &Tools,
) -> Result<(Value, Value), EvalError> {
let Value::Instance(inst) = &instance else {
return Err(
InterpreterError::Runtime("instance_method_call on a non-instance".into()).into()
);
};
if inst.class_name == crate::eval::modules::contextlib_mod::EXITSTACK_CLASS {
if let Some(result) = crate::eval::modules::contextlib_mod::try_exitstack_method(
state,
&instance,
method_name,
call.positional,
tools,
)
.await
{
return Ok((result?, instance));
}
}
let class_name = inst.class_name.clone();
if let Some(def) = lookup_static_method(state, &class_name, method_name) {
let result = call_user_function(state, &def, call.positional, call.keyword, tools).await?;
return Ok((result, instance));
}
if let Some(def) = lookup_class_method(state, &class_name, method_name) {
let (result, _cls) =
call_method(state, &def, Value::Class(class_name.clone()), call, tools).await?;
return Ok((result, instance));
}
let method = lookup_method_in_mro(state, &class_name, method_name);
let Some((_defining_class, def)) = method else {
return Box::pin(instance_attr_call_fallback(
state,
instance,
&class_name,
method_name,
call,
tools,
))
.await;
};
call_method(state, &def, instance, call, tools).await
}
async fn instance_attr_call_fallback(
state: &mut InterpreterState,
instance: Value,
class_name: &str,
method_name: &str,
call: CallArgs<'_>,
tools: &Tools,
) -> Result<(Value, Value), EvalError> {
let field = match &instance {
Value::Instance(inst) => inst.fields.lock().get(method_name).cloned(),
_ => None,
};
if let Some(f) = field {
let result = crate::eval::functions::call_value_as_function(
state,
&f,
call.positional,
call.keyword,
tools,
)
.await?;
return Ok((result, instance));
}
let class_attr = state.classes.get(class_name).and_then(|c| {
c.mro
.iter()
.find_map(|anc| state.classes.get(anc).and_then(|a| a.class_attrs.get(method_name)))
.cloned()
});
if let Some(attr @ (Value::Function(_) | Value::Lambda(_))) = class_attr {
let mut full = Vec::with_capacity(call.positional.len() + 1);
full.push(instance.clone());
full.extend_from_slice(call.positional);
let result = crate::eval::functions::call_value_as_function(
state,
&attr,
&full,
call.keyword,
tools,
)
.await?;
return Ok((result, instance));
}
if let Some((_, getattr)) = lookup_method_in_mro(state, class_name, "__getattr__") {
let attr_arg = Value::String(method_name.into());
let empty_kwargs = indexmap::IndexMap::new();
let getattr_call =
CallArgs { positional: std::slice::from_ref(&attr_arg), keyword: &empty_kwargs };
let (attr_value, _self) =
call_method(state, &getattr, instance.clone(), getattr_call, tools).await?;
let result = crate::eval::functions::call_value_as_function(
state,
&attr_value,
call.positional,
call.keyword,
tools,
)
.await?;
return Ok((result, instance));
}
if matches!(method_name, "count" | "index") {
if let Value::Instance(inst) = &instance {
if let Some(Value::Tuple(field_names)) = state
.classes
.get(inst.class_name.as_str())
.and_then(|c| c.class_attrs.get("_fields"))
.cloned()
{
let items: Vec<Value> = field_names
.iter()
.filter_map(|n| match n {
Value::String(f) => {
Some(inst.fields.lock().get(f.as_str()).cloned().unwrap_or(Value::None))
}
_ => None,
})
.collect();
let result = crate::eval::functions::methods::tuple::dispatch_tuple_method(
&items,
method_name,
call.positional,
call.keyword,
)?;
return Ok((result, instance));
}
}
}
Err(InterpreterError::AttributeError(format!(
"'{class_name}' object has no attribute '{method_name}'"
))
.into())
}
fn apply_method_scope(
state: &mut InterpreterState,
local_scope: &rustc_hash::FxHashMap<String, Value>,
) -> Result<(), EvalError> {
for (name, value) in local_scope {
state.set_variable(name, value.clone()).map_err(EvalError::Interpreter)?;
}
Ok(())
}
pub async fn call_method(
state: &mut InterpreterState,
method: &FunctionDef,
self_value: Value,
call: CallArgs<'_>,
tools: &Tools,
) -> Result<(Value, Value), EvalError> {
crate::eval::functions::dispatch::grow_stack(call_method_inner(
state, method, self_value, call, tools,
))
.await
}
async fn call_method_inner(
state: &mut InterpreterState,
method: &FunctionDef,
self_value: Value,
call: CallArgs<'_>,
tools: &Tools,
) -> Result<(Value, Value), EvalError> {
state.enter_call().map_err(EvalError::Interpreter)?;
state.frame_cell_owners.push(rustc_hash::FxHashMap::default());
let defining_class =
method.name.split_once('.').map_or_else(|| method.name.clone(), |(cls, _)| cls.to_string());
let self_local_name = method.params.args.first().map(|p| p.name.clone());
let frame_pushed = if matches!(&self_value, Value::Instance(_) | Value::Class(_)) {
state.method_frame_stack.push(crate::state::MethodFrame {
defining_class,
self_value: self_value.clone(),
self_local_name: self_local_name.clone(),
});
true
} else {
false
};
let mut full_args = Vec::with_capacity(call.positional.len() + 1);
full_args.push(self_value);
full_args.extend_from_slice(call.positional);
let local_scope = match bind_params_named(
&method.params,
method.display_qualname(),
&full_args,
call.keyword,
state,
tools,
)
.await
{
Ok(scope) => scope,
Err(e) => {
if frame_pushed {
state.method_frame_stack.pop();
}
state.frame_cell_owners.pop();
state.exit_call();
return Err(e);
}
};
let self_param = self_local_name.clone();
let touched: Vec<String> = method
.params
.args
.iter()
.map(|p| p.name.clone())
.chain(method.params.vararg.iter().cloned())
.chain(method.params.kwonlyargs.iter().map(|p| p.name.clone()))
.chain(method.params.kwarg.iter().cloned())
.chain(method.assigned_names.iter().cloned())
.filter(|n| !method.global_names.contains(n))
.collect();
let checkpoint = crate::eval::functions::VariableCheckpoint::capture(state, &touched);
if let Err(e) = apply_method_scope(state, &local_scope) {
checkpoint.restore(state);
if frame_pushed {
state.method_frame_stack.pop();
}
state.frame_cell_owners.pop();
state.exit_call();
return Err(e);
}
let body = state.function_bodies.get(&method.name).cloned();
let exec_result = match body {
Some(stmts) if method.is_generator => {
state.yield_stack.push(Vec::new());
let body_result = execute_body(state, stmts.as_slice(), tools).await;
let collected = state.yield_stack.pop().unwrap_or_default();
match body_result {
Ok(_) | Err(EvalError::Signal(crate::error::ControlFlow::Return(_))) => {
let cursor_id = state.next_cursor_id;
state.next_cursor_id = state.next_cursor_id.wrapping_add(1);
state.lazy_cursors.insert(cursor_id, 0);
Ok(Value::Lazy {
items: collected,
cursor_id,
kind: crate::value::LazyKind::Generator,
})
}
Err(e) => Err(e),
}
}
Some(stmts) => execute_body(state, stmts.as_slice(), tools).await,
None => Ok(Value::None),
};
let configured_self =
self_param.and_then(|name| state.variables.get(&name).cloned()).unwrap_or(Value::None);
checkpoint.restore(state);
if frame_pushed {
state.method_frame_stack.pop();
}
state.frame_cell_owners.pop();
state.exit_call();
let returned = match exec_result {
Ok(val) => val,
Err(EvalError::Signal(crate::error::ControlFlow::Return(val))) => *val,
Err(e) => return Err(e),
};
Ok((returned, configured_self))
}
pub fn instance_attribute(
state: &InterpreterState,
inst: &InstanceValue,
attr: &str,
) -> EvalResult {
if let Some(value) = inst.fields.lock().get(attr) {
return Ok(value.clone());
}
if let Some(value) = lookup_class_attr(state, &inst.class_name, attr) {
return Ok(value);
}
Err(InterpreterError::AttributeError(format!(
"'{}' object has no attribute '{attr}'",
inst.class_name
))
.into())
}
pub fn lookup_class_attr_instance(
state: &InterpreterState,
class_name: &str,
attr: &str,
) -> Option<InstanceValue> {
let class = state.classes.get(class_name)?;
for ancestor_name in &class.mro {
if let Some(ancestor) = state.classes.get(ancestor_name) {
if let Some(Value::Instance(inst)) = ancestor.class_attrs.get(attr) {
return Some(inst.clone());
}
}
}
None
}
pub fn lookup_property(
state: &InterpreterState,
class_name: &str,
attr: &str,
) -> Option<PropertyDef> {
let class = state.classes.get(class_name)?;
for ancestor_name in &class.mro {
if let Some(ancestor) = state.classes.get(ancestor_name) {
if let Some(prop) = ancestor.properties.get(attr) {
return Some(prop.clone());
}
}
}
None
}
pub fn lookup_static_method(
state: &InterpreterState,
class_name: &str,
method_name: &str,
) -> Option<FunctionDef> {
let class = state.classes.get(class_name)?;
for ancestor_name in &class.mro {
if let Some(ancestor) = state.classes.get(ancestor_name) {
if let Some(def) = ancestor.static_methods.get(method_name) {
return Some(def.clone());
}
}
}
None
}
pub fn lookup_class_method(
state: &InterpreterState,
class_name: &str,
method_name: &str,
) -> Option<FunctionDef> {
let class = state.classes.get(class_name)?;
for ancestor_name in &class.mro {
if let Some(ancestor) = state.classes.get(ancestor_name) {
if let Some(def) = ancestor.class_methods.get(method_name) {
return Some(def.clone());
}
}
}
None
}
pub async fn invoke_property_getter(
state: &mut InterpreterState,
getter: &FunctionDef,
instance: Value,
cache_key: Option<&str>,
tools: &Tools,
) -> EvalResult {
if let Some(key) = cache_key {
if let Value::Instance(inst) = &instance {
if let Some(v) = inst.fields.lock().get(key) {
return Ok(v.clone());
}
}
}
let call = CallArgs { positional: &[], keyword: &IndexMap::new() };
let (returned, _self) = call_method(state, getter, instance.clone(), call, tools).await?;
if let Some(key) = cache_key {
if let Value::Instance(inst) = &instance {
inst.fields.lock().insert(key.to_string(), returned.clone());
}
}
Ok(returned)
}
pub async fn invoke_property_setter(
state: &mut InterpreterState,
setter: &FunctionDef,
instance: Value,
value: Value,
tools: &Tools,
) -> Result<Value, EvalError> {
let call = CallArgs { positional: &[value], keyword: &IndexMap::new() };
let (_returned, configured_self) = call_method(state, setter, instance, call, tools).await?;
Ok(configured_self)
}
pub async fn invoke_property_deleter(
state: &mut InterpreterState,
deleter: &FunctionDef,
instance: Value,
tools: &Tools,
) -> Result<Value, EvalError> {
let call = CallArgs { positional: &[], keyword: &IndexMap::new() };
let (_returned, configured_self) = call_method(state, deleter, instance, call, tools).await?;
Ok(configured_self)
}
pub fn class_attribute(state: &InterpreterState, class_name: &str, attr: &str) -> EvalResult {
if attr == "__name__" {
return Ok(Value::String(class_name.into()));
}
if attr == "__qualname__" {
let qualname = state
.classes
.get(class_name)
.map(|c| if c.qualname.is_empty() { class_name } else { c.qualname.as_str() })
.unwrap_or(class_name);
return Ok(Value::String(qualname.into()));
}
if let Some(def) = lookup_static_method(state, class_name, attr) {
return Ok(Value::Function(std::sync::Arc::new(def)));
}
if let Some(def) = lookup_class_method(state, class_name, attr) {
let _ = def;
return Ok(Value::UnboundClassMethod {
class: class_name.to_string(),
method: attr.to_string(),
});
}
if let Some(value) = lookup_class_attr(state, class_name, attr) {
return Ok(value);
}
if let Some(class) = state.classes.get(class_name) {
for ancestor in &class.mro {
if state.classes.get(ancestor).is_some_and(|c| c.properties.contains_key(attr)) {
return Ok(Value::Property {
class_name: ancestor.clone(),
name: attr.to_string(),
});
}
}
}
if let Some((_, def)) = lookup_method_in_mro(state, class_name, attr) {
return Ok(Value::Function(std::sync::Arc::new(def)));
}
if let Some(meta) = state.classes.get(class_name).and_then(|c| c.metaclass.clone()) {
if let Some(value) = lookup_class_attr(state, &meta, attr) {
return Ok(value);
}
if let Some((_, def)) = lookup_method_in_mro(state, &meta, attr) {
return Ok(Value::Function(std::sync::Arc::new(def)));
}
}
Err(InterpreterError::AttributeError(format!(
"type object '{class_name}' has no attribute '{attr}'"
))
.into())
}
pub(crate) fn lookup_class_attr(
state: &InterpreterState,
class_name: &str,
attr: &str,
) -> Option<Value> {
let class = state.classes.get(class_name)?;
for ancestor_name in &class.mro {
if let Some(ancestor) = state.classes.get(ancestor_name) {
if let Some(value) = ancestor.class_attrs.get(attr) {
return Some(value.clone());
}
}
}
None
}
pub struct SuperReceiver<'a> {
pub defining_class: &'a str,
pub instance: InstanceValue,
}
pub async fn super_attribute(
state: &mut InterpreterState,
defining_class: &str,
instance: &InstanceValue,
attr_name: &str,
tools: &Tools,
) -> EvalResult {
let mro = state.classes.get(&instance.class_name).map(|c| c.mro.clone()).ok_or_else(|| {
EvalError::from(InterpreterError::Runtime(format!(
"super(): instance's class '{}' is not registered",
instance.class_name
)))
})?;
let start = mro.iter().position(|c| c == defining_class).ok_or_else(|| {
EvalError::from(InterpreterError::TypeError(format!(
"super(): '{defining_class}' is not in MRO of '{}'",
instance.class_name
)))
})?;
for ancestor_name in mro.iter().skip(start + 1) {
let Some(ancestor) = state.classes.get(ancestor_name) else { continue };
if let Some(prop) = ancestor.properties.get(attr_name) {
let getter = prop.getter.clone();
let cache_key = prop.cached.then_some(attr_name);
return invoke_property_getter(
state,
&getter,
Value::Instance(instance.clone()),
cache_key,
tools,
)
.await;
}
if let Some(v) = ancestor.class_attrs.get(attr_name) {
return Ok(v.clone());
}
if ancestor.methods.contains_key(attr_name) {
break;
}
}
Err(InterpreterError::AttributeError(format!("'super' object has no attribute '{attr_name}'"))
.into())
}
pub async fn super_method_call(
state: &mut InterpreterState,
recv: SuperReceiver<'_>,
method_name: &str,
call: CallArgs<'_>,
tools: &Tools,
) -> Result<(Value, Value), EvalError> {
let SuperReceiver { defining_class, instance } = recv;
let Some(class) = state.classes.get(&instance.class_name) else {
return Err(InterpreterError::Runtime(format!(
"super(): instance's class '{}' is not registered",
instance.class_name
))
.into());
};
let start = class.mro.iter().position(|c| c == defining_class).ok_or_else(|| {
EvalError::from(InterpreterError::TypeError(format!(
"super(): '{defining_class}' is not in MRO of '{}'",
instance.class_name
)))
})?;
let mut found = None;
for ancestor_name in class.mro.iter().skip(start + 1) {
if let Some(ancestor) = state.classes.get(ancestor_name) {
if let Some(def) = ancestor.methods.get(method_name) {
found = Some(def.clone());
break;
}
}
}
let Some(def) = found else {
return match method_name {
"__setattr__" => {
let attr_name = call
.positional
.first()
.and_then(|v| if let Value::String(s) = v { Some(s.clone()) } else { None })
.ok_or_else(|| {
EvalError::from(InterpreterError::TypeError(
"object.__setattr__: first argument must be str".into(),
))
})?;
let value = call.positional.get(1).cloned().ok_or_else(|| {
EvalError::from(InterpreterError::TypeError(
"object.__setattr__: missing value argument".into(),
))
})?;
crate::security::validator::validate_attribute(&attr_name)?;
let inst = instance;
inst.fields.lock().insert(attr_name.into(), value);
let updated = Value::Instance(inst);
if let Some(name) =
state.method_frame_stack.last().and_then(|f| f.self_local_name.clone())
{
state.set_variable(&name, updated.clone()).map_err(EvalError::Interpreter)?;
}
Ok((Value::None, updated))
}
"__delattr__" => {
let attr_name = call
.positional
.first()
.and_then(|v| if let Value::String(s) = v { Some(s.clone()) } else { None })
.ok_or_else(|| {
EvalError::from(InterpreterError::TypeError(
"object.__delattr__: argument must be str".into(),
))
})?;
crate::security::validator::validate_attribute(&attr_name)?;
let inst = instance;
let class_name = inst.class_name.clone();
if inst.fields.lock().remove(attr_name.as_str()).is_none() {
return Err(InterpreterError::AttributeError(format!(
"'{class_name}' object has no attribute '{attr_name}'"
))
.into());
}
let updated = Value::Instance(inst);
if let Some(name) =
state.method_frame_stack.last().and_then(|f| f.self_local_name.clone())
{
state.set_variable(&name, updated.clone()).map_err(EvalError::Interpreter)?;
}
Ok((Value::None, updated))
}
"__init__" => {
let inst = instance;
let is_exc =
state.classes.get(&inst.class_name).is_some_and(class_mro_is_exception);
if is_exc {
inst.fields
.lock()
.insert("args".into(), Value::Tuple(call.positional.to_vec()));
}
let updated = Value::Instance(inst);
if let Some(name) =
state.method_frame_stack.last().and_then(|f| f.self_local_name.clone())
{
state.set_variable(&name, updated.clone()).map_err(EvalError::Interpreter)?;
}
Ok((Value::None, updated))
}
"__getattribute__" => {
let attr_name = call
.positional
.first()
.and_then(|v| if let Value::String(s) = v { Some(s.clone()) } else { None })
.ok_or_else(|| {
EvalError::from(InterpreterError::TypeError(
"object.__getattribute__: argument must be str".into(),
))
})?;
let inst = instance;
let returned = crate::eval::names::getattr_normal_lookup(
state,
Value::Instance(inst.clone()),
&attr_name,
tools,
None,
)
.await?;
Ok((returned, Value::Instance(inst)))
}
_ => Err(InterpreterError::AttributeError(format!(
"'super' object has no attribute '{method_name}'"
))
.into()),
};
};
let caller_self_name = state.method_frame_stack.last().and_then(|f| f.self_local_name.clone());
let (returned, configured_self) =
call_method(state, &def, Value::Instance(instance), call, tools).await?;
if let Some(name) = caller_self_name {
let _ = state.set_variable(&name, configured_self.clone());
}
Ok((returned, configured_self))
}
pub async fn super_class_method_call(
state: &mut InterpreterState,
defining_class: &str,
class_name: &str,
method_name: &str,
call: CallArgs<'_>,
tools: &Tools,
) -> EvalResult {
let Some(class) = state.classes.get(class_name) else {
return Err(InterpreterError::Runtime(format!(
"super(): class '{class_name}' is not registered"
))
.into());
};
let start = match class.mro.iter().position(|c| c == defining_class) {
Some(s) => Some(s),
None if state
.classes
.get(defining_class)
.is_some_and(|c| c.bases.iter().any(|b| b == "type")) =>
{
None
}
None => {
return Err(InterpreterError::TypeError(format!(
"super(): '{defining_class}' is not in MRO of '{class_name}'"
))
.into());
}
};
let mut found = None;
if let Some(start) = start {
for ancestor_name in class.mro.iter().skip(start + 1) {
if let Some(ancestor) = state.classes.get(ancestor_name) {
let def = ancestor
.class_methods
.get(method_name)
.or_else(|| ancestor.static_methods.get(method_name))
.or_else(|| ancestor.methods.get(method_name));
if let Some(def) = def {
found = Some(def.clone());
break;
}
}
}
}
match found {
Some(def) => {
let (returned, _self) =
call_method(state, &def, Value::Class(class_name.to_string()), call, tools).await?;
Ok(returned)
}
None if method_name == "__new__"
&& state
.classes
.get(class_name)
.is_some_and(|c| c.bases.iter().any(|b| b == "type"))
&& call.positional.len() >= 4 =>
{
let Some(Value::String(new_name)) = call.positional.get(1) else {
return Err(InterpreterError::TypeError(
"type.__new__() argument 1 (name) must be str".into(),
)
.into());
};
let new_name = new_name.to_string();
if let Some(ns) = call.positional.get(3).and_then(Value::as_dict) {
let entries: Vec<(String, Value)> = ns
.lock()
.iter()
.filter_map(|(k, v)| match k {
crate::value::ValueKey::String(s) => Some((s.to_string(), v.clone())),
_ => None,
})
.collect();
let target = state
.classes
.entry(new_name.clone())
.or_insert_with(|| ClassValue::new(&new_name));
for (k, v) in entries {
target.class_attrs.insert(k, v);
}
}
Ok(Value::Class(new_name))
}
None if method_name == "__new__" => {
let target = match call.positional.first() {
Some(Value::Class(c)) => c.clone(),
_ => class_name.to_string(),
};
Ok(Value::Instance(InstanceValue {
class_name: target,
fields: crate::value::shared_fields(BTreeMap::new()),
}))
}
None if method_name == "__call__" => {
Box::pin(instantiate_default(state, class_name, call.positional, call.keyword, tools))
.await
}
None if matches!(method_name, "__init_subclass__" | "__set_name__" | "__init__") => {
Ok(Value::None)
}
None => Err(InterpreterError::AttributeError(format!(
"'super' object has no attribute '{method_name}'"
))
.into()),
}
}
fn lookup_method_or_classmethod(
state: &InterpreterState,
class_name: &str,
method_name: &str,
) -> Option<(String, FunctionDef)> {
if let Some(found) = lookup_method_in_mro(state, class_name, method_name) {
return Some(found);
}
let class = state.classes.get(class_name)?;
for ancestor_name in &class.mro {
if let Some(ancestor) = state.classes.get(ancestor_name) {
if let Some(def) = ancestor.class_methods.get(method_name) {
return Some((ancestor_name.clone(), def.clone()));
}
}
}
None
}
pub fn lookup_method_in_mro(
state: &InterpreterState,
class_name: &str,
method_name: &str,
) -> Option<(String, FunctionDef)> {
let class = state.classes.get(class_name)?;
for ancestor_name in &class.mro {
if let Some(ancestor) = state.classes.get(ancestor_name) {
if let Some(def) = ancestor.methods.get(method_name) {
return Some((ancestor_name.clone(), def.clone()));
}
}
}
None
}
pub(crate) fn class_mro_values(state: &InterpreterState, class_name: &str) -> Option<Vec<Value>> {
let class = state.classes.get(class_name)?;
let mut out: Vec<Value> = class
.mro
.iter()
.map(|n| {
if state.classes.contains_key(n) {
Value::Class(n.clone())
} else {
Value::Type(n.clone())
}
})
.collect();
if let Some(last) = class.mro.last() {
let mut cur: &str = last;
while let Some(parent) = crate::eval::exceptions::builtin_exception_parent(cur) {
out.push(Value::Type(parent.to_string()));
cur = parent;
}
}
if !matches!(out.last(), Some(Value::Type(n)) if n == "object") {
out.push(Value::Type("object".to_string()));
}
Some(out)
}
pub(crate) fn builtin_type_mro_values(name: &str) -> Vec<Value> {
if name == "object" {
return vec![Value::Type("object".to_string())];
}
let mut out = vec![Value::Type(name.to_string())];
if name == "bool" {
out.push(Value::Type("int".to_string()));
} else {
let mut cur = name;
while let Some(parent) = crate::eval::exceptions::builtin_exception_parent(cur) {
out.push(Value::Type(parent.to_string()));
cur = parent;
}
}
out.push(Value::Type("object".to_string()));
out
}
pub(crate) fn dynamic_type_new(
state: &mut InterpreterState,
name_v: &Value,
bases_v: &Value,
dict_v: &Value,
) -> Result<Value, EvalError> {
let Value::String(name) = name_v else {
return Err(InterpreterError::TypeError("type() argument 1 must be str".into()).into());
};
let class_name = name.to_string();
let mut bases: Vec<String> = Vec::new();
let base_items: Vec<Value> = match bases_v {
Value::Tuple(items) => items.clone(),
Value::List(l) => l.lock().clone(),
_ => {
return Err(
InterpreterError::TypeError("type() argument 2 must be a tuple".into()).into()
);
}
};
for b in base_items {
match b {
Value::Class(n) => bases.push(n),
Value::ExceptionType(n) => bases.push(n),
Value::Type(n) | Value::BuiltinName(n) if n == "object" || n == "type" => {}
other => {
return Err(InterpreterError::TypeError(format!(
"type() bases must be types, not '{}'",
other.type_name()
))
.into());
}
}
}
let mut class_attrs = BTreeMap::new();
if let Value::Dict(map) = dict_v {
for (k, v) in map.lock().iter() {
if let crate::value::ValueKey::String(s) = k {
class_attrs.insert(s.to_string(), v.clone());
}
}
} else {
return Err(InterpreterError::TypeError("type() argument 3 must be a dict".into()).into());
}
let mro = build_mro(&class_name, &bases, &state.classes)?;
let (slots, slot_names) = parse_slots_attr(class_attrs.get("__slots__"));
state.classes.insert(class_name.clone(), {
let mut cv = ClassValue::new(class_name.clone());
cv.class_attrs = class_attrs;
cv.bases = bases;
cv.mro = mro;
cv.slots = slots;
cv.slot_names = slot_names;
cv
});
Ok(Value::Class(class_name))
}
async fn invoke_metaclass_prepare(
state: &mut InterpreterState,
meta: &str,
class_name: &str,
bases: &[String],
tools: &Tools,
) -> Result<Option<IndexMap<crate::value::ValueKey, Value>>, EvalError> {
let Some((_, method)) = lookup_method_or_classmethod(state, meta, "__prepare__") else {
return Ok(None);
};
let name_v = Value::String(class_name.into());
let bases_t = Value::Tuple(bases.iter().map(|b| Value::Class(b.clone())).collect());
let empty_kw = IndexMap::new();
let returned = crate::eval::functions::call_user_function(
state,
&method,
&[name_v, bases_t],
&empty_kw,
tools,
)
.await?;
let _ = meta;
match returned {
Value::Dict(map) => Ok(Some(map.lock().clone())),
Value::None => Ok(None),
other => Err(InterpreterError::TypeError(format!(
"__prepare__() must return a mapping, not '{}'",
other.type_name()
))
.into()),
}
}
async fn invoke_metaclass_init(
state: &mut InterpreterState,
class_name: &str,
meta: &str,
tools: &Tools,
) -> Result<(), EvalError> {
let Some((_, method)) = lookup_method_or_classmethod(state, meta, "__init__") else {
return Ok(());
};
let class = state
.classes
.get(class_name)
.cloned()
.ok_or_else(|| EvalError::from(InterpreterError::name_not_defined(class_name)))?;
let mut ns = IndexMap::new();
for (k, v) in &class.class_attrs {
ns.insert(crate::value::ValueKey::String(k.as_str().into()), v.clone());
}
let bases_t = Value::Tuple(class.bases.iter().map(|b| Value::Class(b.clone())).collect());
let name_v = Value::String(class_name.into());
let ns_v = Value::Dict(crate::value::shared_dict(ns));
let cls_v = Value::Class(class_name.to_string());
let call = crate::eval::functions::CallArgs {
positional: &[name_v, bases_t, ns_v],
keyword: &IndexMap::new(),
};
let _ = call_method(state, &method, cls_v, call, tools).await?;
let _ = meta; Ok(())
}
async fn invoke_metaclass_new(
state: &mut InterpreterState,
class_name: &str,
meta: &str,
tools: &Tools,
) -> Result<(), EvalError> {
let Some((_, method)) = lookup_method_or_classmethod(state, meta, "__new__") else {
return Ok(());
};
let class = state
.classes
.get(class_name)
.cloned()
.ok_or_else(|| EvalError::from(InterpreterError::name_not_defined(class_name)))?;
let mut ns = IndexMap::new();
for (k, v) in &class.class_attrs {
ns.insert(crate::value::ValueKey::String(k.as_str().into()), v.clone());
}
let bases_t = Value::Tuple(class.bases.iter().map(|b| Value::Class(b.clone())).collect());
let name_v = Value::String(class_name.into());
let ns_v = Value::Dict(crate::value::shared_dict(ns));
let meta_v = Value::Class(meta.to_string());
let call = crate::eval::functions::CallArgs {
positional: &[name_v, bases_t, ns_v],
keyword: &IndexMap::new(),
};
let (returned, _) = call_method(state, &method, meta_v, call, tools).await?;
match returned {
Value::Class(n) => {
state.set_variable(class_name, Value::Class(n)).map_err(EvalError::Interpreter)?;
}
Value::None => {}
other => {
state.set_variable(class_name, other).map_err(EvalError::Interpreter)?;
}
}
Ok(())
}
fn parse_slots_attr(attr: Option<&Value>) -> (bool, Vec<String>) {
let Some(attr) = attr else {
return (false, Vec::new());
};
let names = match attr {
Value::String(s) => vec![s.to_string()],
Value::Tuple(items) => items
.iter()
.filter_map(|v| match v {
Value::String(s) => Some(s.to_string()),
_ => None,
})
.collect(),
Value::List(shared) => {
let guard = shared.lock();
guard
.iter()
.filter_map(|v| match v {
Value::String(s) => Some(s.to_string()),
_ => None,
})
.collect()
}
_ => return (false, Vec::new()),
};
let enforced = !names.iter().any(|n| n == "__dict__");
(enforced, names)
}