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, 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;
for kw in &node.keywords {
let key = kw.arg.as_ref().map(|a| a.as_str());
if key != Some("metaclass") {
return Err(InterpreterError::TypeError(format!(
"class keyword argument '{}' is not supported",
key.unwrap_or("<unknown>")
))
.into());
}
let val = eval_expr(state, &kw.value, tools).await?;
match val {
Value::BuiltinName(n) | Value::Type(n) if n == "type" => {}
Value::Class(n) => metaclass_name = Some(n),
other => {
return Err(InterpreterError::TypeError(format!(
"metaclass must be a type, not '{}'",
other.type_name()
))
.into());
}
}
}
let mut bases: Vec<String> = Vec::new();
let mut enum_kind: Option<crate::value::EnumKind> = None;
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.Flag" => enum_kind = Some(crate::value::EnumKind::Plain),
"enum.IntEnum" | "enum.IntFlag" => {
enum_kind = Some(crate::value::EnumKind::Int);
}
"enum.StrEnum" => enum_kind = Some(crate::value::EnumKind::Str),
_ => {}
}
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(_)));
if !state.classes.contains_key(&base_name) && !is_exception_base {
return Err(InterpreterError::name_not_defined(&base_name).into());
}
bases.push(base_name);
}
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();
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);
}
}
}
}
for stmt in &node.body {
match stmt {
Stmt::FunctionDef(method) => {
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();
classify_decorated_method(
state,
&method.decorator_list,
class_name,
method_name,
params,
method.body.clone(),
source,
&mut methods,
&mut properties,
&mut static_methods,
&mut class_methods,
)?;
}
Stmt::Assign(assign) => {
if let [Expr::Name(target)] = assign.targets.as_slice() {
let value = eval_expr(state, &assign.value, tools).await?;
let attr_name = target.id.as_str().to_string();
let wrapped = wrap_enum_member(enum_kind, class_name, &attr_name, value);
class_attrs.insert(attr_name, wrapped);
}
}
Stmt::AnnAssign(ann) => {
if let Expr::Name(target) = ann.target.as_ref() {
let attr_name = target.id.as_str().to_string();
if !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);
class_attrs.insert(attr_name, wrapped);
}
}
}
_ => {}
}
}
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());
}
}
}
}
}
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.annotations = annotations;
cv.slots = slots;
cv.slot_names = slot_names;
cv
});
state
.set_variable(class_name, Value::Class(class_name.to_string()))
.map_err(EvalError::Interpreter)?;
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, 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)
}
#[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 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);
state.function_bodies.insert(key.clone(), Arc::new(body));
FunctionDef {
name: key,
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,
}
};
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 });
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::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, 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(())
}
async fn invoke_init_subclass(
state: &mut InterpreterState,
class_name: &str,
bases: &[String],
tools: &Tools,
) -> Result<(), EvalError> {
let empty_kwargs = IndexMap::new();
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: &empty_kwargs };
let (_ret, _self) = call_method(state, &def, new_cls.clone(), call, tools).await?;
}
Ok(())
}
fn wrap_enum_member(
enum_kind: Option<crate::value::EnumKind>,
class_name: &str,
member_name: &str,
value: Value,
) -> 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;
}
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)
}
pub async fn instantiate(
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.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.iter().any(|b| {
b == "Exception"
|| b == "BaseException"
|| crate::eval::functions::is_exception_type_name(b)
});
if is_exception_subclass {
let message = match args.len() {
0 => String::new(),
1 => format!("{}", args[0]),
_ => args.iter().map(|v| format!("{v}")).collect::<Vec<_>>().join(", "),
};
return Ok(Value::Exception(
crate::value::ExceptionValue::new(class_name.to_string(), message)
.with_args(args.to_vec()),
));
}
}
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();
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());
}
}
};
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);
}
Ok(Value::Instance(InstanceValue {
class_name: class_name.to_string(),
fields: crate::value::shared_fields(instance_fields),
}))
}
fn empty_for_builtin_factory(name: &str) -> EvalResult {
match name {
"list" => Ok(Value::List(shared_list(Vec::new()))),
"dict" => Ok(Value::Dict(IndexMap::new())),
"set" => Ok(Value::Set(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()
);
};
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) {
return call_method(state, &def, Value::Class(class_name.clone()), call, tools).await;
}
let method = lookup_method_in_mro(state, &class_name, method_name);
let Some((_defining_class, def)) = method else {
return Err(InterpreterError::AttributeError(format!(
"'{class_name}' object has no attribute '{method_name}'"
))
.into());
};
call_method(state, &def, instance, call, tools).await
}
fn apply_method_scope(
state: &mut InterpreterState,
local_scope: &std::collections::HashMap<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> {
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 let Value::Instance(_) = &self_value {
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(&method.params, &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) => 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,
tools: &Tools,
) -> EvalResult {
let call = CallArgs { positional: &[], keyword: &IndexMap::new() };
let (returned, _self) = call_method(state, getter, instance, call, tools).await?;
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__" || attr == "__qualname__" {
return Ok(Value::String(class_name.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);
}
Err(InterpreterError::AttributeError(format!(
"type object '{class_name}' has no attribute '{attr}'"
))
.into())
}
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_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(),
))
})?;
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(),
))
})?;
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))
}
_ => 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))
}
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 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 {
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)),
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(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(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()),
};
(true, names)
}