Class InputBatch

fastdeploy/worker/input_batch.py:26–656

class InputBatch:
    def __getitem__(self, key):
        """Support dictionary-style attribute access"""
        if hasattr(self, key):
            return getattr(self, key)
        raise KeyError(f"'{key}' is not a valid attribute of InputBatch")

    def __setitem__(self, key, value):
        """Support dictionary-style attribute setting, overwrite if exists, add if not exists"""
        setattr(self, key, value)

    def __contains__(self, key):
        """Support 'in' operator to check if attribute exists"""
        return hasattr(self, key)

    def update(self, values: dict):
        """Batch update attributes, similar to dict's update method"""
        for key, value in values.items():
            setattr(self, key, value)

    def pop(self, key, default=None):
        """
        Pop an attribute, similar to dict's pop method

        Args:
            key: Name of the attribute to pop
            default: Default value to return if attribute does not exist

        Returns:
            Popped attribute value, or default if attribute doesn't exist and default is provided
        """
        if hasattr(self, key):
            value = getattr(self, key)
            delattr(self, key)
            return value
        elif default is not None:
            return default
        else:
            raise KeyError(f"'{key}' is not a valid attribute of InputBatch")

    def __delitem__(self, key):
        """
        Delete an attribute using del operator

        Args:
            key: Name of the attribute to delete

        Raises:
            KeyError: If attribute does not exist
        """
        if hasattr(self, key):
            delattr(self, key)
        else:
            raise KeyError(f"'{key}' is not a valid attribute of InputBatch")

    def __init__(self, fd_config: FDConfig) -> None:
        """
        Initialize all share buffers for model inputs.