use crate::common::lock::LazyLock;
use super::type_;
use crate::{
AsObject, Context, Py, PyObject, PyObjectRef, PyPayload, PyRef, PyResult, TryFromObject,
VirtualMachine, atomic_func,
builtins::{PyList, PyStr, PyTuple, PyTupleRef, PyType, PyTypeRef},
class::{PyClassDef, PyClassImpl},
common::hash,
convert::ToPyObject,
function::{FuncArgs, PyComparisonValue},
protocol::{PyMappingMethods, PyNumberMethods},
types::{
AsMapping, AsNumber, Callable, Comparable, Constructor, GetAttr, Hashable, IterNext,
Iterable, PyComparisonOp, Representable,
},
};
use alloc::fmt;
static ATTR_EXCEPTIONS: [&str; 9] = [
"__class__",
"__origin__",
"__args__",
"__unpacked__",
"__parameters__",
"__typing_unpacked_tuple_args__",
"__mro_entries__",
"__reduce_ex__", "__reduce__",
];
static ATTR_BLOCKED: [&str; 3] = ["__bases__", "__copy__", "__deepcopy__"];
#[pyclass(module = "types", name = "GenericAlias")]
pub struct PyGenericAlias {
#[pymember(name = "__origin__")]
origin: PyObjectRef,
#[pymember(name = "__args__")]
args: PyTupleRef,
parameters: PyTupleRef,
#[pymember(name = "__unpacked__")]
starred: bool, }
impl fmt::Debug for PyGenericAlias {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str("GenericAlias")
}
}
impl PyPayload for PyGenericAlias {
#[inline]
fn class(ctx: &Context) -> &'static Py<PyType> {
ctx.types.generic_alias_type
}
}
#[derive(FromArgs)]
pub struct GenericAliasArgs {
#[pyarg(positional)]
origin: PyObjectRef,
#[pyarg(positional)]
args: PyObjectRef,
}
impl Constructor for PyGenericAlias {
type Args = GenericAliasArgs;
fn slot_new(cls: PyTypeRef, args: FuncArgs, vm: &VirtualMachine) -> PyResult {
if !args.kwargs.is_empty() {
return Err(vm.new_type_error("GenericAlias() takes no keyword arguments"));
}
let GenericAliasArgs {
origin,
args: arguments,
} = args.bind_for(vm, Self::NAME)?;
let arguments = if let Ok(tuple) = arguments.try_to_ref::<PyTuple>(vm) {
tuple.to_owned()
} else {
PyTuple::new_ref(vec![arguments], &vm.ctx)
};
let payload = Self::new(origin, arguments, false, vm)?;
payload.into_ref_with_type(vm, cls).map(Into::into)
}
fn py_new(_cls: &Py<PyType>, args: Self::Args, vm: &VirtualMachine) -> PyResult<Self> {
let GenericAliasArgs {
origin,
args: arguments,
} = args;
let args = if let Ok(tuple) = arguments.try_to_ref::<PyTuple>(vm) {
tuple.to_owned()
} else {
PyTuple::new_ref(vec![arguments], &vm.ctx)
};
Self::new(origin, args, false, vm)
}
}
impl PyGenericAlias {
pub fn new(
origin: impl Into<PyObjectRef>,
args: PyTupleRef,
starred: bool,
vm: &VirtualMachine,
) -> PyResult<Self> {
let parameters = make_parameters(&args, vm)?;
Ok(Self {
origin: origin.into(),
args,
parameters,
starred,
})
}
pub fn from_args(
origin: impl Into<PyObjectRef>,
args: PyObjectRef,
vm: &VirtualMachine,
) -> PyResult<Self> {
let args = if let Ok(tuple) = args.try_to_ref::<PyTuple>(vm) {
tuple.to_owned()
} else {
PyTuple::new_ref(vec![args], &vm.ctx)
};
Self::new(origin, args, false, vm)
}
fn repr(&self, vm: &VirtualMachine) -> PyResult<String> {
fn repr_item(obj: &PyObject, vm: &VirtualMachine) -> PyResult<String> {
if obj.is(&vm.ctx.ellipsis) {
return Ok("...".to_string());
}
if vm
.get_attribute_opt(obj, identifier!(vm, __origin__))?
.is_some()
&& vm
.get_attribute_opt(obj, identifier!(vm, __args__))?
.is_some()
{
return Ok(obj.repr(vm)?.to_string());
}
match (
vm.get_attribute_opt(obj, identifier!(vm, __qualname__))?
.and_then(|o| o.downcast_ref::<PyStr>().map(|n| n.to_string())),
vm.get_attribute_opt(obj, identifier!(vm, __module__))?
.and_then(|o| o.downcast_ref::<PyStr>().map(|m| m.to_string())),
) {
(None, _) | (_, None) => Ok(obj.repr(vm)?.to_string()),
(Some(qualname), Some(module)) => Ok(if module == "builtins" {
qualname
} else {
format!("{module}.{qualname}")
}),
}
}
fn repr_arg(obj: &PyObject, vm: &VirtualMachine) -> PyResult<String> {
if obj.class().is(vm.ctx.types.list_type) {
let list = obj.downcast_ref::<crate::builtins::PyList>().unwrap();
let len = list.borrow_vec().len();
let mut parts = Vec::with_capacity(len);
for i in 0..len {
let item = list
.borrow_vec()
.get(i)
.cloned()
.ok_or_else(|| vm.new_index_error("list index out of range"))?;
parts.push(repr_item(&item, vm)?);
}
Ok(format!("[{}]", parts.join(", ")))
} else {
repr_item(obj, vm)
}
}
let repr_str = format!(
"{}[{}]",
repr_item(&self.origin, vm)?,
if self.args.as_slice().is_empty() {
"()".to_owned()
} else {
self.args
.as_slice()
.iter()
.map(|o| repr_arg(o, vm))
.collect::<PyResult<Vec<_>>>()?
.join(", ")
}
);
Ok(if self.starred {
format!("*{repr_str}")
} else {
repr_str
})
}
fn __getitem__(zelf: &Py<Self>, needle: PyObjectRef, vm: &VirtualMachine) -> PyResult {
let new_args = subs_parameters(zelf.as_object(), &zelf.args, &zelf.parameters, needle, vm)?;
Ok(Self::new(zelf.origin.clone(), new_args, false, vm)?.into_pyobject(vm))
}
fn __ror__(zelf: PyObjectRef, other: PyObjectRef, vm: &VirtualMachine) -> PyResult {
type_::or_(other, zelf, vm)
}
fn __or__(zelf: PyObjectRef, other: PyObjectRef, vm: &VirtualMachine) -> PyResult {
type_::or_(zelf, other, vm)
}
}
#[pyclass(
with(
AsNumber,
AsMapping,
Callable,
Comparable,
Constructor,
GetAttr,
Hashable,
Iterable,
Representable
),
flags(BASETYPE, HAS_WEAKREF)
)]
impl Py<PyGenericAlias> {
#[pygetset]
fn __parameters__(&self) -> PyObjectRef {
self.parameters.clone().into()
}
#[pygetset]
fn __typing_unpacked_tuple_args__(&self, vm: &VirtualMachine) -> PyObjectRef {
if self.starred && self.origin.is(vm.ctx.types.tuple_type.as_object()) {
self.args.clone().into()
} else {
vm.ctx.none()
}
}
#[pymethod]
fn __dir__(&self, vm: &VirtualMachine) -> PyResult<PyList> {
let dir = vm.dir(Some(self.origin.clone()))?;
for exc in &ATTR_EXCEPTIONS {
let exc_obj = (*exc).to_pyobject(vm);
if !dir.__contains__(&exc_obj, vm)? {
dir.append((*exc).to_pyobject(vm));
}
}
Ok(dir)
}
#[pymethod]
fn __reduce__(zelf: &Self, vm: &VirtualMachine) -> PyResult<PyTupleRef> {
if zelf.starred {
let next_fn = vm.builtins.get_attr("next", vm)?;
let non_starred =
PyGenericAlias::new(zelf.origin.clone(), zelf.args.clone(), false, vm)?;
let iter_obj = PyGenericAliasIterator {
obj: crate::common::lock::PyMutex::new(Some(non_starred.into_pyobject(vm))),
}
.into_pyobject(vm);
Ok(PyTuple::new_ref(
vec![next_fn, PyTuple::new_ref(vec![iter_obj], &vm.ctx).into()],
&vm.ctx,
))
} else {
Ok(PyTuple::new_ref(
vec![
vm.ctx.types.generic_alias_type.to_owned().into(),
PyTuple::new_ref(vec![zelf.origin.clone(), zelf.args.clone().into()], &vm.ctx)
.into(),
],
&vm.ctx,
))
}
}
#[pymethod]
fn __mro_entries__(&self, _object: PyObjectRef, vm: &VirtualMachine) -> PyTupleRef {
PyTuple::new_ref(vec![self.origin.clone()], &vm.ctx)
}
#[pymethod]
fn __instancecheck__(
_zelf: PyRef<PyGenericAlias>,
_object: PyObjectRef,
vm: &VirtualMachine,
) -> PyResult {
Err(vm.new_type_error("isinstance() argument 2 cannot be a parameterized generic"))
}
#[pymethod]
fn __subclasscheck__(
_zelf: PyRef<PyGenericAlias>,
_object: PyObjectRef,
vm: &VirtualMachine,
) -> PyResult {
Err(vm.new_type_error("issubclass() argument 2 cannot be a parameterized generic"))
}
}
pub(crate) fn make_parameters(args: &Py<PyTuple>, vm: &VirtualMachine) -> PyResult<PyTupleRef> {
make_parameters_from_slice(args.as_slice(), vm)
}
fn make_parameters_from_slice(args: &[PyObjectRef], vm: &VirtualMachine) -> PyResult<PyTupleRef> {
let mut parameters: Vec<PyObjectRef> = Vec::with_capacity(args.len());
for arg in args {
if arg.class().is(vm.ctx.types.type_type) {
continue;
}
if arg.get_attr(identifier!(vm, __typing_subst__), vm).is_ok() {
if tuple_index(¶meters, arg).is_none() {
parameters.push(arg.clone());
}
} else if let Ok(subparams) = arg.get_attr(identifier!(vm, __parameters__), vm)
&& let Ok(sub_params) = subparams.try_to_ref::<PyTuple>(vm)
{
for sub_param in sub_params {
if tuple_index(¶meters, sub_param).is_none() {
parameters.push(sub_param.clone());
}
}
} else if arg.try_to_ref::<PyTuple>(vm).is_ok() || arg.try_to_ref::<PyList>(vm).is_ok() {
let items: Vec<PyObjectRef> = if let Ok(t) = arg.try_to_ref::<PyTuple>(vm) {
t.as_slice().to_vec()
} else {
let list = arg.downcast_ref::<PyList>().unwrap();
list.borrow_vec().to_vec()
};
let sub = vm.with_recursion("while computing __parameters__", || {
make_parameters_from_slice(&items, vm)
})?;
for sub_param in sub.as_slice() {
if tuple_index(¶meters, sub_param).is_none() {
parameters.push(sub_param.clone());
}
}
}
}
Ok(PyTuple::new_ref(parameters, &vm.ctx))
}
#[inline]
fn tuple_index(vec: &[PyObjectRef], item: &PyObject) -> Option<usize> {
vec.iter().position(|element| element.is(item))
}
fn is_unpacked_typevartuple(arg: &PyObject, vm: &VirtualMachine) -> PyResult<bool> {
if arg.class().is(vm.ctx.types.type_type) {
return Ok(false);
}
if let Ok(attr) = arg.get_attr(identifier!(vm, __typing_is_unpacked_typevartuple__), vm) {
attr.try_to_bool(vm)
} else {
Ok(false)
}
}
fn subs_tvars(
obj: PyObjectRef,
params: &Py<PyTuple>,
arg_items: &[PyObjectRef],
vm: &VirtualMachine,
) -> PyResult {
obj.get_attr(identifier!(vm, __parameters__), vm)
.ok()
.and_then(|sub_params| {
PyTupleRef::try_from_object(vm, sub_params)
.ok()
.filter(|sub_params| !sub_params.as_slice().is_empty())
.map(|sub_params| {
let mut sub_args = Vec::new();
for arg in sub_params.as_slice() {
if let Some(idx) = tuple_index(params.as_slice(), arg) {
let param = ¶ms.as_slice()[idx];
let substituted_arg = &arg_items[idx];
if param.class().slots().iter.load().is_some()
&& substituted_arg.try_to_ref::<PyTuple>(vm).is_ok()
{
if let Ok(tuple) = substituted_arg.try_to_ref::<PyTuple>(vm) {
for elem in tuple {
sub_args.push(elem.clone());
}
continue;
}
}
sub_args.push(substituted_arg.clone());
} else {
sub_args.push(arg.clone());
}
}
let sub_args: PyObjectRef = PyTuple::new_ref(sub_args, &vm.ctx).into();
obj.get_item(&*sub_args, vm)
})
})
.unwrap_or(Ok(obj))
}
fn unpack_args(item: PyObjectRef, vm: &VirtualMachine) -> PyTupleRef {
let mut new_args = Vec::new();
let arg_items = if let Ok(tuple) = item.try_to_ref::<PyTuple>(vm) {
tuple.as_slice().to_vec()
} else {
vec![item]
};
for item in arg_items {
if item.class().is(vm.ctx.types.type_type) {
new_args.push(item);
continue;
}
if let Ok(sub_args) = item.get_attr(identifier!(vm, __typing_unpacked_tuple_args__), vm)
&& !sub_args.is(&vm.ctx.none)
&& let Ok(tuple) = sub_args.try_to_ref::<PyTuple>(vm)
{
let has_ellipsis_at_end = tuple
.as_slice()
.last()
.is_some_and(|item| item.is(&vm.ctx.ellipsis));
if !has_ellipsis_at_end {
for arg in tuple {
new_args.push(arg.clone());
}
continue;
}
}
new_args.push(item);
}
PyTuple::new_ref(new_args, &vm.ctx)
}
pub(crate) fn subs_parameters(
alias: &PyObject, args: &Py<PyTuple>,
parameters: &Py<PyTuple>,
item: PyObjectRef,
vm: &VirtualMachine,
) -> PyResult<PyTupleRef> {
let n_params = parameters.as_slice().len();
if n_params == 0 {
return Err(vm.new_type_error(format!("{} is not a generic class", alias.repr(vm)?)));
}
let mut item: PyObjectRef = unpack_args(item, vm).into();
for param in parameters {
if let Ok(prepare) = param.get_attr(identifier!(vm, __typing_prepare_subst__), vm)
&& !prepare.is(&vm.ctx.none)
{
item = if item.try_to_ref::<PyTuple>(vm).is_ok() {
prepare.call((alias.to_owned(), item.clone()), vm)?
} else {
let tuple_args = PyTuple::new_ref(vec![item.clone()], &vm.ctx);
prepare.call((alias.to_owned(), tuple_args.to_pyobject(vm)), vm)?
};
}
}
let arg_items = if let Ok(tuple) = item.try_to_ref::<PyTuple>(vm) {
tuple.as_slice().to_vec()
} else {
vec![item.clone()]
};
let n_items = arg_items.len();
if n_items != n_params {
return Err(vm.new_type_error(format!(
"Too {} arguments for {}; actual {}, expected {}",
if n_items > n_params { "many" } else { "few" },
alias.repr(vm)?,
n_items,
n_params
)));
}
let mut new_args = Vec::new();
for arg in args {
if arg.class().is(vm.ctx.types.type_type) {
new_args.push(arg.clone());
continue;
}
let is_list = arg.try_to_ref::<PyList>(vm).is_ok();
if arg.try_to_ref::<PyTuple>(vm).is_ok() || is_list {
let sub_items: Vec<PyObjectRef> = if let Ok(t) = arg.try_to_ref::<PyTuple>(vm) {
t.as_slice().to_vec()
} else {
arg.downcast_ref::<PyList>().unwrap().borrow_vec().to_vec()
};
let sub_tuple = PyTuple::new_ref(sub_items, &vm.ctx);
let sub_result = subs_parameters(alias, &sub_tuple, parameters, item.clone(), vm)?;
let substituted: PyObjectRef = if is_list {
PyList::from(sub_result.as_slice().to_vec())
.into_ref(&vm.ctx)
.into()
} else {
sub_result.into()
};
new_args.push(substituted);
continue;
}
let unpack = is_unpacked_typevartuple(arg, vm)?;
let substituted_arg = if let Ok(subst) = arg.get_attr(identifier!(vm, __typing_subst__), vm)
{
if let Some(iparam) = tuple_index(parameters.as_slice(), arg) {
subst.call((arg_items[iparam].clone(),), vm)?
} else {
subs_tvars(arg.clone(), parameters, &arg_items, vm)?
}
} else {
subs_tvars(arg.clone(), parameters, &arg_items, vm)?
};
if unpack {
let tuple = match substituted_arg.try_to_ref::<PyTuple>(vm) {
Ok(tuple) => tuple,
Err(_) => {
let arg_type = arg.class().fully_qualified_name(vm)?;
let substituted_type = substituted_arg.class().fully_qualified_name(vm)?;
return Err(vm.new_type_error(format!(
"expected __typing_subst__ of {arg_type} objects to return a tuple, not {substituted_type}",
)));
}
};
for elem in tuple {
new_args.push(elem.clone());
}
} else {
new_args.push(substituted_arg);
}
}
Ok(PyTuple::new_ref(new_args, &vm.ctx))
}
impl AsMapping for PyGenericAlias {
fn as_mapping() -> &'static PyMappingMethods {
static AS_MAPPING: LazyLock<PyMappingMethods> = LazyLock::new(|| PyMappingMethods {
subscript: atomic_func!(|mapping, needle, vm| {
let zelf = PyGenericAlias::mapping_downcast(mapping);
PyGenericAlias::__getitem__(zelf, needle.to_owned(), vm)
}),
..PyMappingMethods::NOT_IMPLEMENTED
});
&AS_MAPPING
}
}
impl AsNumber for PyGenericAlias {
fn as_number() -> &'static PyNumberMethods {
static AS_NUMBER: PyNumberMethods = PyNumberMethods {
or: Some(|a, b, vm| PyGenericAlias::__or__(a.to_owned(), b.to_owned(), vm)),
..PyNumberMethods::NOT_IMPLEMENTED
};
&AS_NUMBER
}
}
impl Callable for PyGenericAlias {
type Args = FuncArgs;
fn call(zelf: &Py<Self>, args: FuncArgs, vm: &VirtualMachine) -> PyResult {
zelf.origin.call(args, vm).map(|obj| {
if let Err(exc) = obj.set_attr(identifier!(vm, __orig_class__), zelf.to_owned(), vm)
&& !exc.fast_isinstance(vm.ctx.exceptions.attribute_error)
&& !exc.fast_isinstance(vm.ctx.exceptions.type_error)
{
return Err(exc);
}
Ok(obj)
})?
}
}
impl Comparable for PyGenericAlias {
fn cmp(
zelf: &Py<Self>,
other: &PyObject,
op: PyComparisonOp,
vm: &VirtualMachine,
) -> PyResult<PyComparisonValue> {
op.eq_only(|| {
let other = class_or_notimplemented!(Self, other);
if zelf.starred != other.starred {
return Ok(PyComparisonValue::Implemented(false));
}
Ok(PyComparisonValue::Implemented(
zelf.origin
.rich_compare_bool(&other.origin, PyComparisonOp::Eq, vm)?
&& zelf.args.as_object().rich_compare_bool(
other.args.as_object(),
PyComparisonOp::Eq,
vm,
)?,
))
})
}
}
impl Hashable for PyGenericAlias {
#[inline]
fn hash(zelf: &Py<Self>, vm: &VirtualMachine) -> PyResult<hash::PyHash> {
Ok(zelf.origin.hash(vm)? ^ zelf.args.as_object().hash(vm)?)
}
}
impl GetAttr for PyGenericAlias {
fn getattro(zelf: &Py<Self>, attr: &Py<PyStr>, vm: &VirtualMachine) -> PyResult {
let attr_str = attr.as_wtf8();
for exc in &ATTR_EXCEPTIONS {
if attr_str == *exc {
return zelf.as_object().generic_getattr(attr, vm);
}
}
for blocked in &ATTR_BLOCKED {
if attr_str == *blocked {
return zelf.as_object().generic_getattr(attr, vm);
}
}
zelf.origin.get_attr(attr, vm)
}
}
impl Representable for PyGenericAlias {
#[inline]
fn repr_str(zelf: &Py<Self>, vm: &VirtualMachine) -> PyResult<String> {
zelf.repr(vm)
}
}
impl Iterable for PyGenericAlias {
fn iter(zelf: PyRef<Self>, vm: &VirtualMachine) -> PyResult {
Ok(PyGenericAliasIterator {
obj: crate::common::lock::PyMutex::new(Some(zelf.into())),
}
.into_pyobject(vm))
}
}
#[pyclass(module = "types", name = "generic_alias_iterator")]
#[derive(Debug, PyPayload)]
pub(crate) struct PyGenericAliasIterator {
obj: crate::common::lock::PyMutex<Option<PyObjectRef>>,
}
#[pyclass(with(Representable, Iterable, IterNext))]
impl Py<PyGenericAliasIterator> {
#[pymethod]
fn __reduce__(&self, vm: &VirtualMachine) -> PyResult<PyTupleRef> {
let iter_fn = vm.builtins.get_attr("iter", vm)?;
let guard = self.obj.lock();
let arg: PyObjectRef = if let Some(ref obj) = *guard {
PyTuple::new_ref(vec![obj.clone()], &vm.ctx).into()
} else {
let empty = PyTuple::new_ref(vec![], &vm.ctx);
PyTuple::new_ref(vec![empty.into()], &vm.ctx).into()
};
Ok(PyTuple::new_ref(vec![iter_fn, arg], &vm.ctx))
}
}
impl Representable for PyGenericAliasIterator {
fn repr_str(_zelf: &Py<Self>, _vm: &VirtualMachine) -> PyResult<String> {
Ok("<generic_alias_iterator>".to_owned())
}
}
impl Iterable for PyGenericAliasIterator {
fn iter(zelf: PyRef<Self>, _vm: &VirtualMachine) -> PyResult {
Ok(zelf.into())
}
}
impl crate::types::IterNext for PyGenericAliasIterator {
fn next(zelf: &Py<Self>, vm: &VirtualMachine) -> PyResult<crate::protocol::PyIterReturn> {
use crate::protocol::PyIterReturn;
let mut guard = zelf.obj.lock();
let obj = match guard.take() {
Some(obj) => obj,
None => return Ok(PyIterReturn::StopIteration(None)),
};
let alias = obj
.downcast_ref::<PyGenericAlias>()
.ok_or_else(|| vm.new_type_error("generic_alias_iterator expected GenericAlias"))?;
let starred = PyGenericAlias::new(alias.origin.clone(), alias.args.clone(), true, vm)?;
Ok(PyIterReturn::Return(starred.into_pyobject(vm)))
}
}
pub(crate) fn subscript_generic(type_params: PyObjectRef, vm: &VirtualMachine) -> PyResult {
let typing_module = vm.import("typing", 0)?;
let generic_type = typing_module.get_attr("Generic", vm)?;
let generic_alias_class = typing_module.get_attr("_GenericAlias", vm)?;
let params = if let Ok(tuple) = type_params.try_to_ref::<PyTuple>(vm) {
tuple.to_owned()
} else {
PyTuple::new_ref(vec![type_params], &vm.ctx)
};
let args = crate::stdlib::_typing::unpack_typevartuples(¶ms, vm)?;
generic_alias_class.call((generic_type, args.to_pyobject(vm)), vm)
}
pub(crate) fn init(context: &'static Context) {
PyGenericAlias::extend_class(context, context.types.generic_alias_type);
PyGenericAliasIterator::extend_class(context, context.types.generic_alias_iterator_type);
}