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 64a567cf..ca55c7f0 100644 --- a/src/labthings_fastapi/client/__init__.py +++ b/src/labthings_fastapi/client/__init__.py @@ -9,13 +9,14 @@ 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 -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,161 +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 - - -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. - """ + parsed_td = ThingDescription.model_validate(thing_description) + client_cls = generate_client_class(parsed_td) + return client_cls - 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] +Value = TypeVar("Value") - # Set __set__ method based on whether writable - if writeable: - def __set__( - self: PropertyClientDescriptor, obj: ThingClient, value: Any - ) -> None: - obj.set_property(self.name, value) - else: - - def __set__( - self: PropertyClientDescriptor, obj: ThingClient, value: Any - ) -> None: - raise ClientPropertyError("This property may not be set.") +class ClientProperty(Generic[Value], FieldTypedBaseDescriptor[ThingClient, Value]): + """A descriptor to make properties of ThingClient objects work.""" - __set__.__annotations__["value"] = model - P.__set__ = __set__ # type: ignore[attr-defined] - if description: - P.__doc__ = description - return P() + def __init__( + self, + read_only: bool = False, + write_only: bool = False, + ) -> None: + """Initialise a ClientProperty. + :param read_only: whether the property should be read-only. + :param write_only: whether the property should be write-only. + """ + super().__init__() + self.read_only = read_only + self.write_only = write_only -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. + def instance_get(self, obj: ThingClient) -> Value: + """Retrieve the property. - :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 action_method(self: ThingClient, **kwargs: Any) -> Any: - return self.invoke_action(action_name, **kwargs) + :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 self.write_only: + raise ClientPropertyError("This property may not be read.") + return obj.get_property(self.name) - 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 __set__(self, obj: ThingClient, value: Value) -> None: + """Retrieve the property. + :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.read_only: + raise ClientPropertyError("This property may not be set.") + return obj.set_property(self.name, value) -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. +def client_property(read_only: bool = False, write_only: bool = False) -> Any: + r"""Create a `ClientProperty` descriptor. + 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`\ . - :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. """ - setattr( - cls, - property_name, - property_descriptor( - property_name, - property.get("type", Any), - description=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.py b/src/labthings_fastapi/code_generation.py new file mode 100644 index 00000000..eb66a4e8 --- /dev/null +++ b/src/labthings_fastapi/code_generation.py @@ -0,0 +1,794 @@ +"""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. +""" + +# 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 +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 subprocess import DEVNULL, run # noqa: S404 +from types import EllipsisType, ModuleType, NoneType +from typing import TYPE_CHECKING, Literal, Sequence +from uuid import uuid4 + +from labthings_fastapi.thing_description._model import ( + DataSchema, + InteractionAffordance, + ThingDescription, + 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 + + +def title_to_snake_case(title: str) -> str: + 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) + return "_".join(w.lower() for w in words) + + +def snake_to_camel_case(snake: str) -> str: + 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: + 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 _name(name: str, ctx: Literal["load", "store"] = "load") -> ast.Name: + r"""Generate a Name object (shorthand for `ast.Name`). + + 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. + + :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`). + + :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: + 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 | 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 _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. + + 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], + level=0, + ) + + +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") + 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, + ) + return _name("Any") + + 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.Assign( + targets=[_name("_model_config", "store")], + value=ast.Dict(keys=[_constant("extra")], values=[_constant("allow")]), + ) + ) + # 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, + ) + ) + + # 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( + _class( + name=name, + bases=[_name("BaseModel")], + 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: + 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 statement in body: + ast.fix_missing_locations(statement) + body_statements = [ast.unparse(s) for s in body] + for model in self.model_definitions: + model_statements = [ast.unparse(s) for s in model.body] + if model_statements == body_statements: + 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: + 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=[], + ) + + +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 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}" + return affordance.title or affordance.description + + +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. + """ + doc = _affordance_docstring(affordance) + if doc: + body.append(ast.Expr(_constant(doc))) + + +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. + """ + class_name = title_to_camel_case(thing_description.title) + "Client" + + # 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: + 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 = converter.convert(property) + class_body.append( + ast.AnnAssign( + target=_name(pname, "store"), + annotation=dtype, + value=ast.Call( + func=_name("client_property"), + args=[], + keywords=[ + ast.keyword( + arg="read_only", value=_constant(property.readOnly) + ), + ast.keyword( + 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` + actions = thing_description.actions or {} + for name, action in actions.items(): + aname = title_to_snake_case(name) + if action.input: + args = converter.input_model_to_arguments(action.input) + else: + args = NO_ARGUMENTS + if action.output: + 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] = [] + _append_docstring_if_present(function_body, action) + function_body.append( + 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 + ] + + [ + ast.keyword(value=_name("kwargs")) # pass additional kwargs + ], + ) + ) + ) + class_body.append( + _function( + name=aname, + args=args, + body=function_body, + returns=rtype, + ) + ) + + # The class definition is here: this includes `class_body` defined above, with the + # actions/properties. + class_definition = _class( + name=class_name, + bases=[_name("ThingClient")], + body=class_body, + ) + + # 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.append( + _import_from( + "labthings_fastapi.client", + ["ThingClient", "client_property", "_optional"], + ) + ) + code.append(_import_from("typing", ["Any", "Union"])) + code.append(_import_from("pydantic", ["BaseModel", "Field"])) + + code += converter.model_definitions + + code.append(class_definition) + + 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 load_ast_as_module(cls_name: str, tree: ast.Module) -> ModuleType: + r"""Generate a `~lt.ThingClient` subclass for a 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. + """ + # 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 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 diff --git a/tests/test_client_generation.py b/tests/test_client_generation.py new file mode 100644 index 00000000..f2e9f228 --- /dev/null +++ b/tests/test_client_generation.py @@ -0,0 +1,96 @@ +import ast +import importlib.util +import os +import tempfile + +from pydantic import BaseModel + +import labthings_fastapi as lt +import labthings_fastapi.code_generation as cg +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 + + +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 generate_and_verify(thing): + td = create_thing_without_server(thing).thing_description() + 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( + 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 cls_name in dir(module) + cls = getattr(module, cls_name) + assert issubclass(cls, lt.ThingClient) + + +def test_mything_generation(): + generate_and_verify(MyThing) + + +class SimpleModel(BaseModel): + a: int + b: str + + +class NestedModel(BaseModel): + c: SimpleModel + + +class ThingWithModels(lt.Thing): + @lt.property + def prop1(self) -> SimpleModel: + return SimpleModel(a=1, b="test") + + @lt.action + def action1(self, arg1: SimpleModel) -> SimpleModel: + return arg1 + + @lt.property + def prop2(self) -> NestedModel: + return NestedModel(c=SimpleModel(a=1, b="test")) + + +def test_with_models(): + generate_and_verify(ThingWithModels) + + +if __name__ == "__main__": + td = create_thing_without_server(ThingWithModels).thing_description() + print("Thing Description:") + print(td.model_dump_json(indent=2, exclude_unset=True)) + print("\nGenerated AST:") + _, ast_module = generate_client_ast(td) + print(ast.dump(ast_module, indent=4)) + print("\nGenerated 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):