import hashlib
import inspect
from functools import cached_property
from typing import TYPE_CHECKING, Any, Optional
from pynenc.conf.config_pynenc import ArgumentPrintMode
if TYPE_CHECKING:
from pynenc import Pynenc
from pynenc.conf.config_pynenc import ConfigPynenc
from pynenc.types import Args, Func
[docs]
class Arguments:
"""
Represents a distinctive set of arguments used in a task call.
This class facilitates the handling of arguments by providing functionalities such as
generating a unique identifier for a set of arguments, and supporting hashability and equality checks.
:param Optional[Args] kwargs:
Keyword arguments to initialize the Arguments object.
Defaults to an empty dictionary if None is provided.
"""
def __init__(
self, kwargs: Optional["Args"] = None, *, app: Optional["Pynenc"] = None
) -> None:
self.kwargs: "Args" = kwargs or {}
self._app = app
[docs]
@classmethod
def from_call(cls, func: "Func", *args: Any, **kwargs: Any) -> "Arguments":
"""
Creates an Arguments object from a function call.
It uses inspect.signature to determine the function's signature and binds the provided arguments to it.
:param Func func: The function whose arguments are to be processed.
:param Any args: Positional arguments of the function call.
:param Any kwargs: Keyword arguments of the function call.
:return: An instance of Arguments representing the arguments of the call.
"""
sig = inspect.signature(func)
bound_args = sig.bind(*args, **kwargs)
bound_args.apply_defaults()
return cls(bound_args.arguments)
@cached_property
def args_id(self) -> str:
"""
Generates a unique identifier for the set of arguments.
The identifier is a SHA-256 hash of the sorted argument names and values, ensuring uniqueness.
:return: A string representing the unique identifier of the arguments.
"""
if not self.kwargs:
return "no_args"
sorted_items = sorted(self.kwargs.items())
args_str = "".join([f"{k}:{v}" for k, v in sorted_items])
return hashlib.sha256(args_str.encode()).hexdigest()
[docs]
def to_json(self, app: "Pynenc") -> dict[str, Any]:
"""
Serializes the Arguments object to a JSON-compatible dictionary.
:param app: Pynenc application instance for serializing complex arguments.
:return: A dictionary with serialized argument data.
"""
serialized_args = {}
for key, value in self.kwargs.items():
serialized_args[key] = app.arg_cache.serialize(value)
return serialized_args
[docs]
@classmethod
def from_json(cls, app: "Pynenc", data: dict[str, Any]) -> "Arguments":
"""
Deserializes a JSON-compatible dictionary to an Arguments object.
:param app: Pynenc application instance for deserializing complex arguments.
:param data: Dictionary with serialized argument data.
:return: An Arguments object with deserialized arguments.
"""
deserialized_args = {}
for key, value in data.items():
deserialized_args[key] = app.arg_cache.deserialize(value)
return cls(deserialized_args)
[docs]
def __hash__(self) -> int:
return hash(self.args_id)
[docs]
def __eq__(self, __value: object) -> bool:
if not isinstance(__value, Arguments):
return False
return self.args_id == __value.args_id
[docs]
def __str__(self) -> str:
if not self._app:
# Fallback behavior when no app is available
return f"args({', '.join(f'{k}=...' for k in self.kwargs)})"
conf = self._app.conf
if not conf.print_arguments:
return "<arguments hidden>"
if not self.kwargs:
return "<no_args>"
mode = conf.argument_print_mode
if mode == ArgumentPrintMode.HIDDEN:
return "<arguments hidden>"
if mode == ArgumentPrintMode.KEYS:
return f"args({', '.join(self.kwargs.keys())})"
items = []
for k, v in self.kwargs.items():
if mode == ArgumentPrintMode.FULL:
items.append(f"{k}={v}")
else: # TRUNCATED
items.append(f"{k}={self._format_value(conf, v)}")
return f"args({', '.join(items)})"
[docs]
def __repr__(self) -> str:
return f"Arguments({self.kwargs})"