codekingpro/portable-devtools

1from __future__ import annotations2 3import dataclasses4import types5import weakref6from collections.abc import Generator, Sequence7from typing import Annotated, Any, Optional, Union, get_origin, get_type_hints8 9from pydantic import BaseModel10from typing_extensions import NotRequired, ReadOnly, Required11 12from langgraph._internal._typing import MISSING13 14 15def _is_optional_type(type_: Any) -> bool:16    """Check if a type is Optional."""17 18    # Handle new union syntax (PEP 604): str | None19    if isinstance(type_, types.UnionType):20        return any(21            arg is type(None) or _is_optional_type(arg) for arg in type_.__args__22        )23 24    if hasattr(type_, "__origin__") and hasattr(type_, "__args__"):25        origin = get_origin(type_)26        if origin is Optional:27            return True28        if origin is Union:29            return any(30                arg is type(None) or _is_optional_type(arg) for arg in type_.__args__31            )32        if origin is Annotated:33            return _is_optional_type(type_.__args__[0])34        return origin is None35    if hasattr(type_, "__bound__") and type_.__bound__ is not None:36        return _is_optional_type(type_.__bound__)37    return type_ is None38 39 40def _is_required_type(type_: Any) -> bool | None:41    """Check if an annotation is marked as Required/NotRequired.42 43    Returns:44        - True if required45        - False if not required46        - None if not annotated with either47    """48    origin = get_origin(type_)49    if origin is Required:50        return True51    if origin is NotRequired:52        return False53    if origin is Annotated or getattr(origin, "__args__", None):54        # See https://typing.readthedocs.io/en/latest/spec/typeddict.html#interaction-with-annotated55        return _is_required_type(type_.__args__[0])56    return None57 58 59def _is_readonly_type(type_: Any) -> bool:60    """Check if an annotation is marked as ReadOnly.61 62    Returns:63        - True if is read only64        - False if not read only65    """66 67    # See: https://typing.readthedocs.io/en/latest/spec/typeddict.html#typing-readonly-type-qualifier68    origin = get_origin(type_)69    if origin is Annotated:70        return _is_readonly_type(type_.__args__[0])71    if origin is ReadOnly:72        return True73    return False74 75 76_DEFAULT_KEYS: frozenset[str] = frozenset()77 78 79def get_field_default(name: str, type_: Any, schema: type[Any]) -> Any:80    """Determine the default value for a field in a state schema.81 82    This is based on:83        If TypedDict:84            - Required/NotRequired85            - total=False -> everything optional86        - Type annotation (Optional/Union[None])87    """88    optional_keys = getattr(schema, "__optional_keys__", _DEFAULT_KEYS)89    irq = _is_required_type(type_)90    if name in optional_keys:91        # Either total=False or explicit NotRequired.92        # No type annotation trumps this.93        if irq:94            # Unless it's earlier versions of python & explicit Required95            return ...96        return None97    if irq is not None:98        if irq:99            # Handle Required[<type>]100            # (we already handled NotRequired and total=False)101            return ...102        # Handle NotRequired[<type>] for earlier versions of python103        return None104    if dataclasses.is_dataclass(schema):105        field_info = next(106            (f for f in dataclasses.fields(schema) if f.name == name), None107        )108        if field_info:109            if (110                field_info.default is not dataclasses.MISSING111                and field_info.default is not ...112            ):113                return field_info.default114            elif field_info.default_factory is not dataclasses.MISSING:115                return field_info.default_factory()116    # Note, we ignore ReadOnly attributes,117    # as they don't make much sense. (we don't care if you mutate the state in your node)118    # and mutating state in your node has no effect on our graph state.119    # Base case is the annotation120    if _is_optional_type(type_):121        return None122    return ...123 124 125def get_enhanced_type_hints(126    type: type[Any],127) -> Generator[tuple[str, Any, Any, str | None], None, None]:128    """Attempt to extract default values and descriptions from provided type, used for config schema."""129    for name, typ in get_type_hints(type).items():130        default = None131        description = None132 133        # Pydantic models134        try:135            if hasattr(type, "model_fields") and name in type.model_fields:136                field = type.model_fields[name]137 138                if hasattr(field, "description") and field.description is not None:139                    description = field.description140 141                if hasattr(field, "default") and field.default is not None:142                    default = field.default143                    if (144                        hasattr(default, "__class__")145                        and getattr(default.__class__, "__name__", "")146                        == "PydanticUndefinedType"147                    ):148                        default = None149 150        except (AttributeError, KeyError, TypeError):151            pass152 153        # TypedDict, dataclass154        try:155            if hasattr(type, "__dict__"):156                type_dict = getattr(type, "__dict__")157 158                if name in type_dict:159                    default = type_dict[name]160        except (AttributeError, KeyError, TypeError):161            pass162 163        yield name, typ, default, description164 165 166def get_update_as_tuples(input: Any, keys: Sequence[str]) -> list[tuple[str, Any]]:167    """Get Pydantic state update as a list of (key, value) tuples."""168    if isinstance(input, BaseModel):169        keep = input.model_fields_set170        defaults = {k: v.default for k, v in type(input).model_fields.items()}171    else:172        keep = None173        defaults = {}174 175    # NOTE: This behavior for Pydantic is somewhat inelegant,176    # but we keep around for backwards compatibility177    # if input is a Pydantic model, only update values178    # that are different from the default values or in the keep set179    return [180        (k, value)181        for k in keys182        if (value := getattr(input, k, MISSING)) is not MISSING183        and (184            value is not None185            or defaults.get(k, MISSING) is not None186            or (keep is not None and k in keep)187        )188    ]189 190 191ANNOTATED_KEYS_CACHE: weakref.WeakKeyDictionary[type[Any], tuple[str, ...]] = (192    weakref.WeakKeyDictionary()193)194 195 196def get_cached_annotated_keys(obj: type[Any]) -> tuple[str, ...]:197    """Return cached annotated keys for a Python class."""198    if obj in ANNOTATED_KEYS_CACHE:199        return ANNOTATED_KEYS_CACHE[obj]200    if isinstance(obj, type):201        keys: list[str] = []202        for base in reversed(obj.__mro__):203            ann = base.__dict__.get("__annotations__")204            # In Python 3.14+, Pydantic models use descriptors for __annotations__205            # so we need to fall back to getattr if __dict__.get returns None206            if ann is None:207                ann = getattr(base, "__annotations__", None)208            if ann is None or isinstance(ann, types.GetSetDescriptorType):209                continue210            keys.extend(ann.keys())211        return ANNOTATED_KEYS_CACHE.setdefault(obj, tuple(keys))212    else:213        raise TypeError(f"Expected a type, got {type(obj)}. ")214