File size: 5,328 Bytes
1b7e88c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 |
import contextvars
from abc import ABC
from typing import Any, List, Optional, Union
from omagent_core.utils.container import container
from pydantic import Field, field_validator
from pydantic_settings import BaseSettings
REQUEST_ID = contextvars.ContextVar("request_id")
class BotBase(BaseSettings, ABC):
name: Optional[str] = Field(default=None, validate_default=True)
id: Optional[str] = Field(default=None, validate_default=True)
component_stm: Optional[str] = None
component_ltm: Optional[str] = None
component_callback: Optional[str] = None
component_input: Optional[str] = None
class Config:
"""Configuration for this pydantic object."""
extra = "allow"
arbitrary_types_allowed = True
@field_validator("name", mode="before")
@classmethod
def get_type(cls, name) -> str:
if not name:
return cls.__name__
else:
return name
@property
def stm(self) -> str:
if self.component_stm is None:
return container.stm
else:
return container.get_component(self.component_stm)
@property
def ltm(self) -> str:
if self.component_ltm is None:
return container.ltm
else:
return container.get_component(self.component_ltm)
@property
def callback(self) -> str:
if self.component_callback is None:
return container.callback
else:
return container.get_component(self.component_callback)
@property
def input(self) -> str:
if self.component_input is None:
return container.input
else:
return container.get_component(self.component_input)
@classmethod
def get_config_template(
cls,
description: bool = True,
env_var: bool = True,
exclude_fields: List[str] = [],
) -> dict:
template = {}
simple_types = (str, int, float, bool, type(None))
def is_simple_type(type_to_check: Any) -> bool:
return isinstance(type_to_check, type) and issubclass(
type_to_check, simple_types
)
def is_botbase_subclass(type_to_check: Any) -> bool:
return (
isinstance(type_to_check, type)
and issubclass(type_to_check, BotBase)
and type_to_check != BotBase
)
for field_name, field in cls.model_fields.items():
# Pass inner attributes
if field_name.startswith("_") or field_name in exclude_fields:
continue
field_type = field.annotation
if hasattr(field_type, "__origin__") and field_type.__origin__ is Union:
types = field_type.__args__
if any(is_botbase_subclass(t) for t in types):
for t in types:
if is_botbase_subclass(t):
template[field_name] = t.get_config_template(
description=description,
env_var=env_var,
exclude_fields=exclude_fields,
)
break
continue
elif not all(is_simple_type(t) for t in types):
continue
elif is_botbase_subclass(field_type):
template[field_name] = field_type.get_config_template(
description=description,
env_var=env_var,
exclude_fields=exclude_fields,
)
continue
elif not is_simple_type(field_type):
continue
field_info = {"value": None}
if description and field.description:
field_info["description"] = field.description
if env_var:
field_info["env_var"] = (
field.alias.upper() if field.alias else field_name.upper()
)
# Get default value for tag as <required>
if field.default_factory is not None:
field_info["value"] = field.default_factory()
elif field.is_required():
field_info["value"] = "<required>"
else:
field_info["value"] = field.default
template[field.alias if field.alias else field_name] = field_info
template["name"] = cls.__name__
return template
@classmethod
def from_config(cls, config_data: dict) -> "BotBase":
def clean_config_dict(config_dict: dict) -> dict:
"""Recursively clean up the configuration dictionary, removing all 'description' and 'env_var' keys"""
cleaned = {}
for key, value in config_dict.items():
if isinstance(value, dict):
if "value" in value:
cleaned[key] = value["value"]
else:
cleaned[key] = clean_config_dict(value)
else:
cleaned[key] = value
return cleaned
class_config = config_data.get(cls.__name__, {})
clean_class_config = clean_config_dict(class_config)
return cls(**clean_class_config)
|