diff --git a/neon_utils/skills/neon_skill.py b/neon_utils/skills/neon_skill.py index b9ce3c66..6394c502 100644 --- a/neon_utils/skills/neon_skill.py +++ b/neon_utils/skills/neon_skill.py @@ -278,17 +278,24 @@ def wrapper(message): request_class = None try: from pydantic import BaseModel + + def is_model(annotation) -> bool: + # Typing generics like `Optional[str]` are not classes + # and make `issubclass` raise TypeError + return isinstance(annotation, type) and \ + issubclass(annotation, BaseModel) + parameters = signature.parameters for arg_name, param in parameters.items(): if arg_name == 'self': continue - if issubclass(param.annotation, BaseModel): + if is_model(param.annotation): # Get the JSON schema for the BaseModel schema = param.annotation.model_json_schema() request_class = param.annotation break - if signature.return_annotation and issubclass(signature.return_annotation, BaseModel): + if is_model(signature.return_annotation): # Get the JSON schema for the return type return_schema = signature.return_annotation.model_json_schema() except ImportError: diff --git a/tests/neon_skill_tests.py b/tests/neon_skill_tests.py index c8241c9f..57bb5b23 100644 --- a/tests/neon_skill_tests.py +++ b/tests/neon_skill_tests.py @@ -37,9 +37,12 @@ from os.path import join from threading import Thread from time import sleep, time +from typing import Optional from ovos_bus_client import Message from ovos_utils.messagebus import FakeBus +from ovos_workshop.decorators import skill_api_method +from pydantic import BaseModel, Field, RootModel from mock import Mock from neon_utils.skills import NeonSkill, CommonMessageSkill, CommonPlaySkill, CommonQuerySkill, NeonFallbackSkill @@ -1308,5 +1311,65 @@ def _find_resource(*args, **kwargs): self.skill.find_resource = real_method +class LocationRequest(BaseModel): + location: Optional[str] = Field(default=None, + description="Location to look up") + + +class YearResponse(RootModel): + root: str = Field(description="Year (YYYY)") + + +class TypedApiSkill(NeonSkill): + @skill_api_method + def get_year(self, request: LocationRequest) -> YearResponse: + """Get the year for a location""" + return f"{type(request).__name__}:{request.location}" + + +class GenericApiSkill(NeonSkill): + @skill_api_method + def get_optional(self, value: Optional[str] = None) -> Optional[str]: + """Echo an optional value""" + return value + + +class NeonSkillPublicApiTests(unittest.TestCase): + @staticmethod + def _get_skill(skill_class, skill_id): + bus = FakeBus() + skill = skill_class(skill_id=skill_id, bus=bus) + return skill, bus + + def test_typed_api_method(self): + skill, bus = self._get_skill(TypedApiSkill, "typed.test") + api = bus.wait_for_response(Message("typed.test.public_api")).data + method = api["get_year"] + self.assertEqual(method["type"], "typed.test.get_year") + self.assertEqual(method["request_schema"], + LocationRequest.model_json_schema()) + self.assertEqual(method["response_schema"], + YearResponse.model_json_schema()) + + # JSON kwargs from the bus are validated into the request model + response = bus.wait_for_response( + Message("typed.test.get_year", + {"args": [], "kwargs": {"location": "Tokyo"}})).data + self.assertEqual(response, {"result": "LocationRequest:Tokyo", + "error": None}) + + def test_generic_annotations(self): + skill, bus = self._get_skill(GenericApiSkill, "generic.test") + api = bus.wait_for_response(Message("generic.test.public_api")).data + method = api["get_optional"] + self.assertIsNone(method["request_schema"]) + self.assertIsNone(method["response_schema"]) + + response = bus.wait_for_response( + Message("generic.test.get_optional", + {"args": [], "kwargs": {"value": "test"}})).data + self.assertEqual(response, {"result": "test", "error": None}) + + if __name__ == '__main__': unittest.main()