nat.builder.function_base#

Base class for NAT functions providing type handling and schema management.

This module contains the FunctionBase abstract base class which provides core functionality for NAT functions including type handling via generics, schema management for inputs and outputs, and type conversion capabilities.

Attributes#

`InputT`
`StreamingOutputT`
`SingleOutputT`
`logger`

Classes#

FunctionBase

Abstract base class providing core functionality for NAT functions.

Module Contents#

InputT#

StreamingOutputT#

SingleOutputT#

logger#

class FunctionBase( *, input_schema: type[pydantic.BaseModel] | None = None, streaming_output_schema: type[pydantic.BaseModel] | type[None] | None = None, single_output_schema: type[pydantic.BaseModel] | type[None] | None = None, converters: list[collections.abc.Callable[[Any], Any]] | None = None, )#

Bases: Generic[InputT, StreamingOutputT, SingleOutputT], abc.ABC

Abstract base class providing core functionality for NAT functions.

This class provides type handling via generics, schema management for inputs and outputs, and type conversion capabilities.

Parameters#

InputTTypeVar: The input type for the function
StreamingOutputTTypeVar: The output type for streaming results
SingleOutputTTypeVar: The output type for single results

Notes#

FunctionBase is the foundation of the NAT function system, providing: - Type handling via generics - Schema management for inputs and outputs - Type conversion capabilities - Abstract interface that concrete function classes must implement

_converter_list = []#

_input_schema#

_converter: nat.utils.type_converter.TypeConverter#

input_type() → type[InputT]#

Get the input type of the function. The input type is determined by the generic parameters of the class.

For example, if a function is defined as def my_function(input: list[int]) -> str, the input_type is list[int].

Returns#

type[InputT]: The input type specified in the generic parameters

Raises#

ValueError: If the input type cannot be determined from the class definition

input_class() → type#

Get the python class of the input type. This is the class that can be used to check if a value is an instance of the input type. It removes any generic or annotation information from the input type.

For example, if a function is defined as def my_function(input: list[int]) -> str, the input_class is list.

Returns#

type: The python type of the input type

input_schema() → type[pydantic.BaseModel]#

Get the Pydantic model schema for validating inputs. The schema must be pydantic models. This allows for type validation and coercion, and documenting schema properties of the input value. If the input type is already a pydantic model, it will be returned as is.

For example, if a function is defined as def my_function(input: list[int]) -> str, the input_schema is:

class InputSchema(BaseModel):
    input: list[int]

Returns#

type[BaseModel]: The Pydantic model class for input validation

property converter_list: list[collections.abc.Callable[[Any], Any]]#

Get the list of type converters used by this function.

Returns#

list[Callable[[typing.Any], typing.Any]]: List of converter functions that transform input types

streaming_output_type() → type[StreamingOutputT]#

Get the streaming output type of the function. The streaming output type is determined by the generic parameters of the class.

For example, if a function is defined as def my_function(input: int) -> AsyncGenerator[dict[str, Any]], the streaming_output_type is dict[str, Any].

Returns#

type[StreamingOutputT]: The streaming output type specified in the generic parameters

Raises#

ValueError: If the streaming output type cannot be determined from the class definition

streaming_output_class() → type#

Get the python class of the output type. This is the class that can be used to check if a value is an instance of the output type. It removes any generic or annotation information from the output type.

For example, if a function is defined as def my_function(input: int) -> AsyncGenerator[dict[str, Any]], the streaming_output_class is dict.

Returns#

type: The python type of the output type

streaming_output_schema() → type[pydantic.BaseModel] | type[None]#

Get the Pydantic model schema for validating streaming outputs. The schema must be pydantic models. This allows for type validation and coercion, and documenting schema properties of the output value. If the output type is already a pydantic model, it will be returned as is.

For example, if a function is defined as def my_function(input: int) -> AsyncGenerator[dict[str, Any]], the streaming_output_schema is:

class StreamingOutputSchema(BaseModel):
    value: dict[str, Any]

Returns#

type[BaseModel] | type[None]: The Pydantic model class for streaming output validation, or NoneType if no streaming output.

single_output_type() → type[SingleOutputT]#

Get the single output type of the function. The single output type is determined by the generic parameters of the class. Returns NoneType if no single output is supported.

For example, if a function is defined as def my_function(input: int) -> list[str], the single_output_type is list[str].

Returns#

type[SingleOutputT]: The single output type specified in the generic parameters

Raises#

ValueError: If the single output type cannot be determined from the class definition

single_output_class() → type#

Get the python class of the output type. This is the class that can be used to check if a value is an instance of the output type. It removes any generic or annotation information from the output type.

For example, if a function is defined as def my_function(input: int) -> list[str], the single_output_class is list.

Returns#

type: The python type of the output type

single_output_schema() → type[pydantic.BaseModel] | type[None]#

Get the Pydantic model schema for validating single outputs. The schema must be pydantic models. This allows for type validation and coercion, and documenting schema properties of the output value. If the output type is already a pydantic model, it will be returned as is.

For example, if a function is defined as def my_function(input: int) -> list[str], the single_output_schema is:

class SingleOutputSchema(BaseModel):
    value: list[str]

Returns#

type[BaseModel] | type[None]: The Pydantic model class for single output validation, or None if no single output

property has_streaming_output: bool#

Check if this function supports streaming output.

Returns#

bool: True if the function supports streaming output, False otherwise

property has_single_output: bool#

Check if this function supports single output.

Returns#

bool: True if the function supports single output, False otherwise

_convert_input(value: Any) → InputT#