pybevy 0.2.1

PyBevy: A Python Real-Time Engine Built on Bevy
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128

use std::{collections::HashMap, sync::Arc};

use bevy::ecs::component::ComponentId;
use pyo3::{exceptions::PyRuntimeError, ffi::PyTypeObject, prelude::*};

use crate::ecs::{
    helpers::validity_guard::ValidityFlag,
    query::{query_param::PyQueryParam, query_runtime::PyQueryIter},
};

/// Runtime wrapper for Single<T> queries that enforces exactly one entity matches.
///
/// This wraps a PyQueryIter but validates that exactly one entity exists matching
/// the query filter before returning the result.
#[pyclass(name = "SingleQuery")]
pub struct PySingleQuery {
    /// The underlying query iterator
    query_iter: Py<PyQueryIter>,
    /// Whether we've already returned the single result
    returned: bool,
    /// Cached single item for __getattr__/__setattr__ delegation (Deref emulation)
    cached_item: Option<Py<PyAny>>,
}

impl PySingleQuery {
    /// Creates a new Single query wrapper
    ///
    /// # Safety
    /// The world pointer must remain valid for the lifetime of this object
    pub unsafe fn new(
        param: Arc<PyQueryParam>,
        world: &mut bevy::prelude::World,
        custom_component_ids: Arc<HashMap<*const PyTypeObject, ComponentId>>,
        validity: ValidityFlag,
    ) -> Self {
        // SAFETY: Caller guarantees world pointer is valid during system execution
        let query_iter = unsafe { PyQueryIter::new(param, world, custom_component_ids, validity) };

        Python::attach(|py| {
            let query_iter_py = Py::new(py, query_iter).expect("Failed to create PyQueryIter");

            PySingleQuery {
                query_iter: query_iter_py,
                returned: false,
                cached_item: None,
            }
        })
    }
}

#[pymethods]
impl PySingleQuery {
    /// Makes this object iterable (but will only yield one item)
    fn __iter__(slf: PyRef<'_, Self>) -> PyRef<'_, Self> {
        slf
    }

    /// Returns the single query result, panicking if zero or multiple entities match
    fn __next__(&mut self, py: Python) -> PyResult<Py<PyAny>> {
        if self.returned {
            return Err(pyo3::exceptions::PyStopIteration::new_err(""));
        }

        // Get the first result
        let first = self.query_iter.call_method0(py, "__next__")?;

        // Check if there's a second result (which would be an error)
        match self.query_iter.call_method0(py, "__next__") {
            Ok(_) => {
                // There's a second result - this is an error
                Err(PyRuntimeError::new_err(
                    "Single<T> query matched multiple entities. Expected exactly one entity.",
                ))
            }
            Err(e) if e.is_instance_of::<pyo3::exceptions::PyStopIteration>(py) => {
                // No second result - this is correct, return the first
                self.returned = true;
                Ok(first)
            }
            Err(e) => {
                // Some other error
                Err(e)
            }
        }
    }

    /// Delegate attribute access to the single result, matching Bevy's Deref behavior.
    /// In Bevy Rust, Single<T> implements Deref<Target=T>, so `single.field` works directly.
    fn __getattr__(&mut self, py: Python, name: &str) -> PyResult<Py<PyAny>> {
        let item = self.get_or_fetch_item(py)?;
        item.getattr(py, name)
    }

    /// Delegate attribute setting to the single result (Bevy's DerefMut).
    fn __setattr__(&mut self, py: Python, name: &str, value: Py<PyAny>) -> PyResult<()> {
        let item = self.get_or_fetch_item(py)?;
        item.setattr(py, name, value)
    }
}

impl PySingleQuery {
    /// Get the single item, fetching it on first access and caching it.
    fn get_or_fetch_item(&mut self, py: Python) -> PyResult<Py<PyAny>> {
        if let Some(ref item) = self.cached_item {
            return Ok(item.clone_ref(py));
        }

        // Get the first result
        let first = self.query_iter.call_method0(py, "__next__")?;

        // Verify exactly one entity matches
        match self.query_iter.call_method0(py, "__next__") {
            Ok(_) => {
                return Err(PyRuntimeError::new_err(
                    "Single<T> query matched multiple entities. Expected exactly one entity.",
                ));
            }
            Err(e) if e.is_instance_of::<pyo3::exceptions::PyStopIteration>(py) => {
                // Correct - exactly one entity
            }
            Err(e) => return Err(e),
        }

        self.returned = true;
        self.cached_item = Some(first.clone_ref(py));
        Ok(first)
    }
}