confit.registry

def make_wrapper(raw_function, callee, config, invoker):
    """
    Prepare validation once, then resolve and execute each registered call
    """
    is_class = isinstance(callee, type)
    target = inspect.unwrap(raw_function) if is_class else raw_function
    signature = inspect.signature(target)
    parameters = signature.parameters
    self_name = next(iter(parameters)) if is_class else None
    extras_name = next(
        (name for name, p in parameters.items() if p.kind == p.VAR_KEYWORD), None
    )
    callee_name = (
        f"{callee.__module__}.{callee.__qualname__}"
        if callee.__module__
        else callee.__qualname__
    )
    # Keep Field defaults, ordinary defaults stay omitted through stacked decorators
    omitted_defaults = {
        name
        for name, p in parameters.items()
        if p.default is not inspect.Parameter.empty
        and not isinstance(p.default, FieldInfo)
    }

    @wraps(target, assigned=WRAPPER_ATTRIBUTES)
    def collect(*args, **kwargs):
        return args, kwargs

    validator = validate_call(collect, config=config)

    @wraps(raw_function, assigned=WRAPPER_ATTRIBUTES)
    def wrapper(*args, **kwargs):
        try:
            extras = {
                k: v
                for k, v in kwargs.items()
                if k not in parameters and extras_name is None
            }
            try:
                bound = signature.bind_partial(
                    *args, **{k: v for k, v in kwargs.items() if k not in extras}
                )
            except TypeError as error:
                error = TypeError(
                    str(error).replace(
                        "multiple values for argument ",
                        "multiple values for argument: ",
                    )
                )
                raise ConfitValidationError(
                    [
                        {
                            "loc": ("[signature]",),
                            "msg": str(error),
                            "type": "arguments_error",
                        }
                    ],
                    source=callee,
                    name=callee_name,
                ) from None
            values = dict(bound.arguments)
            if extras_name in values:
                values.update(values.pop(extras_name))
            values.update(extras)
            resolved = values.pop(self_name) if self_name is not None else None
            returned = None

            def invoked(kw):
                nonlocal returned, resolved
                if self_name is not None:
                    kw = {self_name: resolved, **kw}
                bound = signature.bind_partial()
                bound.arguments.update(
                    {
                        name: kw[name]
                        for name in parameters
                        if name in kw and name != extras_name
                    }
                )
                extras = {
                    k: v
                    for k, v in kw.items()
                    if k not in parameters or k == extras_name
                }
                if extras and extras_name is None:
                    raise ConfitValidationError(
                        [
                            {
                                "loc": (name,),
                                "msg": "unexpected keyword argument",
                                "type": "unexpected_keyword_argument",
                            }
                            for name in extras
                        ],
                        source=callee,
                        name=callee_name,
                    )
                try:
                    validated_args, validated_kwargs = validator(
                        *bound.args, **bound.kwargs, **extras
                    )
                except (ValidationError, ConfitValidationError) as error:
                    raise ConfitValidationError.from_exception(
                        error,
                        source=callee,
                        name=callee_name,
                        signature=signature,
                    ) from None
                # Let the function supply its ordinary defaults
                call = signature.bind_partial(*validated_args, **validated_kwargs)
                for name in omitted_defaults - bound.arguments.keys():
                    call.arguments.pop(name, None)
                returned = raw_function(*call.args, **call.kwargs)
                if self_name is None:
                    resolved = returned
                return resolved

            if invoker is None:
                invoked(values)
            else:
                invoker(invoked, values)
            return returned
        except Exception as error:
            context = error.__context__ if is_class else error.__cause__
            if not is_debug() and isinstance(
                context, (ValidationError, ConfitValidationError)
            ):
                error.__cause__ = None
                error.__suppress_context__ = True
            raise error.with_traceback(remove_lib_from_traceback(error.__traceback__))

    return wrapper

def validate_arguments(
    func: Optional[Callable[P, R]] = None,
    *,
    config: Dict = None,
    invoker: Optional[Callable[[Callable, Dict[str, Any]], Any]] = None,
    registry: Any = None,
) -> Callable[[Callable[P, R]], Draftable[P, R]]:
    """
    Validate function or constructor arguments and support deferred calls with draft

    Parameters
    ----------
    func: Optional[Callable[P, R]]
        The function or class to call
    config: Dict
        The validation configuration object
    invoker: Optional[Callable]
        An optional invoker to apply on the validated function
    registry: Any
        The registry to use to resolve the default parameters

    Returns
    -------
    Callable[[Callable[P, R]], Draftable[P, R]]:
    """
    config = {**(config or {}), "arbitrary_types_allowed": True}

    def validate(_func: Callable) -> Callable:
        is_class = isinstance(_func, type)
        raw_function = _func.__init__ if is_class else _func
        if raw_function is object.__init__:
            # Since `object.__init__` has no named parameters to validate, we
            # validate against an empty constructor to reject unexpected arguments
            def raw_function(self):
                pass

        wrapper = make_wrapper(raw_function, _func, config, invoker)
        draft_type = Draft[_func] if is_class else Draft

        @wraps(raw_function, assigned=WRAPPER_ATTRIBUTES)
        def draft(**kwargs):
            return draft_type(_func, kwargs)

        if is_class:
            old_get_validators = getattr(_func, "__get_validators__", None)
            old_get_pydantic_core_schema = getattr(
                _func, "__get_pydantic_core_schema__", None
            )

            def __get_pydantic_core_schema__(*args, **kwargs):
                from pydantic_core import core_schema

                def pre_validate(value):
                    if isinstance(value, dict):
                        value = Config(value).resolve(registry=registry)
                    return value

                def post_validate(value):
                    if isinstance(value, _func):
                        return value

                    return _func(**value)

                steps = [core_schema.no_info_plain_validator_function(pre_validate)]
                if old_get_pydantic_core_schema is not None:
                    steps.append(old_get_pydantic_core_schema(*args, **kwargs))
                elif old_get_validators is not None:
                    steps.extend(
                        legacy_validator_schema(fn) for fn in old_get_validators()
                    )
                steps.append(
                    core_schema.no_info_plain_validator_function(post_validate)
                )
                return core_schema.chain_schema(steps)

            _func.__get_pydantic_core_schema__ = __get_pydantic_core_schema__
            _func.__init__ = wrapper
        result = _func if is_class else wrapper
        result.draft = draft
        return result

    return validate(func) if func is not None else validate

def get_default_registry() -> Any:
    """
    Get the default registered registry.

    Returns
    -------
    Registry
    """
    return _default_registry

def set_default_registry(registry: CustomRegistry) -> CustomRegistry:
    """
    Set the default registered registry. This is used in
    [`Config.resolve()`][confit.config.Config.resolve] when no registry is provided.

    Parameters
    ----------
    registry: Registry

    Returns
    -------
    Registry
    """
    global _default_registry
    _default_registry = registry
    return registry