use super::signature::Param;
use crate::{
AsObject, Py, PyObject, PyObjectRef, PyPayload, PyRef, PyResult, TryFromObject, VirtualMachine,
builtins::{PyBaseExceptionRef, PyTupleRef, PyType},
common::wtf8::{Wtf8, Wtf8Buf},
convert::ToPyObject,
object::{Traverse, TraverseFn},
};
use core::ops::{Deref, DerefMut, RangeInclusive};
use indexmap::IndexMap;
use itertools::Itertools;
use std::hash::DefaultHasher;
pub trait IntoFuncArgs: Sized {
fn into_args(self, vm: &VirtualMachine) -> FuncArgs;
fn into_method_args(self, obj: PyObjectRef, vm: &VirtualMachine) -> FuncArgs {
let mut args = self.into_args(vm);
let mut with_obj = Vec::with_capacity(args.args.len() + 1);
with_obj.push(obj);
with_obj.append(&mut args.args);
args.args = with_obj;
args
}
}
impl<T> IntoFuncArgs for T
where
T: Into<FuncArgs>,
{
fn into_args(self, _vm: &VirtualMachine) -> FuncArgs {
self.into()
}
}
macro_rules! into_func_args_from_tuple {
($(($n:tt, $T:ident)),*) => {
impl<$($T,)*> IntoFuncArgs for ($($T,)*)
where
$($T: ToPyObject,)*
{
#[inline]
fn into_args(self, vm: &VirtualMachine) -> FuncArgs {
let ($($n,)*) = self;
PosArgs::new(vec![$($n.to_pyobject(vm),)*]).into()
}
#[inline]
fn into_method_args(self, obj: PyObjectRef, vm: &VirtualMachine) -> FuncArgs {
let ($($n,)*) = self;
PosArgs::new(vec![obj, $($n.to_pyobject(vm),)*]).into()
}
}
};
}
into_func_args_from_tuple!((v1, T1));
into_func_args_from_tuple!((v1, T1), (v2, T2));
into_func_args_from_tuple!((v1, T1), (v2, T2), (v3, T3));
into_func_args_from_tuple!((v1, T1), (v2, T2), (v3, T3), (v4, T4));
into_func_args_from_tuple!((v1, T1), (v2, T2), (v3, T3), (v4, T4), (v5, T5));
into_func_args_from_tuple!((v1, T1), (v2, T2), (v3, T3), (v4, T4), (v5, T5), (v6, T6));
#[derive(Debug, Default, Clone, Traverse)]
pub struct FuncArgs {
pub args: Vec<PyObjectRef>,
pub kwargs: KwArgs,
}
impl<A> From<A> for FuncArgs
where
A: Into<PosArgs>,
{
fn from(args: A) -> Self {
Self {
args: args.into().into_vec(),
..Default::default()
}
}
}
impl<Name: ArgName> From<KwArgs<PyObjectRef, Name>> for FuncArgs {
fn from(kwargs: KwArgs<PyObjectRef, Name>) -> Self {
Self {
kwargs: KwArgs::new(kwargs.0),
..Default::default()
}
}
}
impl FromArgs for FuncArgs {
const PARAMS: Option<&'static [Param]> =
Some(&[Param::var_positional("args"), Param::var_keyword("kwargs")]);
fn from_args(_vm: &VirtualMachine, args: &mut FuncArgs) -> Result<Self, ArgumentError> {
Ok(core::mem::take(args))
}
}
impl FuncArgs {
pub fn new<A, K>(args: A, kwargs: K) -> Self
where
A: Into<PosArgs>,
K: Into<KwArgs>,
{
let PosArgs(args, _) = args.into();
Self {
args,
kwargs: kwargs.into(),
}
}
pub fn with_kwargs_names<A, KW>(mut args: A, kwarg_names: KW) -> Self
where
A: ExactSizeIterator<Item = PyObjectRef>,
KW: ExactSizeIterator<Item = String>,
{
let total_argc = args.len();
let kwarg_count = kwarg_names.len();
let pos_arg_count = total_argc - kwarg_count;
let pos_args = args.by_ref().take(pos_arg_count).collect();
let kwargs = kwarg_names.zip_eq(args).collect();
Self {
args: pos_args,
kwargs,
}
}
#[must_use]
pub fn from_vectorcall(
args: &[PyObjectRef],
nargs: usize,
kwnames: Option<&[PyObjectRef]>,
) -> Self {
debug_assert!(nargs <= args.len());
debug_assert!(kwnames.is_none_or(|kw| nargs + kw.len() <= args.len()));
let pos_args = args[..nargs].to_vec();
let kwargs = kwnames.map_or_else(KwArgs::default, |names| {
names
.iter()
.zip(&args[nargs..nargs + names.len()])
.map(|(name, val)| {
let key = name
.downcast_ref::<crate::builtins::PyStr>()
.expect("kwnames must be strings")
.as_wtf8()
.to_owned();
(key, val.clone())
})
.collect()
});
Self {
args: pos_args,
kwargs,
}
}
#[must_use]
pub fn from_vectorcall_owned(
mut args: Vec<PyObjectRef>,
nargs: usize,
kwnames: Option<&[PyObjectRef]>,
) -> Self {
debug_assert!(nargs <= args.len());
debug_assert!(kwnames.is_none_or(|kw| nargs + kw.len() <= args.len()));
let kwargs = kwnames.map_or_else(KwArgs::default, |names| {
let kw_count = names.len();
names
.iter()
.zip(args.drain(nargs..nargs + kw_count))
.map(|(name, val)| {
let key = name
.downcast_ref::<crate::builtins::PyStr>()
.expect("kwnames must be strings")
.as_wtf8()
.to_owned();
(key, val)
})
.collect()
});
args.truncate(nargs);
Self { args, kwargs }
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.args.is_empty() && self.kwargs.is_empty()
}
pub fn prepend_arg(&mut self, item: PyObjectRef) {
self.args.reserve(1);
self.args.insert(0, item)
}
pub fn shift(&mut self) -> PyObjectRef {
self.args.remove(0)
}
#[must_use]
pub fn get_kwarg(&self, key: &str, default: &PyObject) -> PyObjectRef {
self.kwargs
.get(key)
.cloned()
.unwrap_or_else(|| default.to_owned())
}
#[must_use]
pub fn get_optional_kwarg(&self, key: &str) -> Option<PyObjectRef> {
self.kwargs.get(key).cloned()
}
pub fn get_optional_kwarg_with_type(
&self,
key: &str,
ty: &Py<PyType>,
vm: &VirtualMachine,
) -> PyResult<Option<PyObjectRef>> {
match self.get_optional_kwarg(key) {
Some(kwarg) => {
if kwarg.fast_isinstance(ty) {
Ok(Some(kwarg))
} else {
let expected_ty_name = &ty.name();
let kwarg_class = kwarg.class();
let actual_ty_name = &kwarg_class.name();
Err(vm.new_type_error(format!(
"argument of type {expected_ty_name} is required for named parameter `{key}` (got: {actual_ty_name})"
)))
}
}
None => Ok(None),
}
}
pub fn take_positional(&mut self) -> Option<PyObjectRef> {
if self.args.is_empty() {
None
} else {
Some(self.args.remove(0))
}
}
pub fn take_positional_keyword(&mut self, name: &str) -> Option<PyObjectRef> {
self.take_positional().or_else(|| self.take_keyword(name))
}
pub fn take_keyword(&mut self, name: &str) -> Option<PyObjectRef> {
self.kwargs.swap_remove(name)
}
pub fn remaining_keywords(&mut self) -> impl Iterator<Item = (Wtf8Buf, PyObjectRef)> + '_ {
self.kwargs.drain(..)
}
pub fn bind<T: FromArgs>(self, vm: &VirtualMachine) -> PyResult<T> {
self.bind_for(vm, Callee::default())
}
pub fn bind_for<T: FromArgs>(
mut self,
vm: &VirtualMachine,
callee: impl Into<Callee>,
) -> PyResult<T> {
let callee = callee.into();
let instance = callee.instance_args();
let arity = T::arity();
let arity = arity.start().saturating_sub(instance)..=arity.end().saturating_sub(instance);
let num_given = self.args.len().saturating_sub(instance);
let bound = T::from_args(vm, &mut self)
.map_err(|e| e.into_exception(&arity, num_given, callee, vm))?;
if !self.args.is_empty() {
Err(ArgumentError::TooManyArgs.into_exception(&arity, num_given, callee, vm))
} else if let Some(err) = self.check_kwargs_empty_for(vm, callee) {
Err(err)
} else {
Ok(bound)
}
}
pub fn check_kwargs_empty(&self, vm: &VirtualMachine) -> Option<PyBaseExceptionRef> {
self.check_kwargs_empty_for(vm, Callee::default())
}
pub fn check_kwargs_empty_for(
&self,
vm: &VirtualMachine,
callee: impl Into<Callee>,
) -> Option<PyBaseExceptionRef> {
let callee = callee.into();
self.kwargs
.keys()
.next()
.map(|k| callee.unexpected_keyword(&k.to_string(), vm))
}
}
#[derive(Clone, Copy, Debug, Default)]
pub struct Callee {
name: Option<&'static str>,
instance_arg: bool,
}
impl From<&'static str> for Callee {
fn from(name: &'static str) -> Self {
Self::named(name)
}
}
impl Callee {
#[must_use]
pub const fn named(name: &'static str) -> Self {
Self {
name: Some(name),
instance_arg: false,
}
}
#[must_use]
pub fn for_type(class: &crate::Py<crate::builtins::PyType>) -> Self {
Self::named(class.slots.name)
}
#[must_use]
pub fn of<T: crate::PyPayload>(vm: &VirtualMachine) -> Self {
Self::for_type(T::class(&vm.ctx))
}
#[must_use]
pub const fn with_instance_arg(mut self, instance_arg: bool) -> Self {
self.instance_arg = instance_arg;
self
}
const fn instance_args(self) -> usize {
self.instance_arg as usize
}
fn wrong_arity(
self,
arity: &RangeInclusive<usize>,
too_few: bool,
num_given: usize,
vm: &VirtualMachine,
) -> PyBaseExceptionRef {
vm.new_type_error(arity_message(self.name, arity, too_few, num_given))
}
fn unexpected_keyword(self, keyword: &str, vm: &VirtualMachine) -> PyBaseExceptionRef {
vm.new_type_error(unexpected_keyword_message(self.name, keyword))
}
fn missing_argument(
self,
keyword: &str,
pos: usize,
vm: &VirtualMachine,
) -> PyBaseExceptionRef {
vm.new_type_error(missing_argument_message(self.name, keyword, pos))
}
}
fn short_name(name: &str) -> &str {
name.rsplit_once('.').map_or(name, |(_, name)| name)
}
fn call_form<'a>(name: Option<&'a str>, unnamed: &'a str) -> (&'a str, &'static str) {
match name.map(short_name) {
Some(name) => (name, "()"),
None => (unnamed, ""),
}
}
pub(crate) fn arity_message(
name: Option<&str>,
arity: &RangeInclusive<usize>,
too_few: bool,
num_given: usize,
) -> String {
let (limit, bound) = if too_few {
(*arity.start(), "at least ")
} else {
(*arity.end(), "at most ")
};
let bound = if arity.start() == arity.end() {
""
} else {
bound
};
let plural = if limit == 1 { "" } else { "s" };
let name = name
.map(short_name)
.map_or_else(String::new, |name| format!("{name} "));
format!("{name}expected {bound}{limit} argument{plural}, got {num_given}")
}
pub(crate) fn unexpected_keyword_message(name: Option<&str>, keyword: &str) -> String {
let (name, parens) = call_form(name, "this function");
format!("{name}{parens} got an unexpected keyword argument '{keyword}'")
}
pub(crate) fn missing_argument_message(name: Option<&str>, keyword: &str, pos: usize) -> String {
let (name, parens) = call_form(name, "function");
format!("{name}{parens} missing required argument '{keyword}' (pos {pos})")
}
pub enum ArgumentError {
TooFewArgs,
TooManyArgs,
InvalidKeywordArgument(String),
MissingRequiredArgument { name: String, pos: usize },
Exception(PyBaseExceptionRef),
}
impl From<PyBaseExceptionRef> for ArgumentError {
fn from(ex: PyBaseExceptionRef) -> Self {
Self::Exception(ex)
}
}
impl ArgumentError {
fn into_exception(
self,
arity: &RangeInclusive<usize>,
num_given: usize,
callee: Callee,
vm: &VirtualMachine,
) -> PyBaseExceptionRef {
match self {
Self::TooFewArgs => callee.wrong_arity(arity, true, num_given, vm),
Self::TooManyArgs => callee.wrong_arity(arity, false, num_given, vm),
Self::InvalidKeywordArgument(name) => callee.unexpected_keyword(&name, vm),
Self::MissingRequiredArgument { name, pos } => callee.missing_argument(&name, pos, vm),
Self::Exception(ex) => ex,
}
}
}
pub trait FromArgs: Sized {
#[must_use]
fn arity() -> RangeInclusive<usize> {
0..=0
}
const PARAMS: Option<&'static [Param]> = None;
fn from_args(vm: &VirtualMachine, args: &mut FuncArgs) -> Result<Self, ArgumentError>;
}
pub trait FromArgOptional {
type Inner: TryFromObject;
fn from_inner(x: Self::Inner) -> Self;
}
pub trait OptionalArgDefault {
const PY_DEFAULT: super::signature::DefaultRepr;
}
impl<T> OptionalArgDefault for Option<T> {
const PY_DEFAULT: super::signature::DefaultRepr = super::signature::DefaultRepr::None;
}
impl<T> OptionalArgDefault for OptionalArg<T> {
const PY_DEFAULT: super::signature::DefaultRepr =
super::signature::DefaultRepr::Unrepresentable;
}
impl<T: TryFromObject> FromArgOptional for OptionalArg<T> {
type Inner = T;
fn from_inner(x: T) -> Self {
Self::Present(x)
}
}
impl<T: TryFromObject> FromArgOptional for T {
type Inner = Self;
fn from_inner(x: Self) -> Self {
x
}
}
pub trait ArgName {
const NAME: &'static str;
}
macro_rules! arg_name {
($($ty:ident = $name:literal),* $(,)?) => {$(
#[derive(Clone, Copy, Debug)]
pub struct $ty;
impl ArgName for $ty {
const NAME: &'static str = $name;
}
)*};
}
arg_name! {
NameArgs = "args",
NameKwargs = "kwargs",
NameOthers = "others",
NameCoordinates = "coordinates",
NameIntegers = "integers",
NameChanges = "changes",
NameExcInfo = "exc_info",
NameKwds = "kwds",
NameKws = "kws",
NameObjs = "objs",
NameIterables = "iterables",
NameKeywords = "keywords",
NameFields = "fields",
}
#[derive(Clone, Debug)]
pub struct KwArgs<T = PyObjectRef, Name: ArgName = NameKwargs>(
KwArgsMap<T>,
core::marker::PhantomData<Name>,
);
pub type KwArgsMap<T> = IndexMap<Wtf8Buf, T, core::hash::BuildHasherDefault<DefaultHasher>>;
impl<T> Default for KwArgs<T> {
fn default() -> Self {
Self(KwArgsMap::default(), core::marker::PhantomData)
}
}
impl<T, Name: ArgName> Deref for KwArgs<T, Name> {
type Target = KwArgsMap<T>;
fn deref(&self) -> &Self::Target {
&self.0
}
}
impl<T, Name: ArgName> DerefMut for KwArgs<T, Name> {
fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.0
}
}
unsafe impl<T, Name: ArgName> Traverse for KwArgs<T, Name>
where
T: Traverse,
{
fn traverse(&self, tracer_fn: &mut TraverseFn<'_>) {
self.values().for_each(|v| v.traverse(tracer_fn));
}
}
impl<T> KwArgs<T> {
#[must_use]
pub const fn new(map: KwArgsMap<T>) -> Self {
Self(map, core::marker::PhantomData)
}
}
impl<T, Name: ArgName> KwArgs<T, Name> {
#[must_use]
pub fn get(&self, name: &str) -> Option<&T> {
self.0.get(Wtf8::new(name))
}
#[must_use]
pub fn contains_key(&self, name: &str) -> bool {
self.0.contains_key(Wtf8::new(name))
}
pub fn swap_remove(&mut self, name: &str) -> Option<T> {
self.0.swap_remove(Wtf8::new(name))
}
pub fn shift_remove(&mut self, name: &str) -> Option<T> {
self.0.shift_remove(Wtf8::new(name))
}
pub fn pop_kwarg(&mut self, name: &str) -> Option<T> {
self.swap_remove(name)
}
#[must_use]
pub fn into_default(self) -> KwArgs<T> {
KwArgs(self.0, core::marker::PhantomData)
}
}
impl<K: Into<Wtf8Buf>, T> FromIterator<(K, T)> for KwArgs<T> {
fn from_iter<I: IntoIterator<Item = (K, T)>>(iter: I) -> Self {
Self(
iter.into_iter().map(|(k, v)| (k.into(), v)).collect(),
core::marker::PhantomData,
)
}
}
impl<'a, T, Name: ArgName> IntoIterator for &'a KwArgs<T, Name> {
type Item = (&'a Wtf8Buf, &'a T);
type IntoIter = indexmap::map::Iter<'a, Wtf8Buf, T>;
fn into_iter(self) -> Self::IntoIter {
self.0.iter()
}
}
impl<T, Name: ArgName> IntoIterator for KwArgs<T, Name> {
type Item = (Wtf8Buf, T);
type IntoIter = indexmap::map::IntoIter<Wtf8Buf, T>;
fn into_iter(self) -> Self::IntoIter {
self.0.into_iter()
}
}
impl<T, Name: ArgName> FromArgs for KwArgs<T, Name>
where
T: TryFromObject,
{
const PARAMS: Option<&'static [Param]> = Some(&[Param::var_keyword(Name::NAME)]);
fn from_args(vm: &VirtualMachine, args: &mut FuncArgs) -> Result<Self, ArgumentError> {
let mut kwargs = KwArgsMap::default();
for (name, value) in args.remaining_keywords() {
kwargs.insert(name, value.try_into_value(vm)?);
}
Ok(Self(kwargs, core::marker::PhantomData))
}
}
#[derive(Clone)]
pub struct PosArgs<T = PyObjectRef, Name: ArgName = NameArgs>(
Vec<T>,
core::marker::PhantomData<Name>,
);
unsafe impl<T, Name: ArgName> Traverse for PosArgs<T, Name>
where
T: Traverse,
{
fn traverse(&self, tracer_fn: &mut TraverseFn<'_>) {
self.0.traverse(tracer_fn)
}
}
impl<T> PosArgs<T> {
#[must_use]
pub const fn new(args: Vec<T>) -> Self {
Self(args, core::marker::PhantomData)
}
}
impl<T, Name: ArgName> PosArgs<T, Name> {
#[must_use]
pub const fn named(args: Vec<T>) -> Self {
Self(args, core::marker::PhantomData)
}
#[must_use]
pub fn into_vec(self) -> Vec<T> {
self.0
}
pub fn iter(&self) -> core::slice::Iter<'_, T> {
self.0.iter()
}
}
impl<T> From<Vec<T>> for PosArgs<T> {
fn from(v: Vec<T>) -> Self {
Self(v, core::marker::PhantomData)
}
}
impl From<()> for PosArgs<PyObjectRef> {
fn from(_args: ()) -> Self {
Self(Vec::new(), core::marker::PhantomData)
}
}
impl<T, Name: ArgName> AsRef<[T]> for PosArgs<T, Name> {
fn as_ref(&self) -> &[T] {
&self.0
}
}
impl<T: PyPayload, Name: ArgName> PosArgs<PyRef<T>, Name> {
pub fn into_tuple(self, vm: &VirtualMachine) -> PyTupleRef {
vm.ctx
.new_tuple(self.0.into_iter().map(Into::into).collect())
}
}
impl<T, Name: ArgName> FromArgs for PosArgs<T, Name>
where
T: TryFromObject,
{
const PARAMS: Option<&'static [Param]> = Some(&[Param::var_positional(Name::NAME)]);
fn from_args(vm: &VirtualMachine, args: &mut FuncArgs) -> Result<Self, ArgumentError> {
let mut varargs = Vec::new();
while let Some(value) = args.take_positional() {
varargs.push(value.try_into_value(vm)?);
}
Ok(Self(varargs, core::marker::PhantomData))
}
}
impl<T, Name: ArgName> IntoIterator for PosArgs<T, Name> {
type Item = T;
type IntoIter = alloc::vec::IntoIter<T>;
fn into_iter(self) -> Self::IntoIter {
self.0.into_iter()
}
}
impl<T> FromArgs for T
where
T: TryFromObject,
{
fn arity() -> RangeInclusive<usize> {
1..=1
}
fn from_args(vm: &VirtualMachine, args: &mut FuncArgs) -> Result<Self, ArgumentError> {
let value = args.take_positional().ok_or(ArgumentError::TooFewArgs)?;
Ok(value.try_into_value(vm)?)
}
}
#[derive(Debug, result_like::OptionLike, is_macro::Is)]
pub enum OptionalArg<T = PyObjectRef> {
Present(T),
Missing,
}
unsafe impl<T> Traverse for OptionalArg<T>
where
T: Traverse,
{
fn traverse(&self, tracer_fn: &mut TraverseFn<'_>) {
match self {
Self::Present(o) => o.traverse(tracer_fn),
Self::Missing => (),
}
}
}
#[derive(FromArgs)]
pub struct PositionalIterable {
#[pyarg(positional, default, py_default = "()")]
pub iterable: OptionalArg<PyObjectRef>,
}
impl OptionalArg<PyObjectRef> {
pub fn unwrap_or_none(self, vm: &VirtualMachine) -> PyObjectRef {
self.unwrap_or_else(|| vm.ctx.none())
}
}
pub type OptionalOption<T = PyObjectRef> = OptionalArg<Option<T>>;
impl<T> OptionalOption<T> {
#[inline]
pub fn flatten(self) -> Option<T> {
self.into_option().flatten()
}
}
impl<T> FromArgs for OptionalArg<T>
where
T: TryFromObject,
{
const PARAMS: Option<&'static [Param]> = Some(&[Param {
name: "",
kind: super::signature::ParamKind::PositionalOnly,
default: Some(super::signature::DefaultRepr::Unrepresentable),
}]);
fn arity() -> RangeInclusive<usize> {
0..=1
}
fn from_args(vm: &VirtualMachine, args: &mut FuncArgs) -> Result<Self, ArgumentError> {
let r = if let Some(value) = args.take_positional() {
Self::Present(value.try_into_value(vm)?)
} else {
Self::Missing
};
Ok(r)
}
}
impl FromArgs for () {
const PARAMS: Option<&'static [Param]> = Some(&[]);
fn from_args(_vm: &VirtualMachine, _args: &mut FuncArgs) -> Result<Self, ArgumentError> {
Ok(())
}
}
macro_rules! tuple_from_py_func_args {
($($T:ident),+) => {
impl<$($T),+> FromArgs for ($($T,)+)
where
$($T: FromArgs),+
{
const PARAMS: Option<&'static [Param]> = {
if true $(&& $T::PARAMS.is_some())* {
Some(&[$(Param::flatten($T::PARAMS)),*])
} else {
None
}
};
fn arity() -> RangeInclusive<usize> {
let mut min = 0;
let mut max = 0;
$(
let (start, end) = $T::arity().into_inner();
min += start;
max += end;
)+
min..=max
}
fn from_args(vm: &VirtualMachine, args: &mut FuncArgs) -> Result<Self, ArgumentError> {
Ok(($($T::from_args(vm, args)?,)+))
}
}
};
}
tuple_from_py_func_args!(A);
tuple_from_py_func_args!(A, B);
tuple_from_py_func_args!(A, B, C);
tuple_from_py_func_args!(A, B, C, D);
tuple_from_py_func_args!(A, B, C, D, E);
tuple_from_py_func_args!(A, B, C, D, E, F);
tuple_from_py_func_args!(A, B, C, D, E, F, G);
tuple_from_py_func_args!(A, B, C, D, E, F, G, H);