rustpython-vm 0.6.0

/*! Python `property` descriptor class.

*/
use super::PyType;
use crate::common::lock::PyRwLock;
use crate::object::PyAtomicRef;
use crate::{
    AsObject, Context, Py, PyObject, PyObjectRef, PyPayload, PyRef, PyResult, VirtualMachine,
    class::PyClassImpl,
    function::{ArgumentError, FromArgs, FuncArgs, Param, PySetterValue},
    types::{Constructor, GetDescriptor, Initializer},
};
use core::sync::atomic::{AtomicBool, Ordering};

struct SetNameArgs {
    _owner: PyObjectRef,
    name: PyObjectRef,
}

impl FromArgs for SetNameArgs {
    const PARAMS: Option<&'static [Param]> = Some(&[
        Param::positional_only("owner"),
        Param::positional_only("name"),
    ]);

    fn from_args(vm: &VirtualMachine, args: &mut FuncArgs) -> Result<Self, ArgumentError> {
        let [owner, name]: [PyObjectRef; 2] =
            core::mem::take(&mut args.args)
                .try_into()
                .map_err(|args: Vec<PyObjectRef>| {
                    ArgumentError::Exception(vm.new_type_error(format!(
                        "__set_name__() takes 2 positional arguments but {} were given",
                        args.len()
                    )))
                })?;
        Ok(Self {
            _owner: owner,
            name,
        })
    }
}

#[pyclass(module = false, name = "property", traverse)]
#[derive(Debug)]
pub struct PyProperty {
    #[pymember(name = "fget")]
    getter: PyAtomicRef<Option<PyObject>>,
    #[pymember(name = "fset")]
    setter: PyAtomicRef<Option<PyObject>>,
    #[pymember(name = "fdel")]
    deleter: PyAtomicRef<Option<PyObject>>,
    #[pymember(name = "__doc__", writable)]
    doc: PyAtomicRef<Option<PyObject>>,
    name: PyRwLock<Option<PyObjectRef>>,
    #[pytraverse(skip)]
    getter_doc: core::sync::atomic::AtomicBool,
}

impl PyPayload for PyProperty {
    #[inline]
    fn class(ctx: &Context) -> &'static Py<PyType> {
        ctx.types.property_type
    }
}

#[derive(FromArgs)]
pub struct PropertyArgs {
    #[pyarg(any, optional)]
    fget: Option<PyObjectRef>,
    #[pyarg(any, optional)]
    fset: Option<PyObjectRef>,
    #[pyarg(any, optional)]
    fdel: Option<PyObjectRef>,
    #[pyarg(any, optional)]
    doc: Option<PyObjectRef>,
}

impl GetDescriptor for PyProperty {
    fn descr_get(
        zelf_obj: &PyObject,
        obj: Option<&PyObject>,
        _cls: Option<&PyObject>,
        vm: &VirtualMachine,
    ) -> PyResult {
        let (zelf, obj) = Self::_unwrap(zelf_obj, obj, vm)?;
        if vm.is_none(obj) {
            return Ok(zelf_obj.to_owned());
        }

        // Clone and release lock before calling Python code to prevent deadlock
        let value = zelf.getter.load_owned();
        if let Some(getter) = value {
            getter.call((obj.to_owned(),), vm)
        } else {
            let error_msg = zelf.format_property_error(obj, "getter", vm)?;
            Err(vm.new_attribute_error(error_msg))
        }
    }
}

impl PyProperty {
    // Helper method to get property name
    // Returns the name if available, None if not found, or propagates errors
    fn get_property_name(&self, vm: &VirtualMachine) -> PyResult<Option<PyObjectRef>> {
        // First check if name was set via __set_name__
        let value = self.name.read().clone();
        if let Some(name) = value {
            return Ok(Some(name));
        }

        // Clone and release lock before calling Python code to prevent deadlock
        let Some(getter) = self.getter.load_owned() else {
            return Ok(None);
        };

        match getter.get_attr("__name__", vm) {
            Ok(name) => Ok(Some(name)),
            Err(e) => {
                // If it's an AttributeError from the getter, return None
                // Otherwise, propagate the original exception (e.g., RuntimeError)
                if e.class().is(vm.ctx.exceptions.attribute_error) {
                    Ok(None)
                } else {
                    Err(e)
                }
            }
        }
    }

    pub(crate) fn get_fget(&self) -> Option<PyObjectRef> {
        self.getter.load_owned()
    }

    fn get_fset(&self) -> Option<PyObjectRef> {
        self.setter.load_owned()
    }

    fn get_fdel(&self) -> Option<PyObjectRef> {
        self.deleter.load_owned()
    }

    fn doc_getter(&self) -> Option<PyObjectRef> {
        self.doc.load_owned()
    }

    // Python builder functions

    // Helper method to create a new property with updated attributes
    fn clone_property_with(
        zelf: &Py<Self>,
        new_getter: Option<PyObjectRef>,
        new_setter: Option<PyObjectRef>,
        new_deleter: Option<PyObjectRef>,
        vm: &VirtualMachine,
    ) -> PyResult<PyRef<Self>> {
        // Determine doc based on getter_doc flag and whether we're updating the getter
        let doc = if zelf.getter_doc.load(Ordering::Relaxed) && new_getter.is_some() {
            // If the original property uses getter doc and we have a new getter,
            // pass Py_None to let __init__ get the doc from the new getter
            Some(vm.ctx.none())
        } else if zelf.getter_doc.load(Ordering::Relaxed) {
            // If original used getter_doc but we're not changing the getter,
            // pass None to let init get doc from existing getter
            Some(vm.ctx.none())
        } else {
            // Otherwise use the existing doc
            zelf.doc_getter()
        };

        // Create property args with updated values
        let args = PropertyArgs {
            fget: new_getter.or_else(|| zelf.get_fget()),
            fset: new_setter.or_else(|| zelf.get_fset()),
            fdel: new_deleter.or_else(|| zelf.get_fdel()),
            doc,
        };

        // Create new property using py_new and init
        let new_prop = Self::slot_new(zelf.class().to_owned(), FuncArgs::default(), vm)?;
        let new_prop_ref = new_prop.downcast::<Self>().unwrap();
        Self::init(&new_prop_ref, args, vm)?;

        // Copy the name if it exists
        let value = zelf.name.read().clone();
        if let Some(name) = value {
            *new_prop_ref.name.write() = Some(name);
        }

        Ok(new_prop_ref)
    }

    // Helper method to format property error messages
    #[cold]
    fn format_property_error(
        &self,
        obj: &PyObject,
        error_type: &str,
        vm: &VirtualMachine,
    ) -> PyResult<String> {
        let prop_name = self.get_property_name(vm)?;
        let obj_type = obj.class();
        let qualname = obj_type.__qualname__(vm);

        match prop_name {
            Some(name) => Ok(format!(
                "property {} of {} object has no {}",
                name.repr(vm)?,
                qualname.repr(vm)?,
                error_type
            )),
            None => Ok(format!(
                "property of {} object has no {}",
                qualname.repr(vm)?,
                error_type
            )),
        }
    }
}

#[pyclass(
    with(Constructor, Initializer, GetDescriptor),
    flags(BASETYPE, HAS_WEAKREF)
)]
impl Py<PyProperty> {
    // Descriptor methods

    #[pyslot]
    fn descr_set(
        zelf: &PyObject,
        obj: PyObjectRef,
        value: PySetterValue,
        vm: &VirtualMachine,
    ) -> PyResult<()> {
        let zelf = zelf.try_to_ref::<PyProperty>(vm)?;
        match value {
            PySetterValue::Assign(value) => {
                // Clone and release lock before calling Python code to prevent deadlock
                let set = zelf.setter.load_owned();
                if let Some(setter) = set {
                    setter.call((obj, value), vm).map(drop)
                } else {
                    let error_msg = zelf.format_property_error(&obj, "setter", vm)?;
                    Err(vm.new_attribute_error(error_msg))
                }
            }
            PySetterValue::Delete => {
                // Clone and release lock before calling Python code to prevent deadlock
                let del = zelf.deleter.load_owned();
                if let Some(deleter) = del {
                    deleter.call((obj,), vm).map(drop)
                } else {
                    let error_msg = zelf.format_property_error(&obj, "deleter", vm)?;
                    Err(vm.new_attribute_error(error_msg))
                }
            }
        }
    }

    #[pygetset(name = "__name__")]
    fn name_getter(&self, vm: &VirtualMachine) -> PyResult {
        match self.get_property_name(vm)? {
            Some(name) => Ok(name),
            None => Err(vm.new_attribute_error("'property' object has no attribute '__name__'")),
        }
    }

    #[pygetset(name = "__name__", setter)]
    fn name_setter(&self, value: PyObjectRef) {
        *self.name.write() = Some(value);
    }

    #[pymethod]
    fn __set_name__(&self, SetNameArgs { name, .. }: SetNameArgs) {
        *self.name.write() = Some(name);
    }

    #[pymethod]
    fn getter(
        zelf: PyRef<PyProperty>,
        object: Option<PyObjectRef>,
        vm: &VirtualMachine,
    ) -> PyResult<PyRef<PyProperty>> {
        PyProperty::clone_property_with(&zelf, object, None, None, vm)
    }

    #[pymethod]
    fn setter(
        zelf: PyRef<PyProperty>,
        object: Option<PyObjectRef>,
        vm: &VirtualMachine,
    ) -> PyResult<PyRef<PyProperty>> {
        PyProperty::clone_property_with(&zelf, None, object, None, vm)
    }

    #[pymethod]
    fn deleter(
        zelf: PyRef<PyProperty>,
        object: Option<PyObjectRef>,
        vm: &VirtualMachine,
    ) -> PyResult<PyRef<PyProperty>> {
        PyProperty::clone_property_with(&zelf, None, None, object, vm)
    }

    #[pygetset]
    fn __isabstractmethod__(&self, vm: &VirtualMachine) -> PyResult {
        // Helper to check if a method is abstract
        let is_abstract = |method: &PyObject| -> PyResult<bool> {
            match method.get_attr("__isabstractmethod__", vm) {
                Ok(isabstract) => isabstract.try_to_bool(vm),
                Err(_) => Ok(false),
            }
        };

        // Clone and release lock before calling Python code to prevent deadlock
        // Check getter
        if let Some(getter) = self.getter.load_owned()
            && is_abstract(&getter)?
        {
            return Ok(vm.ctx.new_bool(true).into());
        }

        // Check setter
        if let Some(setter) = self.setter.load_owned()
            && is_abstract(&setter)?
        {
            return Ok(vm.ctx.new_bool(true).into());
        }

        // Check deleter
        if let Some(deleter) = self.deleter.load_owned()
            && is_abstract(&deleter)?
        {
            return Ok(vm.ctx.new_bool(true).into());
        }

        Ok(vm.ctx.new_bool(false).into())
    }

    #[pygetset(setter)]
    fn set___isabstractmethod__(&self, value: PyObjectRef, vm: &VirtualMachine) -> PyResult<()> {
        // Clone and release lock before calling Python code to prevent deadlock
        let maybe_getter = self.getter.load_owned();
        if let Some(getter) = maybe_getter {
            getter.set_attr("__isabstractmethod__", value, vm)?;
        }
        Ok(())
    }
}

impl Constructor for PyProperty {
    type Args = FuncArgs;

    fn py_new(_cls: &Py<PyType>, _args: FuncArgs, _vm: &VirtualMachine) -> PyResult<Self> {
        Ok(Self {
            getter: PyAtomicRef::new_empty(),
            setter: PyAtomicRef::new_empty(),
            deleter: PyAtomicRef::new_empty(),
            doc: PyAtomicRef::new_empty(),
            name: PyRwLock::new(None),
            getter_doc: AtomicBool::new(false),
        })
    }
}

impl Initializer for PyProperty {
    type Args = PropertyArgs;

    fn init(zelf: &Py<Self>, args: Self::Args, vm: &VirtualMachine) -> PyResult<()> {
        // Set doc and getter_doc flag
        let mut getter_doc = false;

        // Helper to get doc from getter
        let get_getter_doc = |fget: &PyObject| -> Option<PyObjectRef> {
            fget.get_attr("__doc__", vm)
                .ok()
                .filter(|doc| !vm.is_none(doc))
        };

        let doc = match args.doc {
            Some(doc) if !vm.is_none(&doc) => Some(doc),
            _ => {
                // No explicit doc or doc is None, try to get from getter
                args.fget.as_ref().and_then(|fget| {
                    get_getter_doc(fget).inspect(|_| {
                        getter_doc = true;
                    })
                })
            }
        };

        // Check if this is a property subclass
        let is_exact_property = zelf.class().is(vm.ctx.types.property_type);

        if is_exact_property {
            // For exact property type, store doc in the field
            zelf.doc.store(doc);
        } else {
            // For property subclass, set __doc__ as an attribute
            let doc_to_set = doc.unwrap_or_else(|| vm.ctx.none());
            match zelf.as_object().set_attr("__doc__", doc_to_set, vm) {
                Ok(()) => {}
                Err(e) if !getter_doc && e.class().is(vm.ctx.exceptions.attribute_error) => {
                    // Silently ignore AttributeError for backwards compatibility
                    // (only when not using getter_doc)
                }
                Err(e) => return Err(e),
            }
        }

        zelf.getter.store(args.fget);
        zelf.setter.store(args.fset);
        zelf.deleter.store(args.fdel);
        zelf.getter_doc.store(getter_doc, Ordering::Relaxed);

        Ok(())
    }
}

pub(crate) fn init(context: &'static Context) {
    PyProperty::extend_class(context, context.types.property_type);
}