From 3677216e626d4a7c82f879d6213758248baf37fe Mon Sep 17 00:00:00 2001 From: Richard Bowman Date: Wed, 4 Dec 2024 01:34:06 +0000 Subject: [PATCH 1/9] Code generation for ThingClient subclasses This generates Python code for a module containing a client. The Python code generated for the test thing executes successfully, though the name of the class is not quite right yet. --- .../code_generation/__init__.py | 168 ++++++++++++++++++ tests/test_client_generation.py | 26 +++ 2 files changed, 194 insertions(+) create mode 100644 src/labthings_fastapi/code_generation/__init__.py create mode 100644 tests/test_client_generation.py diff --git a/src/labthings_fastapi/code_generation/__init__.py b/src/labthings_fastapi/code_generation/__init__.py new file mode 100644 index 00000000..0d634b43 --- /dev/null +++ b/src/labthings_fastapi/code_generation/__init__.py @@ -0,0 +1,168 @@ +from inspect import cleandoc +import re +from typing import Sequence + +from labthings_fastapi.thing_description.model import ( + DataSchema, + ThingDescription, + Type, +) + + +def title_to_snake_case(title: str) -> str: + """Convert text to snake_case""" + words = re.findall(r"[a-z0-9]+", title.lower()) + return "_".join(words) + + +def snake_to_camel_case(snake: str) -> str: + """Convert snake_case to CamelCase""" + words = snake.split("_") + return "".join(word.capitalize() for word in words) + + +def title_to_camel_case(title: str) -> str: + """Convert text to CamelCase""" + return snake_to_camel_case(title_to_snake_case(title)) + + +def clean_code(code: str, prefix: str = "") -> str: + """Clean up code by removing leading/trailing whitespace and empty lines""" + lines = cleandoc(code).split("\n") + return "\n".join([prefix + l for l in lines]) + + +def quoted_docstring(docstring: str, indent: int = 4) -> str: + """Wrap a docstring in triple quotes""" + prefix = " " * indent + lines = docstring.split("\n") + lines[0] = f'"""{lines[0]}' + lines.append('"""') + return "".join([f"{prefix}{line}\n" for line in lines]) + + +def dataschema_to_type(schema: DataSchema) -> str: + """Convert a DataSchema to a Python type""" + if isinstance(schema.oneOf, Sequence) and len(schema.oneOf) > 0: + types = [dataschema_to_type(s) for s in schema.oneOf] + return f"Union[{", ".join(types)}]" + if schema.type == Type.string: + return "str" + elif schema.type == Type.integer: + return "int" + elif schema.type == Type.number: + return "float" + elif schema.type == Type.boolean: + return "bool" + elif schema.type == Type.array: + if schema.items is None: + return "list" + return f"list[{dataschema_to_type(schema.items)}]" + elif schema.type == Type.object: + return "dict[str, Any]" + else: + return "Any" + +def property_to_argument( + name: str, + property: DataSchema, + ) -> str: + """Convert a property to a function argument""" + dtype = dataschema_to_type(property) + arg = f"{name}: {dtype}" + if "default" in property.model_fields_set: + if property.default is None: + arg += " = None" + elif isinstance(property.default, str): + arg += f' = "{property.default}"' + elif ( + isinstance(property.default, bool) + or isinstance(property.default, int) + or isinstance(property.default, float) + ): + arg += f" = {property.default}" + else: + raise NotImplementedError(f"Unsupported default value: {property.default}") + return arg + + +def input_model_to_arguments(model: DataSchema) -> list[str]: + """Convert an input model to a list of arguments""" + if model.type is None: + return [] + if model.type != Type.object: + print(f"model.type: {model.type}") + raise NotImplementedError("Only object models are supported") + if not model.properties: + return [] + args = [] + if model.required: + for name in model.required: + property = model.properties[name] + args.append( + property_to_argument(name, property) + ) + for name, property in model.properties.items(): + if model.required and name in model.required: + continue + args.append(property_to_argument(name, property)) + if "=" not in args[-1]: + args[-1] += " = ..." + return args + + +def generate_client(thing_description: ThingDescription) -> str: + """Generate a client from a Thing Description""" + code = ( + "from labthings_fastapi.client import ThingClient\n" + "from typing import Any, Union\n" + "\n" + ) + class_name = title_to_camel_case(thing_description.title) + code += f"class {class_name}Client(ThingClient):\n" + code += f' """A client for the {thing_description.title} Thing"""\n\n' + for name, property in thing_description.properties.items(): + pname = title_to_snake_case(name) + dtype = dataschema_to_type(property) + code += " @property\n" + code += f" def {pname}(self) -> {dtype}:\n" + code += quoted_docstring(property.description, indent=8) + code += f' return self.get_property("{name}")\n\n' + + if not property.readOnly: + code += clean_code( + f''' + @{pname}.setter + def {pname}(self, value: {dtype}): + self.set_property("{name}", value) + ''', + prefix = " ", + ) + "\n\n" + + for name, action in thing_description.actions.items(): + aname = title_to_snake_case(name) + args = input_model_to_arguments(action.input) + output_type = dataschema_to_type(action.output) + code += f" def {aname}(\n" + code += " self,\n" + for arg in args: + code += f" {arg},\n" + code += " **kwargs\n" + code += f" ) -> {output_type}:\n" + code += quoted_docstring(action.description, indent=8) + for arg in args: + k = arg.split(":")[0].strip() + if arg.endswith("..."): + code += clean_code( + f""" + if {k} is not ...: + kwargs[{k}] = {k} + """, + prefix = " ", + ) + "\n" + else: + code += f" kwargs[{k}] = {k}\n" + code += f' return self.invoke_action("{name}", **kwargs)\n\n' + + return code + diff --git a/tests/test_client_generation.py b/tests/test_client_generation.py new file mode 100644 index 00000000..fe1864f4 --- /dev/null +++ b/tests/test_client_generation.py @@ -0,0 +1,26 @@ +import os +import tempfile +import importlib.util + +from labthings_fastapi.code_generation import generate_client +from labthings_fastapi.example_things import MyThing + + +def test_client_generation(): + td = MyThing().thing_description() + code = generate_client(td) + with tempfile.TemporaryDirectory() as d: + fname = os.path.join(d,"client.py") + with open(fname, "w") as f: + f.write(code) + spec = importlib.util.spec_from_file_location("client", f.name) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + assert "MyThingClient" in dir(module) + +if __name__ == "__main__": + td = MyThing().thing_description() + print("Thing Description:") + print(td.model_dump_json(indent=2, exclude_unset=True)) + print("\nGenerated Client:") + print(generate_client(td)) From 63846e5324031f866aa23647f51dcae88f62f155 Mon Sep 17 00:00:00 2001 From: Richard Bowman Date: Tue, 21 Jan 2025 00:46:25 +0000 Subject: [PATCH 2/9] Convert objets to Models Dataschemas of type Object that have defined properties are now converted to Pydantic models. This should allow much better autocompletion. The inconsistency of using `dict[str, any]` as a fallback rankles, but I think it's the best option. --- .../code_generation/__init__.py | 88 +++++++++++++++---- tests/test_client_generation.py | 59 +++++++++++-- 2 files changed, 125 insertions(+), 22 deletions(-) diff --git a/src/labthings_fastapi/code_generation/__init__.py b/src/labthings_fastapi/code_generation/__init__.py index 0d634b43..0524da39 100644 --- a/src/labthings_fastapi/code_generation/__init__.py +++ b/src/labthings_fastapi/code_generation/__init__.py @@ -1,6 +1,6 @@ from inspect import cleandoc import re -from typing import Sequence +from typing import Optional, Sequence from labthings_fastapi.thing_description.model import ( DataSchema, @@ -11,8 +11,10 @@ def title_to_snake_case(title: str) -> str: """Convert text to snake_case""" - words = re.findall(r"[a-z0-9]+", title.lower()) - return "_".join(words) + # First, look for CamelCase so it doesn't get ignored: + uncameled = re.sub(r"([a-z0-9])([A-Z])", r"\1_\2", title) + words = re.findall(r"[a-zA-Z0-9]+", uncameled) + return "_".join(w.lower() for w in words) def snake_to_camel_case(snake: str) -> str: @@ -32,8 +34,10 @@ def clean_code(code: str, prefix: str = "") -> str: return "\n".join([prefix + l for l in lines]) -def quoted_docstring(docstring: str, indent: int = 4) -> str: +def quoted_docstring(docstring: Optional[str], indent: int = 4) -> str: """Wrap a docstring in triple quotes""" + if docstring is None: + return "" prefix = " " * indent lines = docstring.split("\n") lines[0] = f'"""{lines[0]}' @@ -41,10 +45,10 @@ def quoted_docstring(docstring: str, indent: int = 4) -> str: return "".join([f"{prefix}{line}\n" for line in lines]) -def dataschema_to_type(schema: DataSchema) -> str: +def dataschema_to_type(schema: DataSchema, models: dict[str, str], name: str = "anonymous") -> str: """Convert a DataSchema to a Python type""" if isinstance(schema.oneOf, Sequence) and len(schema.oneOf) > 0: - types = [dataschema_to_type(s) for s in schema.oneOf] + types = [dataschema_to_type(s, models) for s in schema.oneOf] return f"Union[{", ".join(types)}]" if schema.type == Type.string: return "str" @@ -57,18 +61,50 @@ def dataschema_to_type(schema: DataSchema) -> str: elif schema.type == Type.array: if schema.items is None: return "list" - return f"list[{dataschema_to_type(schema.items)}]" + if isinstance(schema.items, Sequence): + types = [dataschema_to_type(s, models) for s in schema.items] + return f"tuple[{', '.join(types)}]" + return f"list[{dataschema_to_type(schema.items, models)}]" elif schema.type == Type.object: - return "dict[str, Any]" + # If the object has no properties, return a generic dict + if not schema.properties: + return "dict[str, Any]" + # Objects with properties are converted to Pydantic models + if schema.title: + model_name = title_to_camel_case(schema.title + "_model") + else: + model_name = snake_to_camel_case(name + "_model") + if model_name in models: + i = 0 + while f"{model_name}_{i}" in models: + i += 1 + model_name = f"{model_name}_{i}" + models[model_name] = "# placeholder" + models[model_name] = dataschema_to_model(schema, models, model_name) + return model_name else: return "Any" + +def dataschema_to_model(schema: DataSchema, models: dict[str, str], name: str) -> str: + """Convert a DataSchema to a Pydantic model""" + code = f"class {name}(BaseModel):\n" + for pname, property in schema.properties.items(): + code += " " + property_to_argument(pname, property, models) + "\n" + code += ( + "\n" + " class Config:\n" + " extra = 'allow'\n" + ) + return code + def property_to_argument( name: str, property: DataSchema, + models: dict[str, str] = None, ) -> str: """Convert a property to a function argument""" - dtype = dataschema_to_type(property) + dtype = dataschema_to_type(property, models, name) arg = f"{name}: {dtype}" if "default" in property.model_fields_set: if property.default is None: @@ -86,7 +122,7 @@ def property_to_argument( return arg -def input_model_to_arguments(model: DataSchema) -> list[str]: +def input_model_to_arguments(model: DataSchema, models) -> list[str]: """Convert an input model to a list of arguments""" if model.type is None: return [] @@ -100,12 +136,12 @@ def input_model_to_arguments(model: DataSchema) -> list[str]: for name in model.required: property = model.properties[name] args.append( - property_to_argument(name, property) + property_to_argument(name, property, models) ) for name, property in model.properties.items(): if model.required and name in model.required: continue - args.append(property_to_argument(name, property)) + args.append(property_to_argument(name, property, models)) if "=" not in args[-1]: args[-1] += " = ..." return args @@ -116,18 +152,29 @@ def generate_client(thing_description: ThingDescription) -> str: code = ( "from labthings_fastapi.client import ThingClient\n" "from typing import Any, Union\n" + "from pydantic import BaseModel\n" + "\n" + "\n" + "# Model definitions\n" # will be replaced at the end + "\n" "\n" ) + models: dict[str, str] = {} class_name = title_to_camel_case(thing_description.title) code += f"class {class_name}Client(ThingClient):\n" code += f' """A client for the {thing_description.title} Thing"""\n\n' for name, property in thing_description.properties.items(): pname = title_to_snake_case(name) - dtype = dataschema_to_type(property) + dtype = dataschema_to_type(property, models=models) code += " @property\n" code += f" def {pname}(self) -> {dtype}:\n" code += quoted_docstring(property.description, indent=8) - code += f' return self.get_property("{name}")\n\n' + code += f" val = self.get_property(\"{name}\")\n" + if dtype in models: + # If we've defined a model, convert it + code += f" return {dtype}(**val)\n\n" + else: + code += " return val\n\n" if not property.readOnly: code += clean_code( @@ -141,8 +188,8 @@ def {pname}(self, value: {dtype}): for name, action in thing_description.actions.items(): aname = title_to_snake_case(name) - args = input_model_to_arguments(action.input) - output_type = dataschema_to_type(action.output) + args = input_model_to_arguments(action.input, models) + output_type = dataschema_to_type(action.output, models) code += f" def {aname}(\n" code += " self,\n" for arg in args: @@ -162,7 +209,14 @@ def {pname}(self, value: {dtype}): ) + "\n" else: code += f" kwargs[{k}] = {k}\n" - code += f' return self.invoke_action("{name}", **kwargs)\n\n' + code += f' result = self.invoke_action("{name}", **kwargs)\n' + if output_type in models: + code += f" return {output_type}(**result)\n\n" + else: + code += " return result\n\n" + + # Include the model definitions + code = code.replace("# Model definitions", "\n\n".join(models.values())) return code diff --git a/tests/test_client_generation.py b/tests/test_client_generation.py index fe1864f4..e0859b4e 100644 --- a/tests/test_client_generation.py +++ b/tests/test_client_generation.py @@ -2,24 +2,73 @@ import tempfile import importlib.util +from pydantic import BaseModel + from labthings_fastapi.code_generation import generate_client +import labthings_fastapi.code_generation as cg +from labthings_fastapi.decorators import thing_action, thing_property from labthings_fastapi.example_things import MyThing +from labthings_fastapi.thing import Thing + + +def test_title_to_snake_case(): + assert cg.title_to_snake_case("CamelCase") == "camel_case" + assert cg.title_to_snake_case("Camel") == "camel" + assert cg.title_to_snake_case("camel") == "camel" + assert cg.title_to_snake_case("CAMEL") == "camel" + assert cg.title_to_snake_case("CamelCASE") == "camel_case" + +def test_snake_to_camel_case(): + assert cg.snake_to_camel_case("snake_case") == "SnakeCase" + assert cg.snake_to_camel_case("snake") == "Snake" + assert cg.snake_to_camel_case("SNAKE") == "Snake" + assert cg.snake_to_camel_case("snakeCASE_word") == "SnakecaseWord" -def test_client_generation(): - td = MyThing().thing_description() + +def generate_and_verify(thing): + td = thing().thing_description() code = generate_client(td) with tempfile.TemporaryDirectory() as d: - fname = os.path.join(d,"client.py") + fname = os.path.join(d, "client.py") with open(fname, "w") as f: f.write(code) spec = importlib.util.spec_from_file_location("client", f.name) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) - assert "MyThingClient" in dir(module) + assert f"{thing.__name__}Client" in dir(module) + + +def test_mything_generation(): + generate_and_verify(MyThing) + + +class TestModel(BaseModel): + a: int + b: str + +class NestedModel(BaseModel): + c: TestModel + +class ThingWithModels(Thing): + @thing_property + def prop1(self) -> TestModel: + return TestModel(a=1, b="test") + + @thing_action + def action1(self, arg1: TestModel) -> TestModel: + return arg1 + + @thing_property + def prop2(self) -> NestedModel: + return NestedModel(c=TestModel(a=1, b="test")) + + +def test_with_models(): + generate_and_verify(ThingWithModels) if __name__ == "__main__": - td = MyThing().thing_description() + td = ThingWithModels().thing_description() print("Thing Description:") print(td.model_dump_json(indent=2, exclude_unset=True)) print("\nGenerated Client:") From 816c679ee499d232a67250134d90b206d2c50871 Mon Sep 17 00:00:00 2001 From: Richard Bowman Date: Thu, 4 Jun 2026 17:19:45 +0100 Subject: [PATCH 3/9] Convert code generation to use ast module This should be significantly more secure: I believe it should now be impossible for a malcious Thing Description to create dangerous Python code. In doing this, I've simplified the way the client creates properties - I need to check if it type hints them correctly. There are still a few things to fix: * Default values probably want to be `...` rather than constants, so they're filled in server-side. * This doesn't yet validate/convert return values. * This isn't tested enough and probably doesn't actually work! * We need configurability: as a minimum, recursion limits and fallback to `Any` so it fails gracefully if given a bad TD. --- src/labthings_fastapi/client/__init__.py | 131 +++--- .../code_generation/__init__.py | 412 ++++++++++++------ tests/test_client_generation.py | 44 +- 3 files changed, 361 insertions(+), 226 deletions(-) diff --git a/src/labthings_fastapi/client/__init__.py b/src/labthings_fastapi/client/__init__.py index 64a567cf..3ba01096 100644 --- a/src/labthings_fastapi/client/__init__.py +++ b/src/labthings_fastapi/client/__init__.py @@ -9,7 +9,7 @@ import time from collections.abc import Mapping -from typing import Any, Optional, Union +from typing import Any, Generic, Optional, TypeVar from urllib.parse import urljoin, urlparse import httpx @@ -328,83 +328,64 @@ class PropertyClientDescriptor: path: str -def property_descriptor( - property_name: str, - model: Union[type, BaseModel], - description: Optional[str] = None, - readable: bool = True, - writeable: bool = True, - property_path: Optional[str] = None, -) -> PropertyClientDescriptor: - """Create a correctly-typed descriptor that gets and/or sets a property. - - The returned `.PropertyClientDescriptor` will have ``__get__`` and - (optionally) ``__set__`` methods that are typed according to the - supplied ``model``. The descriptor should be added to a `~lt.ThingClient` - subclass and used to access the relevant property via - `.ThingClient.get_property` and `.ThingClient.set_property`. - - :param property_name: should be the name of the property (i.e. the - name it takes in the thing description, and also the name it is - assigned to in the class). - :param model: the Python ``type`` or a ``pydantic.BaseModel`` that - represents the datatype of the property. - :param description: text to use for a docstring. - :param readable: whether the property may be read (i.e. has ``__get__``). - :param writeable: whether the property may be written to. - :param property_path: the URL of the ``getproperty`` and ``setproperty`` - HTTP endpoints. Currently these must both be the same. These are - relative to the ``base_url``, i.e. the URL of the Thing Description. - - :return: a descriptor allowing access to the specified property. - """ +Value = TypeVar("Value") - class P(PropertyClientDescriptor): - name = property_name - type = model - path = property_path or property_name - - if readable: - - def __get__( - self: PropertyClientDescriptor, - obj: Optional[ThingClient] = None, - _objtype: Optional[type[ThingClient]] = None, - ) -> Any: - if obj is None: - return self - return obj.get_property(self.name) - else: - - def __get__( - self: PropertyClientDescriptor, - obj: Optional[ThingClient] = None, - _objtype: Optional[type[ThingClient]] = None, - ) -> Any: - raise ClientPropertyError("This property may not be read.") - __get__.__annotations__["return"] = model - P.__get__ = __get__ # type: ignore[attr-defined] +class ClientProperty(Generic[Value]): + """A descriptor to make properties of ThingClient objects work.""" - # Set __set__ method based on whether writable - if writeable: + def __init__( + self, + name: str, + readable: bool = True, + writeable: bool = True, + doc: str | None = None, + ) -> None: + """Initialise a ClientProperty. - def __set__( - self: PropertyClientDescriptor, obj: ThingClient, value: Any - ) -> None: - obj.set_property(self.name, value) - else: + :param name: The name of the property. + :param writeable: whether the property should be writeable. + """ + self._name = name + self._readable = readable + self._writeable = writeable + if doc: + self.__doc__ = doc + + def __get__( + self, obj: ThingClient | None, cls: type[ThingClient] | None = None + ) -> Value | Self: + """Retrieve the property. + + :param obj: The client object on which the property is accessed. + """ + if obj is None: + return self + if self._readable: + return obj.get_property(self._name) + else: + raise ClientPropertyError("This property may not be read.") - def __set__( - self: PropertyClientDescriptor, obj: ThingClient, value: Any - ) -> None: + def __set__(self, obj: ThingClient, value: Value) -> None | Self: + """Retrieve the property. + + :param obj: The client object on which the property is accessed. + """ + if self._writeable: + return obj.set_property(self._name, value) + else: raise ClientPropertyError("This property may not be set.") - __set__.__annotations__["value"] = model - P.__set__ = __set__ # type: ignore[attr-defined] - if description: - P.__doc__ = description - return P() + +def client_property( + name: str, doc: str | None, writeable: bool = True, readable: bool = True +) -> Any: + return ClientProperty( + name=name, + doc=doc, + writeable=writeable, + readable=readable, + ) def add_action(cls: type[ThingClient], action_name: str, action: dict) -> None: @@ -447,13 +428,13 @@ def add_property(cls: type[ThingClient], property_name: str, property: dict) -> :param property: a dictionary representing the property, in :ref:`wot_td` format. """ + annotation = property.get("type", Any) setattr( cls, property_name, - property_descriptor( - property_name, - property.get("type", Any), - description=property.get("description", None), + ClientProperty[annotation]( + name=property_name, + doc=property.get("description", None), writeable=not property.get("readOnly", False), readable=not property.get("writeOnly", False), ), diff --git a/src/labthings_fastapi/code_generation/__init__.py b/src/labthings_fastapi/code_generation/__init__.py index 0524da39..92ec73a2 100644 --- a/src/labthings_fastapi/code_generation/__init__.py +++ b/src/labthings_fastapi/code_generation/__init__.py @@ -1,8 +1,21 @@ -from inspect import cleandoc +"""Generate client subclasses based on a Thing Description. + +This module generates a subclass of ThingClient specific to one Thing Description. +Said subclass will have a method corresponding to each Action, and a property +corresponding to each Property. + +The subclass is generated by first constructing an Abstract Syntax Tree +using `ast` and then evaluating it. The same code may also be used to write +a module to a file, allowing static analysis. +""" + +import ast import re +from inspect import cleandoc +from types import NoneType from typing import Optional, Sequence -from labthings_fastapi.thing_description.model import ( +from labthings_fastapi.thing_description._model import ( DataSchema, ThingDescription, Type, @@ -10,7 +23,7 @@ def title_to_snake_case(title: str) -> str: - """Convert text to snake_case""" + """Convert text to snake_case.""" # First, look for CamelCase so it doesn't get ignored: uncameled = re.sub(r"([a-z0-9])([A-Z])", r"\1_\2", title) words = re.findall(r"[a-zA-Z0-9]+", uncameled) @@ -18,24 +31,24 @@ def title_to_snake_case(title: str) -> str: def snake_to_camel_case(snake: str) -> str: - """Convert snake_case to CamelCase""" + """Convert snake_case to CamelCase.""" words = snake.split("_") return "".join(word.capitalize() for word in words) def title_to_camel_case(title: str) -> str: - """Convert text to CamelCase""" + """Convert text to CamelCase.""" return snake_to_camel_case(title_to_snake_case(title)) def clean_code(code: str, prefix: str = "") -> str: - """Clean up code by removing leading/trailing whitespace and empty lines""" + """Clean up code by removing leading/trailing whitespace and empty lines.""" lines = cleandoc(code).split("\n") - return "\n".join([prefix + l for l in lines]) + return "\n".join([prefix + line for line in lines]) def quoted_docstring(docstring: Optional[str], indent: int = 4) -> str: - """Wrap a docstring in triple quotes""" + """Wrap a docstring in triple quotes.""" if docstring is None: return "" prefix = " " * indent @@ -45,178 +58,307 @@ def quoted_docstring(docstring: Optional[str], indent: int = 4) -> str: return "".join([f"{prefix}{line}\n" for line in lines]) -def dataschema_to_type(schema: DataSchema, models: dict[str, str], name: str = "anonymous") -> str: - """Convert a DataSchema to a Python type""" +def _name(name: str) -> ast.Name: + """Generate a Name object (shorthand for `ast.Name`).""" + return ast.Name(id=name, ctx=ast.Load()) + + +def _subscript(value: ast.expr, subscript: ast.expr) -> ast.Subscript: + """Generate a Subscript object (shorthand for `ast.Subscript`).""" + return ast.Subscript(value, subscript, ctx=ast.Load()) + + +def _tuple(*elements: ast.expr) -> ast.Tuple: + """Generate a Tuple (shorthand for `ast.Tuple`).""" + return ast.Tuple(elts=list(elements), ctx=ast.Load()) + + +def _constant(value: str | bool | int | float | NoneType) -> ast.Constant: + """Check a value may be rendered as a constant, and return it.""" + if not isinstance(value, (str, bool, int, float, NoneType)): + raise TypeError("Don't know how to return this as a constant!") + return ast.Constant(value=value) + + +def _import_from(module: str, names: list[str]) -> ast.ImportFrom: + """Import names from a module.""" + return ast.ImportFrom( + module=module, + names=[ast.alias(name=name) for name in names], + level=0, + ) + + +def dataschema_to_type( + schema: DataSchema, models: list[ast.ClassDef], name: str = "anonymous" +) -> ast.expr: + """Convert a DataSchema to a Python type.""" if isinstance(schema.oneOf, Sequence) and len(schema.oneOf) > 0: types = [dataschema_to_type(s, models) for s in schema.oneOf] - return f"Union[{", ".join(types)}]" + return _subscript(_name("Union"), _tuple(*types)) if schema.type == Type.string: - return "str" + return _name("str") elif schema.type == Type.integer: - return "int" + return _name("int") elif schema.type == Type.number: - return "float" + return _name("float") elif schema.type == Type.boolean: - return "bool" + return _name("bool") elif schema.type == Type.array: if schema.items is None: - return "list" + # A dataschema with no `items` member gets typed as a plain list + return _name("list") if isinstance(schema.items, Sequence): + # The WoT spec is based on JSONSchema 2019 + # If `items` is a sequence, it behaves as `prefixItems` in the 2023 + # draft, i.e. it describes a tuple. + # https://json-schema.org/understanding-json-schema/reference/array#tupleValidation types = [dataschema_to_type(s, models) for s in schema.items] - return f"tuple[{', '.join(types)}]" - return f"list[{dataschema_to_type(schema.items, models)}]" + return _subscript(_name("tuple"), _tuple(*types)) + return _subscript(_name("list"), dataschema_to_type(schema.items, models)) elif schema.type == Type.object: # If the object has no properties, return a generic dict if not schema.properties: - return "dict[str, Any]" + return _subscript(_name("dict"), _name("Any")) # Objects with properties are converted to Pydantic models if schema.title: model_name = title_to_camel_case(schema.title + "_model") else: model_name = snake_to_camel_case(name + "_model") - if model_name in models: + model_names = [m.name for m in models] + if model_name in model_names: i = 0 while f"{model_name}_{i}" in models: i += 1 model_name = f"{model_name}_{i}" - models[model_name] = "# placeholder" - models[model_name] = dataschema_to_model(schema, models, model_name) - return model_name + models.append(dataschema_to_model(schema, models, model_name)) + return _name(model_name) else: - return "Any" - -def dataschema_to_model(schema: DataSchema, models: dict[str, str], name: str) -> str: - """Convert a DataSchema to a Pydantic model""" - code = f"class {name}(BaseModel):\n" - for pname, property in schema.properties.items(): - code += " " + property_to_argument(pname, property, models) + "\n" - code += ( - "\n" - " class Config:\n" - " extra = 'allow'\n" + return _name("Any") + + +def dataschema_to_model( + schema: DataSchema, models: list[ast.ClassDef], name: str +) -> ast.ClassDef: + """Convert a DataSchema to a Pydantic model.""" + if schema.properties is None: + msg = f"Can't generate a model from schema '{schema}' as it has no properties." + raise TypeError(msg) + + # The class body will consist of one line setting `_model_config` and + # then one annotated assignment for each property. + class_body: list[ast.stmt] = [] + class_body.append( + ast.Assign( + targets=[ast.Name(id="_model_config", ctx=ast.Store())], + value=ast.Dict(keys=[_constant("extra")], values=[_constant("allow")]), + ) + ) + for pname, prop in schema.properties.items(): + # The fields of the model will appear as assignments. If there's no default, + # it's still an assignment as far as `ast` is concerned, but there's no + # value (or equals sign). + has_default = "default" in prop.model_fields_set + class_body.append( + ast.AnnAssign( + target=ast.Name(id=pname, ctx=ast.Store()), + annotation=dataschema_to_type(prop, models, name), + value=_constant(prop.default) if has_default else None, + simple=1, + ) + ) + return ast.ClassDef( + name=name, + bases=[_name("BaseModel")], + keywords=[], + body=class_body, + decorator_list=[], ) - return code def property_to_argument( - name: str, - property: DataSchema, - models: dict[str, str] = None, - ) -> str: - """Convert a property to a function argument""" + name: str, + property: DataSchema, + models: list[ast.ClassDef], +) -> tuple[ast.arg, ast.expr | None]: + """Convert a property to a function argument and default.""" dtype = dataschema_to_type(property, models, name) - arg = f"{name}: {dtype}" + arg = ast.arg(arg=name, annotation=dtype) if "default" in property.model_fields_set: if property.default is None: - arg += " = None" - elif isinstance(property.default, str): - arg += f' = "{property.default}"' - elif ( - isinstance(property.default, bool) - or isinstance(property.default, int) - or isinstance(property.default, float) - ): - arg += f" = {property.default}" + default = ast.Constant(None) + elif isinstance(property.default, (str, bool, int, float)): + default = ast.Constant(property.default) else: raise NotImplementedError(f"Unsupported default value: {property.default}") - return arg + else: + default = None + return arg, default -def input_model_to_arguments(model: DataSchema, models) -> list[str]: - """Convert an input model to a list of arguments""" +NO_ARGUMENTS = ast.arguments( + posonlyargs=[], + args=[], + vararg=None, + kwonlyargs=[], + kwarg=None, + defaults=[], + kw_defaults=[], +) + + +def input_model_to_arguments( + model: DataSchema, models: list[ast.ClassDef] +) -> ast.arguments: + """Convert an input model to an arguments object. + + :param model: A DataSchema describing the input (of an action). + :param models: A dictionary of Pydantic model definitions we'll need. + :return: an arguments object describing the properties of `model` as + keyword-only arguments. + """ if model.type is None: - return [] + return NO_ARGUMENTS if model.type != Type.object: print(f"model.type: {model.type}") - raise NotImplementedError("Only object models are supported") + raise TypeError("Only object models are supported") if not model.properties: - return [] - args = [] - if model.required: - for name in model.required: - property = model.properties[name] - args.append( - property_to_argument(name, property, models) - ) + return NO_ARGUMENTS + kwargs = [] + kwdefaults = [] + # Make sure required (i.e. no default) arguments come first. + arg_names = list(model.properties.keys()) + required_names = model.required or [] + for name in required_names: + arg_names.remove(name) + arg_names = required_names + arg_names for name, property in model.properties.items(): - if model.required and name in model.required: - continue - args.append(property_to_argument(name, property, models)) - if "=" not in args[-1]: - args[-1] += " = ..." - return args - - -def generate_client(thing_description: ThingDescription) -> str: - """Generate a client from a Thing Description""" - code = ( - "from labthings_fastapi.client import ThingClient\n" - "from typing import Any, Union\n" - "from pydantic import BaseModel\n" - "\n" - "\n" - "# Model definitions\n" # will be replaced at the end - "\n" - "\n" + property = model.properties[name] + arg, default = property_to_argument(name, property, models) + kwargs.append(arg) + # It's possible for required args to have defaults in the schema, + # but we remove these from the method signature. + kwdefaults.append(None if name in required_names else default) + return ast.arguments( + kwonlyargs=kwargs, + kw_defaults=kwdefaults, + posonlyargs=[], + args=[], + vararg=None, + kwarg=None, + defaults=[], ) - models: dict[str, str] = {} + + +def generate_client(thing_description: ThingDescription) -> ast.Module: + """Generate a client from a Thing Description. + + :param thing_description: the Thing Description to use. + """ + # We'll populate this dictionary with auto-generated Pydantic models, + # needed for `DataSchema`s that are typed as `object` + models: list[ast.ClassDef] = [] + class_name = title_to_camel_case(thing_description.title) - code += f"class {class_name}Client(ThingClient):\n" - code += f' """A client for the {thing_description.title} Thing"""\n\n' - for name, property in thing_description.properties.items(): + + # We create the body of the class as a list of statments. This will be + # included in the class definition later. + class_body: list[ast.stmt] = [] + + if thing_description.description: + doc = thing_description.description + else: + doc = f"A client for the {thing_description.title} Thing." + # The class docstring appears as a constant at the top + class_body.append(ast.Expr(value=_constant(doc))) + + # Each property will be an assignment of the form + # `propname: ValueType = client_property(...)` + properties = thing_description.properties or {} + for name, property in properties.items(): pname = title_to_snake_case(name) dtype = dataschema_to_type(property, models=models) - code += " @property\n" - code += f" def {pname}(self) -> {dtype}:\n" - code += quoted_docstring(property.description, indent=8) - code += f" val = self.get_property(\"{name}\")\n" - if dtype in models: - # If we've defined a model, convert it - code += f" return {dtype}(**val)\n\n" - else: - code += " return val\n\n" - - if not property.readOnly: - code += clean_code( - f''' - @{pname}.setter - def {pname}(self, value: {dtype}): - self.set_property("{name}", value) - ''', - prefix = " ", - ) + "\n\n" - - for name, action in thing_description.actions.items(): + class_body.append( + ast.AnnAssign( + target=ast.Name(id=name, ctx=ast.Store()), + annotation=dtype, + value=ast.Call( + func=_name("client_property"), + args=[], + keywords=[ + ast.keyword(arg="name", value=_constant(pname)), + ast.keyword(arg="doc", value=_constant(property.description)), + ast.keyword( + arg="readable", value=_constant(not property.readOnly) + ), + ast.keyword( + arg="writeable", value=_constant(not property.writeOnly) + ), + ], + ), + simple=1, + ) + ) + + # Each action will be a method definition, with appropriately typed arguments and + # return values. These are then passed straight through to `self.invoke_action` + actions = thing_description.actions or {} + for name, action in actions.items(): aname = title_to_snake_case(name) - args = input_model_to_arguments(action.input, models) - output_type = dataschema_to_type(action.output, models) - code += f" def {aname}(\n" - code += " self,\n" - for arg in args: - code += f" {arg},\n" - code += " **kwargs\n" - code += f" ) -> {output_type}:\n" - code += quoted_docstring(action.description, indent=8) - for arg in args: - k = arg.split(":")[0].strip() - if arg.endswith("..."): - code += clean_code( - f""" - if {k} is not ...: - kwargs[{k}] = {k} - """, - prefix = " ", - ) + "\n" - else: - code += f" kwargs[{k}] = {k}\n" - code += f' result = self.invoke_action("{name}", **kwargs)\n' - if output_type in models: - code += f" return {output_type}(**result)\n\n" + if action.input: + args = input_model_to_arguments(action.input, models) + else: + args = NO_ARGUMENTS + if action.output: + rtype = dataschema_to_type(action.output, models) else: - code += " return result\n\n" + rtype = _name("Any") + # The function body simply passes the arguments through to `invoke_action`. + function_body: list[ast.stmt] = [ + ast.Return( + value=ast.Call( + func=ast.Attribute( + value=_name("self"), attr="invoke_action", ctx=ast.Load() + ), + args=[_constant(name)], + keywords=[ + ast.keyword(arg=arg.arg, value=_name(arg.arg)) + for arg in args.kwonlyargs + ], + ) + ) + ] + class_body.append( + ast.FunctionDef( + name=aname, + args=args, + body=function_body, + decorator_list=[], + returns=rtype, + ) + ) + + # The class definition is here: this includes `class_body` defined above, with the + # actions/properties. + class_definition = ast.ClassDef( + name=f"{class_name}Client", + bases=[_name("ThingClient")], + keywords=[], + body=class_body, + decorator_list=[], + ) + + # The module we want to create starts here: + # Import the symbols we'll need + code: list[ast.stmt] = [] + code.append( + _import_from("labthings_fastapi.client", ["ThingClient", "client_property"]) + ) + code.append(_import_from("typing", ["Any", "Union"])) + code.append(_import_from("pydantic", ["BaseModel"])) - # Include the model definitions - code = code.replace("# Model definitions", "\n\n".join(models.values())) + code += models - return code + code.append(class_definition) + return ast.Module(body=code, type_ignores=[]) diff --git a/tests/test_client_generation.py b/tests/test_client_generation.py index e0859b4e..8853081d 100644 --- a/tests/test_client_generation.py +++ b/tests/test_client_generation.py @@ -1,14 +1,15 @@ +import ast +import importlib.util import os import tempfile -import importlib.util from pydantic import BaseModel -from labthings_fastapi.code_generation import generate_client +import labthings_fastapi as lt import labthings_fastapi.code_generation as cg -from labthings_fastapi.decorators import thing_action, thing_property +from labthings_fastapi.code_generation import generate_client from labthings_fastapi.example_things import MyThing -from labthings_fastapi.thing import Thing +from labthings_fastapi.testing import create_thing_without_server def test_title_to_snake_case(): @@ -27,14 +28,18 @@ def test_snake_to_camel_case(): def generate_and_verify(thing): - td = thing().thing_description() - code = generate_client(td) + td = create_thing_without_server(thing).thing_description() + tree = generate_client(td) + ast.fix_missing_locations(tree) + code = ast.unparse(tree) with tempfile.TemporaryDirectory() as d: fname = os.path.join(d, "client.py") with open(fname, "w") as f: f.write(code) spec = importlib.util.spec_from_file_location("client", f.name) + assert spec is not None module = importlib.util.module_from_spec(spec) + assert spec.loader is not None spec.loader.exec_module(module) assert f"{thing.__name__}Client" in dir(module) @@ -47,29 +52,36 @@ class TestModel(BaseModel): a: int b: str + class NestedModel(BaseModel): c: TestModel -class ThingWithModels(Thing): - @thing_property + +class ThingWithModels(lt.Thing): + @lt.property def prop1(self) -> TestModel: return TestModel(a=1, b="test") - - @thing_action + + @lt.action def action1(self, arg1: TestModel) -> TestModel: return arg1 - - @thing_property + + @lt.property def prop2(self) -> NestedModel: return NestedModel(c=TestModel(a=1, b="test")) def test_with_models(): generate_and_verify(ThingWithModels) - + + if __name__ == "__main__": - td = ThingWithModels().thing_description() + td = create_thing_without_server(ThingWithModels).thing_description() print("Thing Description:") print(td.model_dump_json(indent=2, exclude_unset=True)) - print("\nGenerated Client:") - print(generate_client(td)) + print("\nGenerated AST:") + ast_module = generate_client(td) + print(ast.dump(ast_module, indent=4)) + print("\nGenerated module:") + ast.fix_missing_locations(ast_module) + print(ast.unparse(ast_module)) From 71574ce5ffa03b11658109a1904b938ed0a9866a Mon Sep 17 00:00:00 2001 From: Richard Bowman Date: Wed, 10 Jun 2026 23:32:54 +0100 Subject: [PATCH 4/9] Add configurability and tidy a lot This commit adds: * a limit on recursion * a bunch of tidying up * many more comments * cleaner generated code * generated docstrings for actions, properties, and Things. * thing client properties now derive from BaseDescriptor - this required a few changes, but makes the generated code neater. * BaseDescriptor may now be defined on non-Thing classes. --- src/labthings_fastapi/base_descriptor.py | 9 +- src/labthings_fastapi/client/__init__.py | 203 ++---- .../code_generation/__init__.py | 670 ++++++++++++------ tests/test_client_generation.py | 33 +- tests/test_thing_client.py | 28 +- 5 files changed, 593 insertions(+), 350 deletions(-) diff --git a/src/labthings_fastapi/base_descriptor.py b/src/labthings_fastapi/base_descriptor.py index c3b5bc18..ac397dd3 100644 --- a/src/labthings_fastapi/base_descriptor.py +++ b/src/labthings_fastapi/base_descriptor.py @@ -33,7 +33,7 @@ from collections.abc import Iterator from itertools import pairwise from types import MappingProxyType -from typing import TYPE_CHECKING, Any, Generic, Mapping, TypeVar, overload +from typing import Any, Generic, Mapping, TypeVar, overload from weakref import WeakKeyDictionary, ref from typing_extensions import Self @@ -48,13 +48,10 @@ ) from labthings_fastapi.utilities.introspection import get_docstring, get_summary -if TYPE_CHECKING: - from labthings_fastapi.thing import Thing - Value = TypeVar("Value") """The value returned by the descriptor, when called on an instance.""" -Owner = TypeVar("Owner", bound="Thing") +Owner = TypeVar("Owner") """A Thing subclass that owns a descriptor.""" Descriptor = TypeVar("Descriptor", bound="BaseDescriptor") @@ -970,7 +967,7 @@ class Example: # any guarantee docstrings are available. try: src = inspect.getsource(cls) - except (OSError, AttributeError): + except (OSError, AttributeError, TypeError): # An OSError is raised if the source is not available. # An AttributeError is raised if the source was loaded from # a WindowsPath object, perhaps using ``runpy`` diff --git a/src/labthings_fastapi/client/__init__.py b/src/labthings_fastapi/client/__init__.py index 3ba01096..ca55c7f0 100644 --- a/src/labthings_fastapi/client/__init__.py +++ b/src/labthings_fastapi/client/__init__.py @@ -13,9 +13,10 @@ from urllib.parse import urljoin, urlparse import httpx -from pydantic import BaseModel, TypeAdapter, ValidationError -from typing_extensions import Self # 3.9, 3.10 compatibility +from pydantic import BaseModel, RootModel, TypeAdapter, ValidationError +from labthings_fastapi.base_descriptor import FieldTypedBaseDescriptor +from labthings_fastapi.code_generation import generate_client_class from labthings_fastapi.exceptions import ( ClientPropertyError, FailedToInvokeActionError, @@ -25,11 +26,24 @@ ) from labthings_fastapi.outputs.blob import Blob, RemoteBlobData from labthings_fastapi.problem_details import ProblemDetails, docs_url +from labthings_fastapi.thing_description._model import ThingDescription __all__ = ["ThingClient", "poll_invocation"] ACTION_RUNNING_KEYWORDS = ["idle", "pending", "running"] +def _optional() -> Any: + """Return `...` to signify an unspecified default. + + This function returns `...` but is typed as `Any` so that it + may be used as a Pydantic default factory when we don't ever + want to see the default value. + + :returns: `...` as a sentinel for a missing value. + """ + return ... + + class ObjectHasNoLinksError(KeyError): """We attempted to use the `links` key but it was not there. @@ -219,17 +233,30 @@ def invoke_action(self, path: str, **kwargs: Any) -> Any: # noqa: DOC503 :raise GlobalLockBusyError: if the action fails because of the global lock. :raise InvocationCancelledError: if the action is cancelled. """ - for k in kwargs.keys(): - value = kwargs[k] - if isinstance(value, Blob): - # Blob objects may be used as input to a subsequent - # action. When this is done, they should be serialised by - # pydantic, to a dictionary that includes href and media_type. - # - # Note that the blob will not be uploaded: we rely on the blob - # still existing on the server. - kwargs[k] = TypeAdapter(Blob).dump_python(value) - response = self.client.post(urljoin(self.path, path), json=kwargs) + # weed out parameters that are `...`: these should not be sent, so that + # the server fills in its defaults. + payload = {} + for key, value in kwargs.items(): + if value is ...: + # Items that are `...` should not be included + continue + elif hasattr(value, "__get_pydantic_core_schema__") and not isinstance( + value, BaseModel + ): + # For "pydantic-compatible" custom types, we wrap them in a + # root model, so that they're serialised using the metadata + # attached to the type. + # This is important for `Blob` instances, for example. + # Note that `RootModel` accepts expressions and variables in its + # square brackets, hence the type: ignore. + payload[key] = RootModel[type(value)](value) # type: ignore[misc,operator] + else: + # For now, we assume all other types will serialise OK + payload[key] = value + # The next line uses pydantic to serialise any models to simple types. + # We may in future serialise straight to JSON. + plain_payload = TypeAdapter(dict).dump_python(payload, exclude_unset=True) + response = self.client.post(urljoin(self.path, path), json=plain_payload) if response.is_error: message = _construct_failed_to_invoke_message(path, response) raise FailedToInvokeActionError(message) @@ -268,7 +295,9 @@ def follow_link(self, response: dict, rel: str) -> httpx.Response: return r @classmethod - def from_url(cls, thing_url: str, client: Optional[httpx.Client] = None) -> Self: + def from_url( + cls, thing_url: str, client: Optional[httpx.Client] = None + ) -> ThingClient: """Create a ThingClient from a URL. This will dynamically create a subclass with properties and actions, @@ -284,14 +313,14 @@ def from_url(cls, thing_url: str, client: Optional[httpx.Client] = None) -> Self :return: a `~lt.ThingClient` subclass with properties and methods that match the retrieved Thing Description (see :ref:`wot_thing`). """ - td_client = client or httpx + td_client = client or httpx.Client() r = td_client.get(thing_url) r.raise_for_status() subclass = cls.subclass_from_td(r.json()) return subclass(thing_url, client=client) @classmethod - def subclass_from_td(cls, thing_description: dict) -> type[Self]: + def subclass_from_td(cls, thing_description: dict) -> type[ThingClient]: """Create a ThingClient subclass from a Thing Description. Dynamically subclass `~lt.ThingClient` to add properties and @@ -303,142 +332,66 @@ def subclass_from_td(cls, thing_description: dict) -> type[Self]: :return: a `~lt.ThingClient` subclass with the right properties and methods. """ - my_thing_description = thing_description - - class Client(cls): # type: ignore[valid-type, misc] - # mypy wants the superclass to be statically type-able, but - # this isn't possible (for now) if we are to be able to - # use this class method on `ThingClient` subclasses, i.e. - # to provide customisation but also add methods from a - # Thing Description. - thing_description = my_thing_description - - for name, p in thing_description["properties"].items(): - add_property(Client, name, p) - for name, a in thing_description["actions"].items(): - add_action(Client, name, a) - return Client - - -class PropertyClientDescriptor: - """A base class for properties on `~lt.ThingClient` objects.""" - - name: str - type: type | BaseModel - path: str + parsed_td = ThingDescription.model_validate(thing_description) + client_cls = generate_client_class(parsed_td) + return client_cls Value = TypeVar("Value") -class ClientProperty(Generic[Value]): +class ClientProperty(Generic[Value], FieldTypedBaseDescriptor[ThingClient, Value]): """A descriptor to make properties of ThingClient objects work.""" def __init__( self, - name: str, - readable: bool = True, - writeable: bool = True, - doc: str | None = None, + read_only: bool = False, + write_only: bool = False, ) -> None: """Initialise a ClientProperty. - :param name: The name of the property. - :param writeable: whether the property should be writeable. + :param read_only: whether the property should be read-only. + :param write_only: whether the property should be write-only. """ - self._name = name - self._readable = readable - self._writeable = writeable - if doc: - self.__doc__ = doc - - def __get__( - self, obj: ThingClient | None, cls: type[ThingClient] | None = None - ) -> Value | Self: + super().__init__() + self.read_only = read_only + self.write_only = write_only + + def instance_get(self, obj: ThingClient) -> Value: """Retrieve the property. - :param obj: The client object on which the property is accessed. + :param obj: the client object on which the property is accessed. + :return: the value of the property. + :raises ClientPropertyError: if the property is write-only. """ - if obj is None: - return self - if self._readable: - return obj.get_property(self._name) - else: + if self.write_only: raise ClientPropertyError("This property may not be read.") + return obj.get_property(self.name) - def __set__(self, obj: ThingClient, value: Value) -> None | Self: + def __set__(self, obj: ThingClient, value: Value) -> None: """Retrieve the property. - :param obj: The client object on which the property is accessed. + :param obj: the client object on which the property is set. + :param value: the new value for the property. + :raises ClientPropertyError: if the property is read-only. """ - if self._writeable: - return obj.set_property(self._name, value) - else: + if self.read_only: raise ClientPropertyError("This property may not be set.") + return obj.set_property(self.name, value) -def client_property( - name: str, doc: str | None, writeable: bool = True, readable: bool = True -) -> Any: - return ClientProperty( - name=name, - doc=doc, - writeable=writeable, - readable=readable, - ) - - -def add_action(cls: type[ThingClient], action_name: str, action: dict) -> None: - """Add an action to a ThingClient subclass. - - A method will be added to the class that calls the provided action. - Currently, this will have a return type hint but no argument names - or type hints. - - :param cls: the `~lt.ThingClient` subclass to which we are adding the - action. - :param action_name: is both the name we assign the method to, and - the name of the action in the Thing Description. - :param action: a dictionary representing the action, in :ref:`wot_td` - format. - """ +def client_property(read_only: bool = False, write_only: bool = False) -> Any: + r"""Create a `ClientProperty` descriptor. - def action_method(self: ThingClient, **kwargs: Any) -> Any: - return self.invoke_action(action_name, **kwargs) + This function returns a `ClientProperty` and passes all parameters directly + to the constructor. It's typed as `Any` so that we can use it as a field-style + placeholder just like `lt.property`\ . - if "output" in action and "type" in action["output"]: - action_method.__annotations__["return"] = action["output"]["type"] - if "description" in action: - action_method.__doc__ = action["description"] - setattr(cls, action_name, action_method) - - -def add_property(cls: type[ThingClient], property_name: str, property: dict) -> None: - """Add a property to a ThingClient subclass. - - A descriptor will be added to the provided class that makes the - attribute ``property_name`` get and/or set the property described - by the ``property`` dictionary. - - - :param cls: the `~lt.ThingClient` subclass to which we are adding the - property. - :param property_name: is both the name we assign the descriptor to, and - the name of the property in the Thing Description. - :param property: a dictionary representing the property, in :ref:`wot_td` - format. + :param read_only: whether the property is read only. + :param write_only: whether the property is write only. + :return: a `ClientProperty` descriptor. """ - annotation = property.get("type", Any) - setattr( - cls, - property_name, - ClientProperty[annotation]( - name=property_name, - doc=property.get("description", None), - writeable=not property.get("readOnly", False), - readable=not property.get("writeOnly", False), - ), - ) + return ClientProperty(read_only=read_only, write_only=write_only) def _construct_failed_to_invoke_message(path: str, response: httpx.Response) -> str: diff --git a/src/labthings_fastapi/code_generation/__init__.py b/src/labthings_fastapi/code_generation/__init__.py index 92ec73a2..f23b9ec1 100644 --- a/src/labthings_fastapi/code_generation/__init__.py +++ b/src/labthings_fastapi/code_generation/__init__.py @@ -11,19 +11,43 @@ import ast import re -from inspect import cleandoc -from types import NoneType -from typing import Optional, Sequence +import sys +import tempfile +import warnings +from collections.abc import Mapping +from importlib.util import module_from_spec, spec_from_file_location +from pathlib import Path +from types import EllipsisType, NoneType +from typing import TYPE_CHECKING, Literal, Sequence +from uuid import uuid4 from labthings_fastapi.thing_description._model import ( DataSchema, + InteractionAffordance, ThingDescription, Type, ) +if TYPE_CHECKING: + from labthings_fastapi.client import ThingClient + def title_to_snake_case(title: str) -> str: - """Convert text to snake_case.""" + r"""Convert text to snake_case. + + Identify the words in a string (i.e. made up of letters and numbers) + and join them with underscore characters. + ``CamelCase`` words are split, so also gain underscores before each + capital letter within the word. + + This function is suitable for operation on arbitrary strings. Each + sequence of non-word characters within the string will be converted + to a single underscore. Initial or final non-word characters are + ignored. + + :param title: the ``CamelCase`` text to convert. + :return: the text, in ``snake_case``\ . + """ # First, look for CamelCase so it doesn't get ignored: uncameled = re.sub(r"([a-z0-9])([A-Z])", r"\1_\2", title) words = re.findall(r"[a-zA-Z0-9]+", uncameled) @@ -31,57 +55,86 @@ def title_to_snake_case(title: str) -> str: def snake_to_camel_case(snake: str) -> str: - """Convert snake_case to CamelCase.""" + r"""Convert snake_case to CamelCase. + + :param snake: the ``snake_case`` text to convert. + :return: the text, in ``CamelCase``\ . + """ words = snake.split("_") return "".join(word.capitalize() for word in words) def title_to_camel_case(title: str) -> str: - """Convert text to CamelCase.""" - return snake_to_camel_case(title_to_snake_case(title)) + r"""Convert text to CamelCase. + :param title: the text, in title case. + :return: the text, in ``camel_case``\ . + """ + return snake_to_camel_case(title_to_snake_case(title)) -def clean_code(code: str, prefix: str = "") -> str: - """Clean up code by removing leading/trailing whitespace and empty lines.""" - lines = cleandoc(code).split("\n") - return "\n".join([prefix + line for line in lines]) +def _name(name: str, ctx: Literal["load", "store"] = "load") -> ast.Name: + r"""Generate a Name object (shorthand for `ast.Name`). -def quoted_docstring(docstring: Optional[str], indent: int = 4) -> str: - """Wrap a docstring in triple quotes.""" - if docstring is None: - return "" - prefix = " " * indent - lines = docstring.split("\n") - lines[0] = f'"""{lines[0]}' - lines.append('"""') - return "".join([f"{prefix}{line}\n" for line in lines]) + Return an `ast.Name` object, with ``id`` set to the ``name`` + argument and the context set to ``ast.Load()`` or ``ast.Store()`` as + appropriate. + This is primarily provided as a shorthand, to simplify some of the longer + expressions in code generation. -def _name(name: str) -> ast.Name: - """Generate a Name object (shorthand for `ast.Name`).""" - return ast.Name(id=name, ctx=ast.Load()) + :param name: the string to be used as a name. + :param ctx: the context (``"load"`` or ``"store"``). + :return: an `ast.Name` object. + """ + return ast.Name(id=name, ctx=ast.Store() if ctx == "store" else ast.Load()) def _subscript(value: ast.expr, subscript: ast.expr) -> ast.Subscript: - """Generate a Subscript object (shorthand for `ast.Subscript`).""" + """Generate a Subscript object (shorthand for `ast.Subscript`). + + :param value: the quantity before the square brackets. + :param subscript: the quantity in the square brackets. + :return: an `ast.Subscript` object, using the "load" context. + """ return ast.Subscript(value, subscript, ctx=ast.Load()) def _tuple(*elements: ast.expr) -> ast.Tuple: - """Generate a Tuple (shorthand for `ast.Tuple`).""" + r"""Generate a Tuple (shorthand for `ast.Tuple`). + + This function provides a convenient shorthand for `ast.Tuple`\ . + + :param \*elements: positional arguments become the elements of the tuple. + :return: an `ast.Tuple` instance, using the "load" context. + """ return ast.Tuple(elts=list(elements), ctx=ast.Load()) -def _constant(value: str | bool | int | float | NoneType) -> ast.Constant: - """Check a value may be rendered as a constant, and return it.""" - if not isinstance(value, (str, bool, int, float, NoneType)): +def _constant(value: str | bool | int | float | EllipsisType | None) -> ast.Constant: + """Check a value may be rendered as a constant, and return it. + + This is shorthand for `ast.Constant` and also includes a runtime type check. + + :param value: the value of the constant. + :return: an `ast.Constant` wrapping the value. + :raises TypeError: if the type of ``value`` can't be a constant. + """ + if not isinstance(value, (str, bool, int, float, NoneType, EllipsisType)): raise TypeError("Don't know how to return this as a constant!") return ast.Constant(value=value) def _import_from(module: str, names: list[str]) -> ast.ImportFrom: - """Import names from a module.""" + r"""Import names from a module. + + This is shorthand for an `ast.ImportFrom` object. It should be + a safer equivalent to ``f"from {module} import {", ".join(names)}"``\ . + + :param module: the module to import from + :param names: the names to import + :return: an `ast.ImportFrom` object. + """ return ast.ImportFrom( module=module, names=[ast.alias(name=name) for name in names], @@ -89,182 +142,341 @@ def _import_from(module: str, names: list[str]) -> ast.ImportFrom: ) -def dataschema_to_type( - schema: DataSchema, models: list[ast.ClassDef], name: str = "anonymous" -) -> ast.expr: - """Convert a DataSchema to a Python type.""" - if isinstance(schema.oneOf, Sequence) and len(schema.oneOf) > 0: - types = [dataschema_to_type(s, models) for s in schema.oneOf] - return _subscript(_name("Union"), _tuple(*types)) - if schema.type == Type.string: - return _name("str") - elif schema.type == Type.integer: - return _name("int") - elif schema.type == Type.number: - return _name("float") - elif schema.type == Type.boolean: - return _name("bool") - elif schema.type == Type.array: - if schema.items is None: - # A dataschema with no `items` member gets typed as a plain list - return _name("list") - if isinstance(schema.items, Sequence): - # The WoT spec is based on JSONSchema 2019 - # If `items` is a sequence, it behaves as `prefixItems` in the 2023 - # draft, i.e. it describes a tuple. - # https://json-schema.org/understanding-json-schema/reference/array#tupleValidation - types = [dataschema_to_type(s, models) for s in schema.items] - return _subscript(_name("tuple"), _tuple(*types)) - return _subscript(_name("list"), dataschema_to_type(schema.items, models)) - elif schema.type == Type.object: - # If the object has no properties, return a generic dict - if not schema.properties: - return _subscript(_name("dict"), _name("Any")) - # Objects with properties are converted to Pydantic models - if schema.title: - model_name = title_to_camel_case(schema.title + "_model") - else: - model_name = snake_to_camel_case(name + "_model") - model_names = [m.name for m in models] - if model_name in model_names: - i = 0 - while f"{model_name}_{i}" in models: - i += 1 - model_name = f"{model_name}_{i}" - models.append(dataschema_to_model(schema, models, model_name)) - return _name(model_name) - else: +NO_ARGUMENTS = ast.arguments( + posonlyargs=[], + args=[ast.arg("self")], + vararg=None, + kwonlyargs=[], + kwarg=ast.arg("kwargs", _name("Any")), + defaults=[], + kw_defaults=[], +) +"""A shortcut to specify that a function takes no arguments.""" + + +class DataSchemaConverter: + """A class that converts `DataSchema` objects to Python types.""" + + def __init__(self, max_depth: int = 10, warn_on_fallback: bool = True) -> None: + """Initialise a DataSchemaConverter. + + :param max_depth: the maximum recursion depth. Set this to zero to disable + the generation of Pydantic models for ``"object"`` types, unless they + are at top level. + :param warn_on_fallback: whether to raise a warning when we return `typing.Any` + if a schema can't be converted. + """ + self.model_definitions: list[ast.ClassDef] = [] + self.max_depth = max_depth + self.warn_on_fallback = warn_on_fallback + + def convert(self, schema: DataSchema, recursion_depth: int = 0) -> ast.expr: + r"""Convert a DataSchema to a Python type. + + This maps types described by a Thing Description `DataSchema` onto Python + types, returning an `ast.expr` object suitable for use as an annotation. + + The value of the ``type`` property of the ``schema`` (which is a string) + determines the type we generate: + + ``"string"`` + is mapped to the `str` builtin. + + ``"integer"`` + is mapped to the `int` builtin. + + ``"number"`` + is mapped to the `float` builtin. + + ``"boolean"`` + is mapped to the `bool` builtin. + + ``"null"`` + is mapped to the `None` builtin. + + ``"array"`` + is mapped to a `list` or a `tuple`\ . As Thing Description is based on + the 2019 draft of JSONSchema, there are three possible ways to describe + an array: + + * If the array has no ``items`` keyword, it is mapped to a plain `list`\ . + * If the array has an ``items`` keyword that is a `DataSchema`\ , it will be + rendered as ``list[T]`` where ``T`` is generated from the ``items`` value + (recursively). + * If the array has an ``items`` keyword that is a sequence of `DataSchema` + objects, it is rendered as a tuple, where the type of each element is + given by the corresponding element of the ``items`` value. This has been + changed in the most recent draft of JSONSchema. See + `Tuple Validation `_ + + ``"object"`` + is either mapped to `dict` or to a generated `pydantic.BaseModel` subclass, + depending on whether the ``properties`` key is set. + + :param schema: the `DataSchema` to convert. + :param recursion_depth: how many levels of recursion we've currently worked + down. + :return: an expression that may be used as a type annotation. + + .. _tuple_validation: https://json-schema.org/understanding-json-schema/reference/array#tupleValidation + """ + if recursion_depth > self.max_depth: + # If we've exceeded the maximum recursion depth, fall back to `Any` + warnings.warn( + f"Recursion depth exceeded, turning {schema} into `Any`", + stacklevel=1, + ) + return _name("Any") + if isinstance(schema.oneOf, Sequence) and len(schema.oneOf) > 0: + types = [self.convert(s) for s in schema.oneOf] + return _subscript(_name("Union"), _tuple(*types)) + if schema.type == Type.string: + return _name("str") + elif schema.type == Type.integer: + return _name("int") + elif schema.type == Type.number: + return _name("float") + elif schema.type == Type.boolean: + return _name("bool") + elif schema.type == Type.null: + return _name("None") + elif schema.type == Type.array: + if isinstance(schema.items, Sequence): + # The WoT spec is based on JSONSchema 2019 + # If `items` is a sequence, it behaves as `prefixItems` in the 2023 + # draft, i.e. it describes a tuple. + # https://json-schema.org/understanding-json-schema/reference/array#tupleValidation + types = [self.convert(s, recursion_depth + 1) for s in schema.items] + return _subscript(_name("tuple"), _tuple(*types)) + if isinstance(schema.items, DataSchema): + # If a single item type is specified (including if it's a union), return + # a specifically-typed list. + return _subscript( + _name("list"), self.convert(schema.items, recursion_depth + 1) + ) + if schema.items is None: + # A dataschema with no `items` member gets typed as a plain list + return _name("list") + elif schema.type == Type.object: + # If the object has no properties, return a generic dict + if isinstance(schema.properties, Mapping): + return self.convert_to_model(schema, recursion_depth=recursion_depth) + return _name("dict") + if self.warn_on_fallback: + warnings.warn( + f"Could not convert {schema} to a Python type, falling back to `Any`.", + stacklevel=1, + ) return _name("Any") - -def dataschema_to_model( - schema: DataSchema, models: list[ast.ClassDef], name: str -) -> ast.ClassDef: - """Convert a DataSchema to a Pydantic model.""" - if schema.properties is None: - msg = f"Can't generate a model from schema '{schema}' as it has no properties." - raise TypeError(msg) - - # The class body will consist of one line setting `_model_config` and - # then one annotated assignment for each property. - class_body: list[ast.stmt] = [] - class_body.append( - ast.Assign( - targets=[ast.Name(id="_model_config", ctx=ast.Store())], - value=ast.Dict(keys=[_constant("extra")], values=[_constant("allow")]), - ) - ) - for pname, prop in schema.properties.items(): - # The fields of the model will appear as assignments. If there's no default, - # it's still an assignment as far as `ast` is concerned, but there's no - # value (or equals sign). - has_default = "default" in prop.model_fields_set + def convert_to_model(self, schema: DataSchema, recursion_depth: int) -> ast.Name: + """Convert a DataSchema to a Pydantic model. + + This will iterate through the required and optional ``properties`` of the + `DataSchema` and build a class definition for a `pydantic.BaseModel` subclass. + + We will then check to see if an equivalent model has been built already: if so, + we will reference it. We built the model first, so that we can check based on + the generated model, rather than the input schema or the name. + + If the model we've just built is unique, we'll add it to + ``self.model_definitions`` with a unique name, and return a reference to the + new model. + + :param schema: the DataSchema instance. + :param recursion_depth: how deep we are within a nested type. + :return: an `ast.Name` object referencing the generated model. + :raises TypeError: if there is no ``properties`` field. + :raises KeyError: if the schema requires a property that doesn't exist in + ``properties``. + """ + if schema.properties is None: + msg = f"Can't generate a model from schema '{schema}' without properties." + raise TypeError(msg) + + # The class body will consist of a docstring, + # one line setting `_model_config` and + # one annotated assignment for each property. + class_body: list[ast.stmt] = [] + if schema.description: # Add a docstring, if there is a description + class_body.append(ast.Expr(_constant(schema.description))) class_body.append( - ast.AnnAssign( - target=ast.Name(id=pname, ctx=ast.Store()), - annotation=dataschema_to_type(prop, models, name), - value=_constant(prop.default) if has_default else None, - simple=1, + ast.Assign( + targets=[_name("_model_config", "store")], + value=ast.Dict(keys=[_constant("extra")], values=[_constant("allow")]), ) ) - return ast.ClassDef( - name=name, - bases=[_name("BaseModel")], - keywords=[], - body=class_body, - decorator_list=[], - ) - + # Start with the required properties - these look like ``pname: type`` + required = schema.required or [] + for pname in required: + try: + prop = schema.properties[pname] + except KeyError as e: + msg = f"Schema requires a non-existent property '{pname}': {schema}" + raise KeyError(msg) from e + class_body.append( + ast.AnnAssign( # Note there is no value, because it's a required field. + target=_name(pname, "store"), + annotation=self.convert(prop, recursion_depth + 1), + simple=1, + ) + ) + # Now the optional ones: these look like + # ``pname: type = Field(default_factory=optional)`` + # Note that ``optional`` will return `...` so we must always serialise with + # ``exclude_unset=True`` or we'll get an error. + for pname, prop in schema.properties.items(): + if pname in required: + continue # we already dealt with required properties above. + class_body.append( + ast.AnnAssign( + target=_name(pname, "store"), + annotation=self.convert(prop, recursion_depth + 1), + value=ast.Call( + _name("Field"), + [], + [ast.keyword("default_factory", _name("_optional"))], + ), + simple=1, + ) + ) -def property_to_argument( - name: str, - property: DataSchema, - models: list[ast.ClassDef], -) -> tuple[ast.arg, ast.expr | None]: - """Convert a property to a function argument and default.""" - dtype = dataschema_to_type(property, models, name) - arg = ast.arg(arg=name, annotation=dtype) - if "default" in property.model_fields_set: - if property.default is None: - default = ast.Constant(None) - elif isinstance(property.default, (str, bool, int, float)): - default = ast.Constant(property.default) + # Deduplicate: if a model with this body already exists, return that name. + existing_name = self.find_model_definition_by_body(class_body) + if existing_name: + return existing_name + + # We need to make a new model definition + name = self.make_unique_model_name(schema) + self.model_definitions.append( + ast.ClassDef( + name=name, + bases=[_name("BaseModel")], + keywords=[], + body=class_body, + decorator_list=[], + ) + ) + return _name(name) + + def find_model_definition_by_body(self, body: list[ast.stmt]) -> ast.Name | None: + r"""Locate a model definition with the supplied class body, or return `None`\ . + + :param body: the body of the model definition. + :return: the name, as an `ast.Name` object, or `None` if the supplied body is + new. + """ + for model in self.model_definitions: + if model.body == body: + return _name(model.name) + return None + + def make_unique_model_name( + self, schema: DataSchema, name: str = "UnnamedModel" + ) -> str: + """Generate a unique name for a schema. + + :param schema: the schema we're naming. + :param name: a name for the schema, used if it has no title. + :return: a unique name. + """ + if schema.title: + model_name = title_to_camel_case(schema.title + "_model") else: - raise NotImplementedError(f"Unsupported default value: {property.default}") - else: - default = None - return arg, default + model_name = title_to_camel_case(name) + model_names = [m.name for m in self.model_definitions] + # Append a number to the model name until it's unique. + i = 0 + while model_name in model_names: + i += 1 + model_name = f"{model_name}{i}" + return model_name + + def input_model_to_arguments(self, model: DataSchema) -> ast.arguments: + """Convert an input model to an arguments object. + + :param model: A DataSchema describing the input (of an action). + :return: an arguments object describing the properties of `model` as + keyword-only arguments. + + :raises TypeError: if the model doesn't describe an object. + """ + if model.type is None: + return NO_ARGUMENTS + if model.type != Type.object: + print(f"model.type: {model.type}") + raise TypeError("Only object models are supported") + if not model.properties: + return NO_ARGUMENTS + kwargs = [] + kwdefaults: list[ast.expr | None] = [] + # Make sure required (i.e. no default) arguments come first. + arg_names = list(model.properties.keys()) + required_names = model.required or [] + for name in required_names: + arg_names.remove(name) + arg_names = required_names + arg_names + for pname, property in model.properties.items(): + kwargs.append(ast.arg(pname, self.convert(property))) + kwdefaults.append(None if pname in required_names else _constant(...)) + return ast.arguments( + kwonlyargs=kwargs, + kw_defaults=kwdefaults, + posonlyargs=[ast.arg("self")], + args=[], + vararg=None, + kwarg=ast.arg("kwargs", _name("Any")), + defaults=[], + ) -NO_ARGUMENTS = ast.arguments( - posonlyargs=[], - args=[], - vararg=None, - kwonlyargs=[], - kwarg=None, - defaults=[], - kw_defaults=[], -) +def _affordance_docstring(affordance: InteractionAffordance) -> str | None: + r"""Extract a docstring from an affordance description. + :param affordance: the affordance description. + :return: a docstring or `None`\ . + """ + if affordance.description or affordance.title: + if affordance.description and affordance.title: + return f"{affordance.title}\n\n{affordance.description}" + else: + return affordance.title or affordance.description + return None -def input_model_to_arguments( - model: DataSchema, models: list[ast.ClassDef] -) -> ast.arguments: - """Convert an input model to an arguments object. - :param model: A DataSchema describing the input (of an action). - :param models: A dictionary of Pydantic model definitions we'll need. - :return: an arguments object describing the properties of `model` as - keyword-only arguments. +def _append_docstring_if_present( + body: list[ast.stmt], affordance: InteractionAffordance +) -> None: + """Append a docstring, if relevant information is included. + + :param body: a list of expressions to append to. + :param affordance: an affordance to built the docstring from. """ - if model.type is None: - return NO_ARGUMENTS - if model.type != Type.object: - print(f"model.type: {model.type}") - raise TypeError("Only object models are supported") - if not model.properties: - return NO_ARGUMENTS - kwargs = [] - kwdefaults = [] - # Make sure required (i.e. no default) arguments come first. - arg_names = list(model.properties.keys()) - required_names = model.required or [] - for name in required_names: - arg_names.remove(name) - arg_names = required_names + arg_names - for name, property in model.properties.items(): - property = model.properties[name] - arg, default = property_to_argument(name, property, models) - kwargs.append(arg) - # It's possible for required args to have defaults in the schema, - # but we remove these from the method signature. - kwdefaults.append(None if name in required_names else default) - return ast.arguments( - kwonlyargs=kwargs, - kw_defaults=kwdefaults, - posonlyargs=[], - args=[], - vararg=None, - kwarg=None, - defaults=[], - ) + doc = _affordance_docstring(affordance) + if doc: + body.append(ast.Expr(_constant(doc))) -def generate_client(thing_description: ThingDescription) -> ast.Module: +def generate_client_ast(thing_description: ThingDescription) -> tuple[str, ast.Module]: """Generate a client from a Thing Description. - + + This function returns an abstract syntax tree (which may either be executed + directly, or dumped to a module on disk) and the name of an autogenerated client + class. + :param thing_description: the Thing Description to use. + :return: a tuple of a class name and the abstract syntax tree of a Python module. """ - # We'll populate this dictionary with auto-generated Pydantic models, - # needed for `DataSchema`s that are typed as `object` - models: list[ast.ClassDef] = [] - - class_name = title_to_camel_case(thing_description.title) + class_name = title_to_camel_case(thing_description.title) + "Client" - # We create the body of the class as a list of statments. This will be + # We create the body of the class as a list of statements. This will be # included in the class definition later. class_body: list[ast.stmt] = [] + # The converter is responsible for turning DataSchema into types, and + # keeping track of any models we will need to define. + converter = DataSchemaConverter() + if thing_description.description: doc = thing_description.description else: @@ -277,28 +489,27 @@ def generate_client(thing_description: ThingDescription) -> ast.Module: properties = thing_description.properties or {} for name, property in properties.items(): pname = title_to_snake_case(name) - dtype = dataschema_to_type(property, models=models) + dtype = converter.convert(property) class_body.append( ast.AnnAssign( - target=ast.Name(id=name, ctx=ast.Store()), + target=_name(pname, "store"), annotation=dtype, value=ast.Call( func=_name("client_property"), args=[], keywords=[ - ast.keyword(arg="name", value=_constant(pname)), - ast.keyword(arg="doc", value=_constant(property.description)), ast.keyword( - arg="readable", value=_constant(not property.readOnly) + arg="read_only", value=_constant(property.readOnly) ), ast.keyword( - arg="writeable", value=_constant(not property.writeOnly) + arg="write_only", value=_constant(property.writeOnly) ), ], ), simple=1, ) ) + _append_docstring_if_present(class_body, property) # Each action will be a method definition, with appropriately typed arguments and # return values. These are then passed straight through to `self.invoke_action` @@ -306,15 +517,17 @@ def generate_client(thing_description: ThingDescription) -> ast.Module: for name, action in actions.items(): aname = title_to_snake_case(name) if action.input: - args = input_model_to_arguments(action.input, models) + args = converter.input_model_to_arguments(action.input) else: args = NO_ARGUMENTS if action.output: - rtype = dataschema_to_type(action.output, models) + rtype = converter.convert(action.output) else: rtype = _name("Any") # The function body simply passes the arguments through to `invoke_action`. - function_body: list[ast.stmt] = [ + function_body: list[ast.stmt] = [] + _append_docstring_if_present(function_body, action) + function_body.append( ast.Return( value=ast.Call( func=ast.Attribute( @@ -324,10 +537,13 @@ def generate_client(thing_description: ThingDescription) -> ast.Module: keywords=[ ast.keyword(arg=arg.arg, value=_name(arg.arg)) for arg in args.kwonlyargs + ] + + [ + ast.keyword(value=_name("kwargs")) # pass additional kwargs ], ) ) - ] + ) class_body.append( ast.FunctionDef( name=aname, @@ -341,24 +557,78 @@ def generate_client(thing_description: ThingDescription) -> ast.Module: # The class definition is here: this includes `class_body` defined above, with the # actions/properties. class_definition = ast.ClassDef( - name=f"{class_name}Client", + name=class_name, bases=[_name("ThingClient")], keywords=[], body=class_body, decorator_list=[], ) - # The module we want to create starts here: + # The module we want to create starts here, with the module docstring: + code: list[ast.stmt] = [ + ast.Expr(_constant(f"A client for the {thing_description.title} Thing.")) + ] # Import the symbols we'll need - code: list[ast.stmt] = [] code.append( - _import_from("labthings_fastapi.client", ["ThingClient", "client_property"]) + _import_from( + "labthings_fastapi.client", + ["ThingClient", "client_property", "_optional"], + ) ) code.append(_import_from("typing", ["Any", "Union"])) - code.append(_import_from("pydantic", ["BaseModel"])) + code.append(_import_from("pydantic", ["BaseModel", "Field"])) - code += models + code += converter.model_definitions code.append(class_definition) - return ast.Module(body=code, type_ignores=[]) + tree = ast.Module(body=code, type_ignores=[]) + # location attributes would point to lines in source code, if they existed. + # these are left as None, which causes errors later on - the line below + # populates them with dummy data so we can, for example, use `ast.unparse` + # or `compile`. + ast.fix_missing_locations(tree) + return class_name, tree + + +def generate_client_class(thing_description: ThingDescription) -> "type[ThingClient]": + """Generate a `~lt.ThingClient` subclass for a particular Thing. + + :param thing_description: the description of the Thing we're working with. + :return: a client class for that particular Thing. + :raises RuntimeError: if the generated module can't be loaded. + """ + cls_name, tree = generate_client_ast(thing_description) + # Now, use `ast.unparse` to write this to a file, which we can then import. + with tempfile.TemporaryDirectory() as dir: + fname = Path(dir) / "generated_client.py" + with open(fname, mode="w") as f: + f.write(ast.unparse(tree)) + # We import the dynamically-generated code to create a class. + # Running dynamically-generated code is potentially dangerous, if user input + # could lead to code execution. + # + # The code that generates this + # client class is very careful only to pass through user values as (1) names or + # (2) string constants. The use of `ast` to do this ensures that we don't fall + # foul of string escaping issues, and names are constrained only ever to work as + # names. + module_name = f"labthings_fastapi.generated_clients.{cls_name}.{uuid4()}" + spec = spec_from_file_location(module_name, fname) + if spec is None: + raise RuntimeError("Couldn't generate a spec to load the generated module.") + module = module_from_spec(spec) + if spec.loader is None: + raise RuntimeError("Spec.loader was None when autogenerating a client.") + # Temporarily add the module to `sys.modules` so that `BaseDescriptor` can + # retrieve the docstrings. + sys.modules[module_name] = module + try: + spec.loader.exec_module(module) + finally: + del sys.modules[module_name] + # Note that we need to keep the temp file until after executing the module, + # so the file still exists. It will be deleted at the end of this indented + # block. + # Finally, access the class in the now-executed module. + return getattr(module, cls_name) diff --git a/tests/test_client_generation.py b/tests/test_client_generation.py index 8853081d..f2e9f228 100644 --- a/tests/test_client_generation.py +++ b/tests/test_client_generation.py @@ -7,7 +7,10 @@ import labthings_fastapi as lt import labthings_fastapi.code_generation as cg -from labthings_fastapi.code_generation import generate_client +from labthings_fastapi.code_generation import ( + generate_client_ast, + generate_client_class, +) from labthings_fastapi.example_things import MyThing from labthings_fastapi.testing import create_thing_without_server @@ -29,46 +32,50 @@ def test_snake_to_camel_case(): def generate_and_verify(thing): td = create_thing_without_server(thing).thing_description() - tree = generate_client(td) + cls_name, tree = generate_client_ast(td) ast.fix_missing_locations(tree) code = ast.unparse(tree) with tempfile.TemporaryDirectory() as d: fname = os.path.join(d, "client.py") with open(fname, "w") as f: f.write(code) - spec = importlib.util.spec_from_file_location("client", f.name) + spec = importlib.util.spec_from_file_location( + f"labthings_fastapi.generated_client.{cls_name}", f.name + ) assert spec is not None module = importlib.util.module_from_spec(spec) assert spec.loader is not None spec.loader.exec_module(module) - assert f"{thing.__name__}Client" in dir(module) + assert cls_name in dir(module) + cls = getattr(module, cls_name) + assert issubclass(cls, lt.ThingClient) def test_mything_generation(): generate_and_verify(MyThing) -class TestModel(BaseModel): +class SimpleModel(BaseModel): a: int b: str class NestedModel(BaseModel): - c: TestModel + c: SimpleModel class ThingWithModels(lt.Thing): @lt.property - def prop1(self) -> TestModel: - return TestModel(a=1, b="test") + def prop1(self) -> SimpleModel: + return SimpleModel(a=1, b="test") @lt.action - def action1(self, arg1: TestModel) -> TestModel: + def action1(self, arg1: SimpleModel) -> SimpleModel: return arg1 @lt.property def prop2(self) -> NestedModel: - return NestedModel(c=TestModel(a=1, b="test")) + return NestedModel(c=SimpleModel(a=1, b="test")) def test_with_models(): @@ -80,8 +87,10 @@ def test_with_models(): print("Thing Description:") print(td.model_dump_json(indent=2, exclude_unset=True)) print("\nGenerated AST:") - ast_module = generate_client(td) + _, ast_module = generate_client_ast(td) print(ast.dump(ast_module, indent=4)) print("\nGenerated module:") - ast.fix_missing_locations(ast_module) print(ast.unparse(ast_module)) + with open("scratch/generated_client_test.py", "w") as f: + f.write(ast.unparse(ast_module)) + cls = generate_client_class(td) diff --git a/tests/test_thing_client.py b/tests/test_thing_client.py index 69a473c2..b34be5c9 100644 --- a/tests/test_thing_client.py +++ b/tests/test_thing_client.py @@ -161,21 +161,24 @@ def test_property_descriptor_errors(mocker): "writeOnly": False, "readOnly": False, "type": "integer", - "forms": [], + "forms": [ + {"op": "readproperty", "href": "http://whatever/"}, + {"op": "writeproperty", "href": "http://whatever/"}, + ], }, "readonly": { "title": "test", "writeOnly": False, "readOnly": True, "type": "integer", - "forms": [], + "forms": [{"op": "readproperty", "href": "http://whatever/"}], }, "writeonly": { "title": "test", "writeOnly": True, "readOnly": False, "type": "integer", - "forms": [], + "forms": [{"op": "writeproperty", "href": "http://whatever/"}], }, }, "actions": {}, @@ -245,12 +248,23 @@ def test_call_action_with_args_and_return(thing_client_and_thing): assert thing.int_prop == 6 -def test_call_action_wrong_arg(thing_client): - """Test calling an action with wrong argument.""" - err = "Error when invoking action increment_by_input: 'value' - Field required" +def test_call_action_wrong_arg(mocker, thing_client): + """Test calling an action with wrong argument. + This should pick up a missing argument before sending anything to + the server - but if we force it to send bad input to the server, it should + relay the validation error. + """ + # First, check that missing arguments raise a TypeError + mocker.spy(thing_client, "invoke_action") + err = "missing 1 required keyword-only argument: 'value'" + with pytest.raises(TypeError, match=err): + thing_client.increment_by_input() + assert thing_client.invoke_action.call_count == 0 # No request was made + + err = "Error when invoking action increment_by_input: 'value' - Field required" with pytest.raises(FailedToInvokeActionError, match=err): - thing_client.increment_by_input(input=5) + thing_client.invoke_action("increment_by_input", input=5) def test_call_action_wrong_type(thing_client): From ef98766436e54aae9110116a1f885922d3aa6be1 Mon Sep 17 00:00:00 2001 From: Richard Bowman Date: Thu, 11 Jun 2026 00:09:58 +0100 Subject: [PATCH 5/9] Fix typing for python > 3.11 and move module. This moves the single file out of the folder, keeping the module name the same. I've also provided the `type_params` argument on Python > 3.11. --- .../__init__.py => code_generation.py} | 71 ++++++++++++++++--- 1 file changed, 63 insertions(+), 8 deletions(-) rename src/labthings_fastapi/{code_generation/__init__.py => code_generation.py} (93%) diff --git a/src/labthings_fastapi/code_generation/__init__.py b/src/labthings_fastapi/code_generation.py similarity index 93% rename from src/labthings_fastapi/code_generation/__init__.py rename to src/labthings_fastapi/code_generation.py index f23b9ec1..e861592d 100644 --- a/src/labthings_fastapi/code_generation/__init__.py +++ b/src/labthings_fastapi/code_generation.py @@ -125,6 +125,66 @@ def _constant(value: str | bool | int | float | EllipsisType | None) -> ast.Cons return ast.Constant(value=value) +def _class(name: str, bases: list[ast.expr], body: list[ast.stmt]) -> ast.ClassDef: + """Define a new class. + + :param name: the name of the new class. + :param bases: the base class(es). + :param body: the body of the class definition. + :return: an `ast` class definition. + """ + if sys.version_info < (3, 12): + return ast.ClassDef( + name=name, + bases=bases, + keywords=[], + body=body, + decorator_list=[], + ) + else: + # Since 3.12, we need to add type_params + return ast.ClassDef( + name=name, + bases=bases, + keywords=[], + body=body, + decorator_list=[], + type_params=[], + ) + + +def _function( + name: str, args: ast.arguments, body: list[ast.stmt], returns: ast.expr | None +) -> ast.FunctionDef: + """Define a new function. + + :param name: the name of the function. + :param args: the arguments. + :param body: the function body, as a list of statements. + :param returns: a return type annotation. + :return: a function definition object. + """ + if sys.version_info < (3, 12): + return ast.FunctionDef( + name=name, + args=args, + body=body, + decorator_list=[], + returns=returns, + type_comment=None, + ) + else: + return ast.FunctionDef( + name=name, + args=args, + body=body, + decorator_list=[], + returns=returns, + type_comment=None, + type_params=[], + ) + + def _import_from(module: str, names: list[str]) -> ast.ImportFrom: r"""Import names from a module. @@ -349,12 +409,10 @@ def convert_to_model(self, schema: DataSchema, recursion_depth: int) -> ast.Name # We need to make a new model definition name = self.make_unique_model_name(schema) self.model_definitions.append( - ast.ClassDef( + _class( name=name, bases=[_name("BaseModel")], - keywords=[], body=class_body, - decorator_list=[], ) ) return _name(name) @@ -545,23 +603,20 @@ def generate_client_ast(thing_description: ThingDescription) -> tuple[str, ast.M ) ) class_body.append( - ast.FunctionDef( + _function( name=aname, args=args, body=function_body, - decorator_list=[], returns=rtype, ) ) # The class definition is here: this includes `class_body` defined above, with the # actions/properties. - class_definition = ast.ClassDef( + class_definition = _class( name=class_name, bases=[_name("ThingClient")], - keywords=[], body=class_body, - decorator_list=[], ) # The module we want to create starts here, with the module docstring: From 9c10f9a457dcc8c8a2bc4f9c6f9cfbfae514583c Mon Sep 17 00:00:00 2001 From: Richard Bowman Date: Thu, 11 Jun 2026 00:21:17 +0100 Subject: [PATCH 6/9] Deduplicated generated models This compares unparsed strings rather than AST nodes, as the latter don't properly show as equal. `ast.compare` will help here, but not until we drop a few Python versions. --- src/labthings_fastapi/code_generation.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/labthings_fastapi/code_generation.py b/src/labthings_fastapi/code_generation.py index e861592d..caafe5e7 100644 --- a/src/labthings_fastapi/code_generation.py +++ b/src/labthings_fastapi/code_generation.py @@ -415,6 +415,7 @@ def convert_to_model(self, schema: DataSchema, recursion_depth: int) -> ast.Name body=class_body, ) ) + ast.fix_missing_locations(self.model_definitions[-1]) return _name(name) def find_model_definition_by_body(self, body: list[ast.stmt]) -> ast.Name | None: @@ -424,8 +425,12 @@ def find_model_definition_by_body(self, body: list[ast.stmt]) -> ast.Name | None :return: the name, as an `ast.Name` object, or `None` if the supplied body is new. """ + for statement in body: + ast.fix_missing_locations(statement) + body_statements = [ast.unparse(s) for s in body] for model in self.model_definitions: - if model.body == body: + model_statements = [ast.unparse(s) for s in model.body] + if model_statements == body_statements: return _name(model.name) return None From 7fe91b9b2629818ceb0fafe3be7d3cc364402e3b Mon Sep 17 00:00:00 2001 From: Richard Bowman Date: Thu, 11 Jun 2026 10:46:42 +0100 Subject: [PATCH 7/9] Don't duplicate one-line docstrings One-line docstrings end up in both the `title` and `description` fields. We now deduplicate this, so the generated docstrings match the originals. --- src/labthings_fastapi/code_generation.py | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/src/labthings_fastapi/code_generation.py b/src/labthings_fastapi/code_generation.py index caafe5e7..ba84ab94 100644 --- a/src/labthings_fastapi/code_generation.py +++ b/src/labthings_fastapi/code_generation.py @@ -499,12 +499,11 @@ def _affordance_docstring(affordance: InteractionAffordance) -> str | None: :param affordance: the affordance description. :return: a docstring or `None`\ . """ - if affordance.description or affordance.title: - if affordance.description and affordance.title: + if affordance.description and affordance.title: + if not affordance.description.startswith(affordance.title): + # If there's a title and description, combine them. return f"{affordance.title}\n\n{affordance.description}" - else: - return affordance.title or affordance.description - return None + return affordance.title or affordance.description def _append_docstring_if_present( From c338c3bbcaf79624d57b1c8e7d9a3c5755d35b61 Mon Sep 17 00:00:00 2001 From: Richard Bowman Date: Thu, 11 Jun 2026 12:30:35 +0100 Subject: [PATCH 8/9] Add a function to generate a module on disk. This executes `ruff` in a subprocess, which `ruff` flags as potentially insecure. I've thought through this carefully and I'm happy we can ignore the warning, as I don't see any opportunity to inject malicious input. --- src/labthings_fastapi/code_generation.py | 103 +++++++++++++++++++++-- 1 file changed, 97 insertions(+), 6 deletions(-) diff --git a/src/labthings_fastapi/code_generation.py b/src/labthings_fastapi/code_generation.py index ba84ab94..eb30877f 100644 --- a/src/labthings_fastapi/code_generation.py +++ b/src/labthings_fastapi/code_generation.py @@ -9,6 +9,9 @@ a module to a file, allowing static analysis. """ +# Ruff flags subprocess as insecure: there's a comment next to where +# `run` is used explaining reasoning on security. That requires us to +# ignore the S404 code when we import from subprocess. import ast import re import sys @@ -17,7 +20,8 @@ from collections.abc import Mapping from importlib.util import module_from_spec, spec_from_file_location from pathlib import Path -from types import EllipsisType, NoneType +from subprocess import DEVNULL, run # noqa: S404 +from types import EllipsisType, ModuleType, NoneType from typing import TYPE_CHECKING, Literal, Sequence from uuid import uuid4 @@ -28,6 +32,15 @@ Type, ) +# We have a soft dependency on Ruff. +try: + import ruff # type: ignore[import-untyped] + + RUFF_BINARY = ruff.find_ruff_bin() +except ImportError: + RUFF_BINARY = None + + if TYPE_CHECKING: from labthings_fastapi.client import ThingClient @@ -650,14 +663,19 @@ def generate_client_ast(thing_description: ThingDescription) -> tuple[str, ast.M return class_name, tree -def generate_client_class(thing_description: ThingDescription) -> "type[ThingClient]": - """Generate a `~lt.ThingClient` subclass for a particular Thing. +def load_ast_as_module(cls_name: str, tree: ast.Module) -> ModuleType: + r"""Generate a `~lt.ThingClient` subclass for a particular Thing. - :param thing_description: the description of the Thing we're working with. - :return: a client class for that particular Thing. + This writes the generated code to a temporary file and imports it. + It is *temporarily* added to `sys.modules` in order to allow inspection + of the docstrings by code in `BaseDescriptor`\ . + + :param cls_name: the name of the generated class, used as part of + the module path. + :param tree: the generated abstract syntax tree. + :return: a module, which has been imported. :raises RuntimeError: if the generated module can't be loaded. """ - cls_name, tree = generate_client_ast(thing_description) # Now, use `ast.unparse` to write this to a file, which we can then import. with tempfile.TemporaryDirectory() as dir: fname = Path(dir) / "generated_client.py" @@ -690,4 +708,77 @@ def generate_client_class(thing_description: ThingDescription) -> "type[ThingCli # so the file still exists. It will be deleted at the end of this indented # block. # Finally, access the class in the now-executed module. + return module + + +def generate_client_class(thing_description: ThingDescription) -> "type[ThingClient]": + """Generate a `~lt.ThingClient` subclass for a particular Thing. + + :param thing_description: the description of the Thing we're working with. + :return: a client class for that particular Thing. + :raises RuntimeError: if the generated module can't be loaded. + """ + cls_name, tree = generate_client_ast(thing_description) + try: + module = load_ast_as_module(cls_name, tree) + except RuntimeError: + raise return getattr(module, cls_name) + + +def write_client_module( + thing_description: ThingDescription, destination: str, use_ruff: bool = True +) -> tuple[str, Path]: + r"""Generate a client for a Thing, and write it as a Python file. + + This uses `ast.unparse` to generate Python code from the output of + `generate_client_ast`, writes it to a file, and attempts to improve the + formatting using ``ruff``. + + Note that ``ruff`` will currently only be found if it is importable (i.e. + installed in the current environment): system-level ``ruff`` binaries will + not (yet) be found. + + Using ``ruff`` results in no unused imports, and better-formatted docstrings. + Formatting is still not ideal: property docstrings are not rendered + as triple-quoted constants. + + :param thing_description: the `ThingDescription` of the Thing we're generating + a client for. + :param destination: the location of the generated module. This should either + be a Python file (i.e. ending in `.py`) or a folder. If it is a folder, + the folder must exist and we'll create a file named as the (lower-case) + class name. + :param use_ruff: whether to attempt to format the generated code with Ruff. + :return: a tuple of ``(name, destination)`` where ``name`` is the name of the + generated client class, and ``destination`` is the output file path. + :raises ValueError: if the destination path isn't a Python file or a folder. + """ + cls_name, tree = generate_client_ast(thing_description) + dest = Path(destination) + if dest.is_dir(): + dest /= f"{cls_name.lower()}.py" + if not dest.suffix == ".py": + msg = f"The output path must be a Python file or a folder. Got '{dest}'." + raise ValueError(msg) + with open(dest, "w") as f: + f.write(ast.unparse(tree)) + # Optionally tidy up the source with `ruff` + # Tidy up generated code with `ruff` + if use_ruff and RUFF_BINARY is None: + warnings.warn( + "Can't locate `ruff` so code won't be neatly formatted.", + stacklevel=1, + ) + if use_ruff and RUFF_BINARY: + # Note: we ignore S603 here as it's just a reminder to check we're not passing + # untrusted input to commands: we're not, the only variable is a file we've just + # written, and `ruff` won't execute that file. + # Tidy up docstrings and unused imports + run( # noqa: S603 + [RUFF_BINARY, "check", dest, "--select", "D209,F401", "--fix"], + stdout=DEVNULL, + ) + # Format the file nicely + run([RUFF_BINARY, "format", dest], stdout=DEVNULL) # noqa: S603 + return cls_name, dest From 2a30113bb4c61ed4060616b661db88598f634183 Mon Sep 17 00:00:00 2001 From: Richard Bowman Date: Tue, 30 Jun 2026 14:35:22 +0100 Subject: [PATCH 9/9] Don't warn for types that ought to be `Any`. The code mapping JSONSchema to types had a warning by default when it typed anything as `Any`. I have softened this, so now it is happy to use `Any` for types that have a sufficiently vague schema (i.e. no type, oneof, const or enum). --- src/labthings_fastapi/code_generation.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/src/labthings_fastapi/code_generation.py b/src/labthings_fastapi/code_generation.py index eb30877f..eb66a4e8 100644 --- a/src/labthings_fastapi/code_generation.py +++ b/src/labthings_fastapi/code_generation.py @@ -335,7 +335,17 @@ def convert(self, schema: DataSchema, recursion_depth: int = 0) -> ast.expr: if isinstance(schema.properties, Mapping): return self.convert_to_model(schema, recursion_depth=recursion_depth) return _name("dict") + elif ( + schema.type is None + and schema.enum is None + and schema.oneOf is None + and schema.const is None + ): + # If the schema is really empty, it should be `Any` and there's no need for + # a warning. + return _name("Any") if self.warn_on_fallback: + # If we've found a type that we don't understand, warn then use `Any` warnings.warn( f"Could not convert {schema} to a Python type, falling back to `Any`.", stacklevel=1,