Schema Extractor

The Schema Extractor dynamically discovers and extracts metadata from Python functions and classes in nodepacks.

Features

• Function Discovery: Extracts schemas from functions in ops.py files
• Progress Class Discovery: Extracts schemas from classes with __call__ methods in progress_ops.py files
• Type Hint Parsing: Extracts type information from function signatures and annotations
• Docstring Extraction: Captures documentation strings for nodes

Progress Node Support

The schema extractor scans progress_ops.py files for classes that: - Have a __call__ method - Are not private (don't start with _) - Are defined at module level

Progress node schemas include an is_progress_node: true flag to distinguish them from regular function nodes.

See the Progress Nodes Guide for details on creating progress nodes.

API Reference

psynapse_backend.schema_extractor

AnnotatedDict

Bases: dict, Generic[T]

A dictionary type hint that specifies the expected output keys.

Usage

def my_func() -> AnnotatedDict[Literal["key1", "key2"]]: return {"key1": value1, "key2": value2}

This creates multiple output handles in the node editor, one for each key.

get_type_name(type_hint)

Convert a type hint to a string representation.

Parameters:

Name	Type	Description	Default
type_hint	Any	The type hint to convert.	required

Returns:

Type	Description
str	A string representation of the type hint.

get_literal_values(type_hint)

Extract literal values from a Literal type hint.

Parameters:

Name	Type	Description	Default
type_hint	Any	The type hint to check.	required

Returns:

Type	Description
list[str] | None	A list of string values if the type hint is Literal, None otherwise.

parse_annotated_dict_keys(type_hint)

Parse AnnotatedDict[Literal['key1', 'key2', ...]] and return the key names.

Parameters:

Name	Type	Description	Default
type_hint	Any	The type hint to check.	required

Returns:

Type	Description
list[str] | None	A list of key names if the type hint is AnnotatedDict with Literal keys, None otherwise.

extract_function_schema(func, filepath)

Extract schema information from a function.

Parameters:

Name	Type	Description	Default
func	callable	The function to extract schema information from.	required
filepath	str	The file path of the function.	required

Returns:

Type	Description
dict[str, Any]	A dictionary of schema information.

detect_class_node_type(cls)

Detect the node type of a class based on its attributes.

Parameters:

Name	Type	Description	Default
cls	type	The class to inspect.	required

Returns:

Type	Description
str | None	"progress" if the class has _progress_reporter,
str | None	"stream" if the class has _stream_reporter,
str | None	None otherwise.

extract_class_schema(cls, filepath, node_type=None)

Extract schema information from a class with call method.

Parameters:

Name	Type	Description	Default
cls	type	The class to extract schema information from.	required
filepath	str	The file path of the class.	required
node_type	str | None	The type of node ("progress", "stream", or None for auto-detection).	None

Returns:

Type	Description
dict[str, Any]	A dictionary of schema information.

extract_schemas_from_file(filepath, extract_classes=False, node_type='progress')

Extract schemas from all functions or classes in a Python file.

Parameters:

Name	Type	Description	Default
filepath	str	The file path of the Python file.	required
extract_classes	bool	If True, extract classes with call method. If False, extract functions.	False
node_type	str	The type of node for classes ("progress" or "stream").	'progress'

Returns:

Type	Description
list[dict[str, Any]]	A list of dictionaries of schema information.

extract_all_schemas(nodepacks_dir)

Extract schemas from all ops.py, progress_ops.py, and stream_ops.py files in the nodepacks directory.

Parameters:

Name	Type	Description	Default
nodepacks_dir	str	The directory containing the nodepacks.	required

Returns:

Type	Description
list[dict[str, Any]]	A list of dictionaries of schema information.