-
Notifications
You must be signed in to change notification settings - Fork 4
Expand file tree
/
Copy pathproperties.py
More file actions
1338 lines (1085 loc) · 53.9 KB
/
properties.py
File metadata and controls
1338 lines (1085 loc) · 53.9 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""Define properties of `.Thing` objects.
:ref:`properties` are attributes of a `.Thing` that may be read or written to
over HTTP, and they are described in :ref:`gen_docs`. They are implemented with
a function `.property` (usually referenced as ``lt.property``), which is
intentionally similar to Python's built in `property`.
Properties can be defined in two ways as shown below:
.. code-block:: python
import labthings_fastapi as lt
class Counter(lt.Thing):
"A counter that knows what's remaining."
count: int = lt.property(default=0, readonly=True)
"The number of times we've incremented the counter."
target: int = lt.property(default=10)
"The number of times to increment before we stop."
@lt.property
def remaining(self) -> int:
"The number of steps remaining."
return self.target - self.count
@remaining.setter
def _set_remaining(self, value: int) -> None:
self.target = self.count + value
The first two properties are simple variables: they may be read and assigned
to, and will behave just like a regular variable. Their syntax is similar to
`dataclasses` or `pydantic` in that `.property` is used as a "field specifier"
to set options like the default value, and the type annotation is on the
class attribute. Documentation is in strings immediately following the
properties, which is understood by most automatic documentation tools.
``remaining`` is defined using a "getter" function, meaning this code will
be run each time ``counter.remaining`` is accessed. Its type will be the
return type of the function, and its docstring will come from the function
too. Setters with only a getter are read-only.
Adding a "setter" to properties is optional, and makes them read-write.
"""
from __future__ import annotations
import builtins
from collections.abc import Mapping
from types import EllipsisType
from typing import (
Annotated,
Any,
Callable,
Generic,
TypeVar,
overload,
TYPE_CHECKING,
)
from typing_extensions import Self
from weakref import WeakSet
from fastapi import Body, FastAPI
from pydantic import BaseModel, ConfigDict, RootModel, ValidationError, create_model
from .feature_flags import FEATURE_FLAGS
from .thing_description import type_to_dataschema
from .thing_description._model import (
DataSchema,
Form,
PropertyAffordance,
PropertyOp,
)
from .utilities import (
LabThingsRootModelWrapper,
labthings_data,
wrap_plain_types_in_rootmodel,
)
from .utilities.introspection import return_type
from .base_descriptor import (
DescriptorInfoCollection,
FieldTypedBaseDescriptor,
FieldTypedBaseDescriptorInfo,
)
from .exceptions import (
FeatureNotAvailableError,
NotConnectedToServerError,
ReadOnlyPropertyError,
MissingTypeError,
UnsupportedConstraintError,
)
if TYPE_CHECKING:
from .thing import Thing
# Note on ignored linter codes:
#
# DOC101 and DOC103 are a result of overloads not having docstrings. While
# the related D codes (checked by Ruff) don't flag overloads, pydoclint
# doesn't ignore overloads. This is most likely a pydoclint bug that
# we are working around.
# DOC201 is ignored on properties.
# Because we are overriding the
# builtin `property`, we are using `@builtins.property` which is not
# recognised by pydoclint as a property. I've therefore ignored those
# codes manually.
# pydocstyle ("D" codes) is run in Ruff and correctly recognises
# builtins.property as a property decorator.
CONSTRAINT_ARGS = {
"gt",
"ge",
"lt",
"le",
"multiple_of",
"allow_inf_nan",
"min_length",
"max_length",
"pattern",
}
"""The set of supported constraint arguments for properties."""
# The following exceptions are raised only when creating/setting up properties.
class OverspecifiedDefaultError(ValueError):
"""The default value has been specified more than once.
This error is raised when a `.DataProperty` is instantiated with both a
``default`` value and a ``default_factory`` provided.
"""
class MissingDefaultError(ValueError):
"""The default value has not been specified.
This error is raised when a `.DataProperty` is instantiated without a
``default`` value or a ``default_factory`` function.
"""
Value = TypeVar("Value")
"""The value returned by a property."""
Owner = TypeVar("Owner", bound="Thing")
"""The `.Thing` instance on which a property is bound."""
BasePropertyT = TypeVar("BasePropertyT", bound="BaseProperty")
"""An instance of (a subclass of) BaseProperty."""
def default_factory_from_arguments(
default: Value | EllipsisType = ...,
default_factory: Callable[[], Value] | None = None,
) -> Callable[[], Value]:
"""Process default arguments to get a default factory function.
This function takes the ``default`` and ``default_factory`` arguments
and will either return the ``default_factory`` if it is provided, or
will wrap the default value provided in a factory function.
Note that this wrapping does not copy the default value each time it is
called, so mutable default values are **only** safe if supplied as a
factory function.
This is used to avoid repeating the logic of checking whether a default
value or a factory function has been provided, and it returns a factory
rather than a default value so that it may be called multiple times to
get copies of the default value.
This function also ensures the default is specified exactly once, and
raises exceptions if it is not.
This logic originally lived only in the initialiser of `.DataProperty`
but it was needed in the `.property` and `.setting` functions in order
to correctly type them (so that specifying both or neither of the
``default`` and ``default_factory`` arguments would raise an error
with mypy).
:param default: the default value, or an ellipsis if not specified.
:param default_factory: a function that returns the default value.
:return: a function that returns the default value.
:raises OverspecifiedDefaultError: if both ``default`` and
``default_factory`` are specified.
:raises MissingDefaultError: if neither ``default`` nor ``default_factory``
are specified.
"""
if default is ... and default_factory is None:
# If the default is an ellipsis, we have no default value.
# Not having a default_factory alongside this
# is not allowed for DataProperty, so we raise an error.
raise MissingDefaultError()
if default is not ... and default_factory is not None:
# If both default and default_factory are set, we raise an error.
raise OverspecifiedDefaultError()
if default is not ...:
# We return a function that returns the static default value.
# This means we always have a factory function, which simplifies
# the rest of the code.
return lambda: default
if not callable(default_factory):
raise MissingDefaultError("The default_factory must be callable.")
return default_factory
# See comment at the top of the file regarding ignored linter rules.
@overload # use as a decorator @property
def property(
getter: Callable[[Owner], Value],
) -> FunctionalProperty[Owner, Value]: ...
@overload # use as `field: int = property(default=0)`
def property(
*, default: Value, readonly: bool = False, **constraints: Any
) -> Value: ...
@overload # use as `field: int = property(default_factory=lambda: 0)`
def property(
*, default_factory: Callable[[], Value], readonly: bool = False, **constraints: Any
) -> Value: ...
def property(
getter: Callable[[Owner], Value] | EllipsisType = ...,
*,
default: Value | EllipsisType = ...,
default_factory: Callable[[], Value] | None = None,
readonly: bool = False,
**constraints: Any,
) -> Value | FunctionalProperty[Owner, Value]:
r"""Define a Property on a `.Thing`\ .
This function may be used to define :ref:`properties` in
two ways, as either a decorator or a field specifier. See the
examples in the :mod:`.property` documentation.
Properties should always have a type annotation. This type annotation
will be used in automatic documentation and also to serialise the value
to JSON when it is sent over the network. This mean that the type of your
property should either be JSON serialisable (i.e. simple built-in types)
or a subclass of `pydantic.BaseModel`.
:param getter: is a method of a class that returns the value
of this property. This is usually supplied by using ``property``
as a decorator.
:param default: is the default value. Either this, ``getter`` or
``default_factory`` must be specified. Specifying both
or neither will raise an exception.
:param default_factory: should return your default value.
This may be used as an alternative to ``default`` if you
need to use a mutable datatype. For example, it would be
better to specify ``default_factory=list`` than
``default=[]`` because the second form would be shared
between all `.Thing`\ s with this property.
:param readonly: whether the property should be read-only
via the `.ThingClient` interface (i.e. over HTTP or via
a `.DirectThingClient`). This is automatically true if
``property`` is used as a decorator and no setter is
specified.
:param \**constraints: additional keyword arguments are passed
to `pydantic.Field` and allow constraints to be added to the
property. For example, ``ge=0`` constrains a numeric property
to be non-negative. See `pydantic.Field` for the full range
of constraint arguments.
:return: a property descriptor, either a `.FunctionalProperty`
if used as a decorator, or a `.DataProperty` if used as
a field.
:raises MissingDefaultError: if no valid default value is supplied,
and a getter is not in use.
:raises OverspecifiedDefaultError: if the default is specified more
than once (e.g. ``default``, ``default_factory``, or ``getter``).
**Typing Notes**
This function has somewhat complicated type hints, for two reasons.
Firstly, it may be used either as a decorator or as a field specifier,
so ``default`` performs double duty as a default value or a getter.
Secondly, when used as a field specifier the type hint for the
property is attached to the attribute of the class to which the
function's output is assigned. This means ``property`` does not know
its type hint until after it's been called.
When used as a field specifier, ``property`` returns a generic
`.DataProperty` descriptor instance, which will determine its type
when it is attached to the `.Thing`. The type hint on the return
value of ``property`` in that situation is a "white lie": we annotate
the return as having the same type as the ``default`` value (or the
``default_factory`` return value). This means that type checkers such
as ``mypy`` will check that the default is valid for the type of the
field, and won't raise an error about assigning, for example, an
instance of ``DataProperty[int]`` to a field annotated as ``int``.
Finally, the type of the ``default`` argument includes `.EllipsisType`
so that we can use ``...`` as its default value. This allows us to
distinguish between ``default`` not being set (``...``) and a desired
default value of ``None``. Similarly, ``...`` is the default value for
``getter`` so we can raise a more helpful error if a non-callable
value is passed as the first argument.
"""
if getter is not ...:
# If the getter argument is callable, we're being used as a decorator
# without arguments.
if not callable(getter):
raise MissingDefaultError(
"A non-callable getter was passed to `property`. Usually,"
"this means the default value was not passed as a keyword "
"argument, which is required."
)
if default_factory or default is not ...:
raise OverspecifiedDefaultError(
"A getter was specified at the same time as a default. Only "
"one of a getter, default, and default_factory may be used."
)
return FunctionalProperty(
fget=getter,
)
return DataProperty( # type: ignore[return-value]
default_factory=default_factory_from_arguments(default, default_factory),
readonly=readonly,
constraints=constraints,
)
class BaseProperty(FieldTypedBaseDescriptor[Owner, Value], Generic[Owner, Value]):
"""A descriptor that marks Properties on Things.
This class is used to determine whether an attribute of a `.Thing` should
be treated as a Property (see :ref:`wot_properties` - essentially, it
means the value should be available over HTTP).
`.BaseProperty` should not be used directly, instead it is recommended to
use `.property` to declare properties on your `.Thing` subclass.
"""
def __init__(self, constraints: Mapping[str, Any] | None = None) -> None:
"""Initialise a BaseProperty.
:param constraints: is passed as keyword arguments to `pydantic.Field`
to add validation constraints to the property. See `pydantic.Field`
for details. The module-level constant `CONSTRAINT_ARGS` lists
the supported constraint arguments.
:raises UnsupportedConstraintError: if unsupported constraint arguments
are supplied. See `CONSTRAINT_ARGS` for the supported arguments.
"""
super().__init__()
self._model: type[BaseModel] | None = None
self.readonly: bool = False
self.constraints = constraints or {}
for key in self.constraints:
if key not in CONSTRAINT_ARGS:
raise UnsupportedConstraintError(
f"Unknown constraint argument: {key}. \n"
f"Supported arguments are: {', '.join(CONSTRAINT_ARGS)}."
)
constraints: Mapping[str, Any]
"""Validation constraints applied to this property.
This mapping contains keyword arguments that will be passed to
`pydantic.Field` to add validation constraints to the property.
See `pydantic.Field` for details. The module-level constant
`CONSTRAINT_ARGS` lists the supported constraint arguments.
Note that these constraints will be enforced when values are
received over HTTP, but they are not automatically enforced
when setting the property directly on the `.Thing` instance
from Python code.
"""
@builtins.property
def model(self) -> type[BaseModel]:
"""A Pydantic model for the property's type.
`pydantic` models are used to serialise and deserialise values from
and to JSON. If the property is defined with a type hint that is not
a `pydantic.BaseModel` subclass, this property will ensure it is
wrapped in a `pydantic.RootModel` so it can be used with FastAPI.
If `.BaseProperty.value_type` is already a `pydantic.BaseModel`
subclass, this returns it unchanged.
:return: a Pydantic model for the property's type.
"""
if self._model is None:
self._model = wrap_plain_types_in_rootmodel(
self.value_type,
constraints=self.constraints,
)
return self._model
def default(self, obj: Owner | None) -> Value:
"""Return the default value of this property.
:param obj: the `.Thing` instance on which we are looking for the default.
or `None` if referring to the class. For now, this is ignored.
:return: the default value of this property.
:raises FeatureNotAvailableError: as this must be overridden.
"""
raise FeatureNotAvailableError(
f"{obj.name if obj else self.__class__}.{self.name} cannot be reset, "
f"as it's not supported by {self.__class__}."
)
def reset(self, obj: Owner) -> None:
"""Reset the property's value to a default state.
If there is a defined default value for the property, this method
should reset the property to that default.
Not every property is expected to implement ``reset`` so it is important
to handle `.FeatureNotAvailableError` exceptions, which will be raised if this
method is not overridden.
:param obj: the `.Thing` instance we want to reset.
:raises FeatureNotAvailableError: as only some subclasses implement resetting.
"""
raise FeatureNotAvailableError(
f"{obj.name}.{self.name} cannot be reset, as it's not supported by "
f"{self.__class__}."
)
def is_resettable(self, obj: Owner | None) -> bool:
r"""Determine if it's possible to reset this property.
By default, this returns `True` if ``reset`` has been overridden.
If you override ``reset`` but want more control over this behaviour,
you probably need to override `is_resettable`\ .
:param obj: the `.Thing` instance we want to reset.
:return: `True` if a call to ``reset()`` should work.
"""
return BaseProperty.reset is not self.__class__.reset
def add_to_fastapi(self, app: FastAPI, thing: Owner) -> None:
"""Add this action to a FastAPI app, bound to a particular Thing.
:param app: The FastAPI application we are adding endpoints to.
:param thing: The `.Thing` we are adding the endpoints for.
:raises NotConnectedToServerError: if the `.Thing` does not have
a ``path`` set.
"""
if thing.path is None:
raise NotConnectedToServerError(
"Can't add the endpoint without thing.path!"
)
# We can't use the decorator in the usual way, because we'd need to
# annotate the type of `body` with `self.model` which is only defined
# at runtime.
# The solution below is to manually add the annotation, before passing
# the function to the decorator.
if not self.readonly:
# The function is initially defined with a ``body`` argument of type
# ``Any`` but this will be replaced with the correct annotation a
# few lines below.
def set_property(body: Any) -> None:
if isinstance(body, RootModel):
body = body.root
self.__set__(thing, body)
set_property.__annotations__["body"] = Annotated[self.model, Body()]
app.put(
thing.path + self.name,
status_code=201,
response_description="Property set successfully",
summary=f"Set {self.title}",
description=f"## {self.title}\n\n{self.description or ''}",
)(set_property)
@app.get(
thing.path + self.name,
response_model=self.model,
response_description=f"Value of {self.name}",
summary=self.title,
description=f"## {self.title}\n\n{self.description or ''}",
)
def get_property() -> Any:
return self.__get__(thing)
if self.is_resettable(thing):
@app.post(
thing.path + self.name + "/reset",
summary=f"Reset {self.title}.",
description=(
f"## Reset {self.title}\n\n"
"This endpoint will reset the property to its default value. "
"The default value should be detailed in the Thing Description.\n\n"
"Not every property supports the reset-to-default operation, and "
"this endpoint is only present (e.g. in the OpenAPI docs) "
"for those that do.\n\n"
"This endpoint is identical to using the ``reset_property`` action"
rf"with the ``name`` argument set to ``{self.name}``\ ."
),
)
def reset() -> None:
self.reset(thing)
def property_affordance(
self, thing: Owner, path: str | None = None
) -> PropertyAffordance:
"""Represent the property in a Thing Description.
:param thing: the `.Thing` to which we are attached.
:param path: the URL of the `.Thing`. If not present, we will retrieve
the ``path`` from ``thing``.
:return: A description of the property in :ref:`wot_td` format.
:raises NotConnectedToServerError: if the `.Thing` does not have
a ``path`` set.
"""
path = path or thing.path
if path is None:
raise NotConnectedToServerError(
"Can't create an affordance without thing.path!"
)
ops = [PropertyOp.readproperty]
if not self.readonly:
ops.append(PropertyOp.writeproperty)
forms = [
Form[PropertyOp](
href=path + self.name,
op=ops,
),
]
data_schema: DataSchema = type_to_dataschema(self.model)
extra_fields = {}
try:
# Try to get hold of the default - may raise FeatureNotAvailableError
default = self.default(thing)
# Validate and dump it with the model to ensure it's simple types only
default_validated = self.model.model_validate(default)
extra_fields["default"] = default_validated.model_dump()
except FeatureNotAvailableError:
pass # Default should only be included if it's needed.
pa: PropertyAffordance = PropertyAffordance(
title=self.title,
forms=forms,
description=self.description,
readOnly=self.readonly,
writeOnly=False, # write-only properties are not yet supported
**extra_fields,
)
# We merge the data schema with the property affordance (which subclasses the
# DataSchema model) with the affordance second so its values take priority.
# Note that this works because all of the fields that get filled in by
# DataSchema are optional - so the PropertyAffordance is still valid without
# them.
return PropertyAffordance(
**{
**data_schema.model_dump(exclude_none=True),
**pa.model_dump(exclude_none=True),
}
)
def __set__(self, obj: Owner, value: Any) -> None:
"""Set the property (stub method).
This is a stub ``__set__`` method to mark this as a data descriptor.
:param obj: The Thing on which we are setting the value.
:param value: The new value for the Thing.
:raises NotImplementedError: as this must be overridden by concrete classes.
"""
raise NotImplementedError(
"__set__ must be overridden by property implementations."
)
def descriptor_info(
self, owner: Owner | None = None
) -> PropertyInfo[Self, Owner, Value]:
r"""Return an object that allows access to this descriptor's metadata.
:param owner: An instance to bind the descriptor info to. If `None`\ ,
the returned object will be unbound and will only refer to the class.
:return: A `PropertyInfo` instance describing this property.
"""
return PropertyInfo(self, owner, self._owner_ref())
class DataProperty(BaseProperty[Owner, Value], Generic[Owner, Value]):
"""A Property descriptor that acts like a regular variable.
`.DataProperty` descriptors remember their value, and can be read and
written to like a regular Python variable.
"""
@overload
def __init__( # noqa: DOC101,DOC103
self,
default: Value,
*,
readonly: bool = False,
constraints: Mapping[str, Any] | None = None,
) -> None: ...
@overload
def __init__( # noqa: DOC101,DOC103
self,
*,
default_factory: Callable[[], Value],
readonly: bool = False,
constraints: Mapping[str, Any] | None = None,
) -> None: ...
def __init__(
self,
default: Value | EllipsisType = ...,
*,
default_factory: Callable[[], Value] | None = None,
readonly: bool = False,
constraints: Mapping[str, Any] | None = None,
) -> None:
"""Create a property that acts like a regular variable.
`.DataProperty` descriptors function just like variables, in that
they can be read and written to as attributes of the `.Thing` and
their value will be the same every time it is read (i.e. it changes
only when it is set). This differs from `.FunctionalProperty` which
uses a "getter" function just like `builtins.property` and may
return a different value each time.
`.DataProperty` instances may always be set, when they are accessed
as an attribute of the `.Thing` instance. The ``readonly`` parameter
applies only to client code, whether it is remote or a
`.DirectThingClient` wrapper.
The type of the property's value will be inferred either from the
type subscript or from an annotation on the class attribute. This
is done in ``__get_name__`` because neither is available during
``__init__``.
:param default: the default value. This or ``default_factory`` must
be provided. Note that, as ``None`` is a valid default value,
this uses ``...`` instead as a way of checking whether ``default``
has been set.
:param default_factory: a function that returns the default value.
This is appropriate for datatypes such as lists, where using
a mutable default value can lead to odd behaviour.
:param readonly: if ``True``, the property may not be written to via
HTTP, or via `.DirectThingClient` objects, i.e. it may only be
set as an attribute of the `.Thing` and not from a client.
:param constraints: is passed as keyword arguments to `pydantic.Field`
to add validation constraints to the property. See `pydantic.Field`
for details.
"""
super().__init__(constraints=constraints)
self._default_factory = default_factory_from_arguments(
default=default, default_factory=default_factory
)
self.readonly = readonly
def instance_get(self, obj: Owner) -> Value:
"""Return the property's value.
This will supply a default if the property has not yet been set.
:param obj: The `.Thing` on which the property is being accessed.
:return: the value of the property.
"""
if self.name not in obj.__dict__:
# Note that a static default is converted to a factory function
# in __init__.
obj.__dict__[self.name] = self._default_factory()
return obj.__dict__[self.name]
def __set__(
self, obj: Owner, value: Value, emit_changed_event: bool = True
) -> None:
"""Set the property's value.
This sets the property's value, and notifies any observers.
If property validation is enabled by `.FEATURE_FLAGS.validate_properties_on_set`
this will validate the value against the property's model, and an error
will be raised if the value is not valid.
:param obj: the `.Thing` to which we are attached.
:param value: the new value for the property.
:param emit_changed_event: whether to emit a changed event.
"""
if FEATURE_FLAGS.validate_properties_on_set:
property_info = self.descriptor_info(obj)
obj.__dict__[self.name] = property_info.validate(value)
else:
obj.__dict__[self.name] = value
if emit_changed_event:
self.emit_changed_event(obj, value)
def default(self, obj: Owner | None) -> Value:
"""Return the default value of this property.
Note that this implementation is independent of the `.Thing` instance,
as there's currently no way to specify a per-instance default.
:param obj: the `.Thing` instance we want to reset.
:return: the default value of this property.
"""
return self._default_factory()
def reset(self, obj: Owner) -> None:
r"""Reset the property to its default value.
This resets to the value returned by ``default`` for `.DataProperty`\ .
:param obj: the `.Thing` instance we want to reset.
"""
self.__set__(obj, self.default(obj))
def _observers_set(self, obj: Thing) -> WeakSet:
"""Return the observers of this property.
Each observer in this set will be notified when the property is changed.
See ``.DataProperty.emit_changed_event``
:param obj: the `.Thing` to which we are attached.
:return: the set of observers corresponding to ``obj``.
"""
ld = labthings_data(obj)
if self.name not in ld.property_observers:
ld.property_observers[self.name] = WeakSet()
return ld.property_observers[self.name]
def emit_changed_event(self, obj: Thing, value: Value) -> None:
"""Notify subscribers that the property has changed.
This function is run when properties are updated. It must be run from
within a thread. This could be the `Invocation` thread of a running action, or
the property should be updated over via a client/http. It must be run from a
thread as it is communicating with the event loop via an `asyncio` blocking
portal and can cause deadlock if run in the event loop.
This method will raise a `.ServerNotRunningError` if the event loop is not
running, and should only be called after the server has started.
:param obj: the `.Thing` to which we are attached.
:param value: the new property value, to be sent to observers.
"""
obj._thing_server_interface.start_async_task_soon(
self.emit_changed_event_async,
obj,
value,
)
async def emit_changed_event_async(self, obj: Thing, value: Value) -> None:
"""Notify subscribers that the property has changed.
This function may only be run in the `anyio` event loop. See
`.DataProperty.emit_changed_event`.
:param obj: the `.Thing` to which we are attached.
:param value: the new property value, to be sent to observers.
"""
for observer in self._observers_set(obj):
await observer.send(
{"messageType": "propertyStatus", "data": {self._name: value}}
)
class FunctionalProperty(BaseProperty[Owner, Value], Generic[Owner, Value]):
"""A property that uses a getter and a setter.
For properties that should work like variables, use `.DataProperty`. For
properties that need to run code every time they are read, use this class.
Functional properties should work very much like Python's `builtins.property`
except that they are also available over HTTP.
"""
def __init__(
self,
fget: Callable[[Owner], Value],
constraints: Mapping[str, Any] | None = None,
) -> None:
"""Set up a FunctionalProperty.
Create a descriptor for a property that uses a getter function.
This class also inherits from `builtins.property` to help type checking
tools understand that it functions like a property.
:param fget: the getter function, called when the property is read.
:param constraints: is passed as keyword arguments to `pydantic.Field`
to add validation constraints to the property. See `pydantic.Field`
for details.
:raises MissingTypeError: if the getter does not have a return type annotation.
"""
super().__init__(constraints=constraints)
self._fget = fget
self._type = return_type(self._fget)
if self._type is None:
msg = (
f"{fget} does not have a valid type. "
"Return type annotations are required for property getters."
)
raise MissingTypeError(msg)
self._fset: Callable[[Owner, Value], None] | None = None
self.readonly: bool = True
@builtins.property
def fget(self) -> Callable[[Owner], Value]: # noqa: DOC201
"""The getter function."""
return self._fget
@builtins.property
def fset(self) -> Callable[[Owner, Value], None] | None: # noqa: DOC201
"""The setter function."""
return self._fset
def getter(self, fget: Callable[[Owner], Value]) -> Self:
"""Set the getter function of the property.
This function returns the descriptor, so it may be used as a decorator.
If the function has a docstring, it will be used as the property docstring.
:param fget: The new getter function.
:return: this descriptor (i.e. ``self``). This allows use as a decorator.
"""
self._fget = fget
self._type = return_type(self._fget)
self.__doc__ = fget.__doc__
return self
def setter(self, fset: Callable[[Owner, Value], None]) -> Self:
r"""Set the setter function of the property.
This function returns the descriptor, so it may be used as a decorator.
Once a setter has been added to a property, it will automatically become
writeable from client code (over HTTP and via `.DirectThingClient`).
To override this behaviour you may set ``readonly`` back to ``True``.
.. code-block:: python
class MyThing(lt.Thing):
def __init__(self, thing_server_interface):
super().__init__(thing_server_interface=thing_server_interface)
self._myprop: int = 0
@lt.property
def myprop(self) -> int:
"An example property that is an integer"
return self._myprop
@myprop.setter
def _set_myprop(self, val: int) -> None:
self._myprop = val
myprop.readonly = True # Prevent client code from setting it
.. note::
The example code above is not quite what would be done for the built-in
``@property`` decorator, because our setter does not have the same name
as the getter. Using a different name avoids type checkers such as
``mypy`` raising an error that the getter has been redefined with a
different type. The behaviour is identical whether the setter and getter
have the same name or not. The only difference is that the `.Thing`
will have an additional method called ``_set_myprop`` in the example
above.
:param fset: The new setter function.
:return: this descriptor (i.e. ``self``). This allows use as a decorator.
**Typing Notes**
Python's built-in ``property`` is treated as a special case by ``mypy``
and others, and our descriptor is not treated in the same way.
Naming the setter and getter the same is required by `builtins.property`
because the property must be overwritten when the setter is added, as
`builtins.property` is not mutable.
Our descriptor is mutable, so the setter may be added without having to
overwrite the object. While it would be nice to use exactly the same
conventions as `builtins.property`, it currently causes type errors that
must be silenced manually. We suggest using a different name for the setter
as an alternative to adding ``# type: ignore[no-redef]`` to the setter
function.
It will cause problems elsewhere in the code if descriptors are assigned
to more than one attribute, and this is checked in
`.BaseDescriptor.__set_name__`\ . We therefore return the setter rather
than the descriptor if the names don't match. The type hint does not
reflect this, as it would cause problems when the names do match (the
descriptor would become a ``FunctionalProperty | Callable`` and thus
typing errors would happen whenever it's accessed).
"""
self._fset = fset
self.readonly = False
if fset.__name__ != self.fget.__name__:
# Don't return the descriptor if it's named differently.
# see typing notes in docstring.
return fset # type: ignore[return-value]
return self
def instance_get(self, obj: Owner) -> Value:
"""Get the value of the property.
:param obj: the `.Thing` on which the attribute is accessed.
:return: the value of the property.
"""
return self.fget(obj)
def __set__(self, obj: Owner, value: Value) -> None:
"""Set the value of the property.
If property validation is enabled by `.FEATURE_FLAGS.validate_properties_on_set`
this will validate the value against the property's model, and an error
will be raised if the value is not valid.
:param obj: the `.Thing` on which the attribute is accessed.
:param value: the value of the property.
:raises ReadOnlyPropertyError: if the property cannot be set.
"""
if self.fset is None:
raise ReadOnlyPropertyError(f"Property {self.name} of {obj} has no setter.")
if FEATURE_FLAGS.validate_properties_on_set:
property_info = self.descriptor_info(obj)
value = property_info.validate(value)
self.fset(obj, value)
class PropertyInfo(
FieldTypedBaseDescriptorInfo[BasePropertyT, Owner, Value],
Generic[BasePropertyT, Owner, Value],
):
"""Access to the metadata of a Property.
This class provides a way to access the metadata of a Property, without
needing to retrieve the Descriptor object directly. It may be bound to a
`.Thing` instance, or may be accessed from the class.
"""
@builtins.property
def model(self) -> type[BaseModel]: # noqa: DOC201
"""A `pydantic.BaseModel` describing this property's value."""
return self.get_descriptor().model
@builtins.property
def model_instance(self) -> BaseModel: # noqa: DOC201
"""An instance of ``self.model`` populated with the current value.
:raises TypeError: if the return value can't be wrapped in a model.
"""
value = self.get()
if isinstance(value, BaseModel):
return value
else:
# If the return value isn't a model, we need to wrap it in a RootModel
# which we do using the model in self.model
cls = self.model
if not issubclass(cls, RootModel):
msg = (
f"LabThings couldn't wrap the return value of {self.name} in "
f"a model. This either means your property has an incorrect "
f"type, or there is a bug in LabThings.\n\n"
f"Value: {value}\n"
f"Expected type: {self.value_type}\n"
f"Actual type: {type(value)}\n"
f"Model: {self.model}\n"
)
raise TypeError(msg)
return cls(root=value)
@builtins.property
def default(self) -> Value: # noqa: DOC201
"""The default value of this property.
.. warning::
Note that this is an optional feature, so calling code must handle
`.FeatureNotAvailableError` exceptions.
"""
return self.get_descriptor().default(self.owning_object)
@builtins.property
def is_resettable(self) -> bool: # noqa: DOC201
"""Whether the property may be reset using the ``reset()`` method."""
return self.get_descriptor().is_resettable(self.owning_object)
def reset(self) -> None:
"""Reset the property to a default value.
.. warning::
Note that this is an optional feature, so calling code must handle
`.FeatureNotAvailableError` exceptions.
"""
return self.get_descriptor().reset(self.owning_object_or_error())