2
# This file is part of Checkbox.
4
# Copyright 2012-2015 Canonical Ltd.
6
# Zygmunt Krynicki <zygmunt.krynicki@canonical.com>
8
# Checkbox is free software: you can redistribute it and/or modify
9
# it under the terms of the GNU General Public License version 3,
10
# as published by the Free Software Foundation.
12
# Checkbox is distributed in the hope that it will be useful,
13
# but WITHOUT ANY WARRANTY; without even the implied warranty of
14
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15
# GNU General Public License for more details.
17
# You should have received a copy of the GNU General Public License
18
# along with Checkbox. If not, see <http://www.gnu.org/licenses/>.
22
:mod:`plainbox.impl.pod`
23
========================
25
This module contains the :class:`POD` and :class:`Field` classes that simplify
26
creation of declarative struct-like data holding classes. POD classes get a
27
useful repr() method, useful initializer and accessors for each of the fields
28
defined inside. POD classes can be inherited (properly detecting any field
33
>>> class Person(POD):
34
... name = Field("name of the person", str, MANDATORY)
35
... age = Field("age of the person", int)
38
Creating POD instances, positional arguments match field definition order:
40
>>> joe = Person("joe", age=42)
42
Full-blown comparison (not only equality):
44
>>> joe == Person("joe", 42)
47
Reading and writing attributes also works (obviously):
57
For a full description check out the documentation of the :class:`POD` and
60
from collections import OrderedDict
61
from collections import namedtuple
62
from functools import total_ordering
63
from logging import getLogger
64
from textwrap import dedent
66
from plainbox.i18n import gettext as _
67
from plainbox.vendor import morris
69
__all__ = ('POD', 'PODBase', 'podify', 'Field', 'MANDATORY', 'UNSET',
70
'read_only_assign_filter', 'type_convert_assign_filter',
71
'type_check_assign_filter', 'modify_field_docstring')
74
_logger = getLogger("plainbox.pod")
79
"""A simple object()-like singleton that has a more useful repr()."""
82
return self.__class__.__name__
85
class MANDATORY(_Singleton):
88
Class for the special MANDATORY object.
90
This object can be used as a value in :attr:`Field.initial`.
92
Using ``MANDATORY`` on a field like that makes the explicit initialization
93
of the field mandatory during POD initialization. Please use this value to
94
require that the caller supplies a given argument to the POD you are
99
MANDATORY = MANDATORY()
102
class UNSET(_Singleton):
105
Class of the special UNSET object.
107
Singleton that is implicitly assigned to the values of all fields during
108
POD initialization. This way all fields will have a value, even early at
109
the time a POD is initialized. This can be important if the POD is somehow
110
repr()-ed or inspected in other means.
112
This object is also used by the :func:`read_only_assign_filter` function.
122
A field in a plain-old-data class.
124
Each field declares one attribute that can be read and written to. Just
125
like a C structure. Attributes are readable _and_ writable but there is a
126
lot of flexibility in what happens.
129
Name of the field (this is how this field can be accessed on the class
130
or instance that contains it). This gets set by
131
:meth:`_FieldCollection.inspect_namespace()`
133
Name of the POD dictionary entry used as backing store. This is set the
134
same as ``name`` above. By default that's just name prepended with the
137
An optional type hit. This is not used by default but assign filters
138
can inspect and use this for type checking. It can also be used for
139
documenting the intent of the field.
141
The docstring of the field, as initialized by the caller.
143
Initial value of this field, can be changed by passing arguments to
144
:meth:`POD.__init__()`. May be set to ``MANDATORY`` for a special
147
If not None this is a callable that produces the ``initial`` value for
150
If True, a on_{name}_changed
151
A flag controlling if notification events are sent for each
152
modification of POD data through field.
154
An (optional) function to use as the first responder to the change
155
notification signal. This field is only used if the ``notify``
156
attribute is set to ``True``.
157
:attr assign_filter_list:
158
An (optional) list of assignment filter functions.
160
A field is initialized based on the arguments passed to the POD
161
initializer. If no argument is passed that would correspond to a given
162
field the *initial* value is used. The *initial* value is either a constant
163
(reference) stored in the ``initial`` property of the field or the return
164
value of the callable in ``initial_fn``. Please make sure to use
165
``initial_fn`` if the value is not immutable as otherwise the produced
166
value may be unintentionally shared by multiple objects.
168
If the ``initial`` value is the special constant ``MANDATORY`` then the
169
corresponding field must be explicitly initialized by the POD initializer
170
argument list or a TypeError is raised.
172
The ``notify`` flag controls the existence of the ``on_{name}_changed(old,
173
new)`` signal on the class that includes the field. Applications can
174
connect to that signal to observe changes. The signal is fired whenever the
175
newly-assigned value compares *unequal* to the value currently stored in
178
The ``notify_fn`` is an optional function that is used instead of the
179
default (internal) :meth:`on_changed()` method of the Field class itself.
180
If specified it must have the same three-argument signature. It will be
181
called whenever the value of the field changes. Note that it will also be
182
called on the initial assignment, when the ``old`` argument it receives
183
will be set to the special ``UNSET`` object.
185
Lastly a docstring and type hint can be provided for documentation. The
186
type check is not enforced.
188
Assignment filters are used to inspect and optionally modify a value during
189
assignment (including the assignment done on object initialization) and can
190
be used for various operations (including type conversions and validation).
191
Assignment filters are called whenever a field is used to write to a POD.
193
Since assignment filters are arranged in a list and executed in-order, they
194
can also be used to modify the value as it gets propagated through the list
197
The signature of each filter is ``fn(pod, field, old_value, new_value)``.
198
The return value is the value shown to the subsequent filter or finally
204
def __init__(self, doc=None, type=None, initial=None, initial_fn=None,
205
notify=False, notify_fn=None, assign_filter_list=None):
206
"""Initialize (define) a new POD field."""
207
self.__doc__ = dedent(doc) if doc is not None else None
209
self.initial = initial
210
self.initial_fn = initial_fn
212
self.notify_fn = notify_fn
213
self.assign_filter_list = assign_filter_list
214
self.name = None # Set via :meth:`gain_name()`
215
self.instance_attr = None # ditto
216
self.signal_name = None # ditto
218
for fn in self.assign_filter_list or ():
219
if hasattr(fn, 'field_docstring_ext'):
220
doc_extra.append(fn.field_docstring_ext.format(field=self))
223
'\n\nSide effects of assign filters:\n'
224
+ '\n'.join(' - {}'.format(extra) for extra in doc_extra))
225
self.counter = self.__class__._counter
226
self.__class__._counter += 1
229
def change_notifier(self):
231
Decorator for changing the change notification function.
233
This decorator can be used to define all the fields in one block and
234
all the notification function in another block. It helps to make the
239
>>> class Person(POD):
242
... @name.change_notifier
243
... def _name_changed(self, old, new):
244
... print("changed from {!r} to {!r}".format(old, new))
245
>>> person = Person()
246
changed from UNSET to None
247
>>> person.name = "bob"
248
changed from None to 'bob'
251
Keep in mind that the decorated function is converted to a signal
252
automatically. The name of the function is also irrelevant, the POD
253
core automatically creates signals that have consistent names of
254
``on_{field}_changed()``.
263
"""Get a debugging representation of a field."""
264
return "<{} name:{!r}>".format(self.__class__.__name__, self.name)
267
def is_mandatory(self) -> bool:
268
"""Flag indicating if the field needs a mandatory initializer."""
269
return self.initial is MANDATORY
271
def gain_name(self, name: str) -> None:
276
Name of the field as it appears in a class definition
278
Method called at most once on each Field instance embedded in a
279
:class:`POD` subclass. This method informs the field of the name it was
280
assigned to in the class.
283
self.instance_attr = "_{}".format(name)
284
self.signal_name = "on_{}_changed".format(name)
286
def alter_cls(self, cls: type) -> None:
288
Modify class definition this field belongs to.
290
This method is called during class construction. It allows the field to
291
alter the class and add the on_{field.name}_changed signal. The signal
292
is only added if notification is enabled *and* if there is no such
293
signal in the first place (this allows inheritance not to create
294
separate but identically-named signals and allows signal handlers
295
connected via the base class to work on child classes.
299
assert self.signal_name is not None
300
if not hasattr(cls, self.signal_name):
301
signal_def = morris.signal(
302
self.notify_fn if self.notify_fn is not None
303
else self.on_changed,
304
signal_name='{}.{}'.format(cls.__name__, self.signal_name))
305
setattr(cls, self.signal_name, signal_def)
307
def __get__(self, instance: object, owner: type) -> "Any":
309
Get field value from an object or from a class.
311
This method is part of the Python descriptor protocol.
316
return getattr(instance, self.instance_attr)
318
def __set__(self, instance: object, new_value: "Any") -> None:
320
Set field value from on an object.
322
This method is part of the Python descriptor protocol.
324
Assignments respect the assign filter chain, that is, the new value is
325
being pushed through the chain of callbacks (each has a chance to alter
326
the value) until it is finally assigned. Any of the callbacks can raise
327
an exception and abort the setting process.
329
This can be used to implement simple type checking, value checking or
330
even type and value conversions.
332
if self.assign_filter_list is not None or self.notify:
333
old_value = getattr(instance, self.instance_attr, UNSET)
334
# Run the value through assign filters
335
if self.assign_filter_list is not None:
336
for assign_filter in self.assign_filter_list:
337
new_value = assign_filter(instance, self, old_value, new_value)
338
# Do value modification check if change notification is enabled
339
if self.notify and hasattr(instance, self.instance_attr):
340
if new_value != old_value:
341
setattr(instance, self.instance_attr, new_value)
342
on_field_change = getattr(instance, self.signal_name)
343
on_field_change(old_value, new_value)
346
setattr(instance, self.instance_attr, new_value)
348
def on_changed(self, pod: "POD", old: "Any", new: "Any") -> None:
350
The first responder of the per-field modification signal.
353
The object that contains the modified values
355
The old value of the field
357
The new value of the field
359
_logger.debug("<%s %s>.%s(%r, %r)", pod.__class__.__name__, id(pod),
360
self.signal_name, old, new)
366
"""Base class for POD-like classes."""
369
namedtuple_cls = namedtuple('PODBase', '')
371
def __init__(self, *args, **kwargs):
373
Initialize a new POD object.
375
Positional arguments bind to fields in declaration order. Keyword
376
arguments bind to fields in any order but fields cannot be initialized
380
If there are more positional arguments than fields to initialize
382
If a keyword argument doesn't correspond to a field name.
384
If a field is initialized twice (first with positional arguments,
385
then again with keyword arguments).
387
If a ``MANDATORY`` field is not initialized.
389
field_list = self.__class__.field_list
390
# Set all of the instance attributes to the special UNSET value, this
391
# is useful if something fails and the object is inspected somehow.
392
# Then all the attributes will be still UNSET.
393
for field in field_list:
394
setattr(self, field.instance_attr, UNSET)
395
# Check if the number of positional arguments is correct
396
if len(args) > len(field_list):
397
raise TypeError("too many arguments")
398
# Initialize mandatory fields using positional arguments
399
for field, field_value in zip(field_list, args):
400
setattr(self, field.name, field_value)
401
# Initialize fields using keyword arguments
402
for field_name, field_value in kwargs.items():
403
field = getattr(self.__class__, field_name, None)
404
if not isinstance(field, Field):
405
raise TypeError("no such field: {}".format(field_name))
406
if getattr(self, field.instance_attr) is not UNSET:
408
"field initialized twice: {}".format(field_name))
409
setattr(self, field_name, field_value)
410
# Initialize remaining fields using their default initializers
411
for field in field_list:
412
if getattr(self, field.instance_attr) is not UNSET:
414
if field.is_mandatory:
416
"mandatory argument missing: {}".format(field.name))
417
if field.initial_fn is not None:
418
field_value = field.initial_fn()
420
field_value = field.initial
421
setattr(self, field.name, field_value)
424
"""Get a debugging representation of a POD object."""
425
return "{}({})".format(
426
self.__class__.__name__,
428
'{}={!r}'.format(field.name, getattr(self, field.name))
429
for field in self.__class__.field_list]))
431
def __eq__(self, other: "POD") -> bool:
433
Check that this POD is equal to another POD.
435
POD comparison is implemented by converting them to tuples and
436
comparing the two tuples.
438
if not isinstance(other, POD):
439
return NotImplemented
440
return self.as_tuple() == other.as_tuple()
442
def __lt__(self, other: "POD") -> bool:
444
Check that this POD is "less" than an another POD.
446
POD comparison is implemented by converting them to tuples and
447
comparing the two tuples.
449
if not isinstance(other, POD):
450
return NotImplemented
451
return self.as_tuple() < other.as_tuple()
453
def as_tuple(self) -> tuple:
455
Return the data in this POD as a tuple.
457
Order of elements in the tuple corresponds to the order of field
460
return self.__class__.namedtuple_cls(*[
461
getattr(self, field.name)
462
for field in self.__class__.field_list
465
def as_dict(self) -> dict:
467
Return the data in this POD as a dictionary.
470
UNSET values are not added to the dictionary.
473
field.name: getattr(self, field.name)
474
for field in self.__class__.field_list
475
if getattr(self, field.name) is not UNSET
479
class _FieldCollection:
482
Support class for constructing POD meta-data information.
484
Helper class that simplifies :class:`PODMeta` code that harvests
485
:class:`Field` instances during class construction. Looking at the
486
namespace and a list of base classes come up with a list of Field objects
487
that belong to the given POD.
490
A list of :class:`Field` instances
491
:attr field_origin_map:
492
A dictionary mapping from field name to the *name* of the class that
498
self.field_origin_map = {} # field name -> defining class name
500
def inspect_cls_for_decorator(self, cls: type) -> None:
501
"""Analyze a bare POD class."""
502
self.inspect_base_classes(cls.__bases__)
503
self.inspect_namespace(cls.__dict__, cls.__name__)
505
def inspect_base_classes(self, base_cls_list: "List[type]") -> None:
507
Analyze base classes of a POD class.
509
Analyze a list of base classes and check if they have consistent
510
fields. All analyzed fields are added to the internal data structures.
512
:param base_cls_list:
513
A list of classes to inspect. Only subclasses of POD are inspected.
515
for base_cls in base_cls_list:
516
if not issubclass(base_cls, PODBase):
518
base_cls_name = base_cls.__name__
519
for field in base_cls.field_list:
520
self.add_field(field, base_cls_name)
522
def inspect_namespace(self, namespace: dict, cls_name: str) -> None:
524
Analyze namespace of a POD class.
526
Analyze a namespace of a newly (being formed) class and check if it has
527
consistent fields. All analyzed fields are added to the internal data
531
This method calls :meth:`Field.gain_name()` on all fields it finds.
534
for field_name, field in namespace.items():
535
if not isinstance(field, Field):
537
field.gain_name(field_name)
539
fields.sort(key=lambda field: field.counter)
541
self.add_field(field, cls_name)
543
def get_namedtuple_cls(self, name: str) -> type:
545
Create a new namedtuple that corresponds to the fields seen so far.
548
Name of the namedtuple class
550
A new namedtuple class
552
return namedtuple(name, [field.name for field in self.field_list])
554
def add_field(self, field: Field, base_cls_name: str) -> None:
556
Add a field to the collection.
559
A :class:`Field` instance
560
:param base_cls_name:
561
The name of the class that defines the field
563
If any of the base classes have overlapping fields.
565
assert field.name is not None
566
field_name = field.name
567
if field_name not in self.field_origin_map:
568
self.field_origin_map[field_name] = base_cls_name
569
self.field_list.append(field)
571
raise TypeError("field {1}.{0} clashes with {2}.{0}".format(
572
field_name, base_cls_name, self.field_origin_map[field_name]))
578
Meta-class for all POD classes.
580
This meta-class is responsible for correctly handling field inheritance.
581
This class sets up ``field_list`` and ``namedtuple_cls`` attributes on the
585
def __new__(mcls, name, bases, namespace):
586
fc = _FieldCollection()
587
fc.inspect_base_classes(bases)
588
fc.inspect_namespace(namespace, name)
589
namespace['field_list'] = fc.field_list
590
namespace['namedtuple_cls'] = fc.get_namedtuple_cls(name)
591
cls = super().__new__(mcls, name, bases, namespace)
592
for field in fc.field_list:
597
def __prepare__(mcls, name, bases, **kwargs):
599
Get a namespace for defining new POD classes.
601
Prepare the namespace for the definition of a class using PODMeta as a
602
meta-class. Since we want to observe the order of fields, using an
603
OrderedDict makes that task trivial.
610
Decorator for POD classes.
612
The decorator offers an alternative from using the POD class (with the
613
PODMeta meta-class). Instead of using that, one can use the ``@podify``
614
decorator on a PODBase-derived class.
616
if not isinstance(cls, type) or not issubclass(cls, PODBase):
617
raise TypeError("cls must be a subclass of PODBase")
618
fc = _FieldCollection()
619
fc.inspect_cls_for_decorator(cls)
620
cls.field_list = fc.field_list
621
cls.namedtuple_cls = fc.get_namedtuple_cls(cls.__name__)
622
for field in fc.field_list:
628
class POD(PODBase, metaclass=PODMeta):
631
Base class that removes boilerplate from plain-old-data classes.
633
Use POD as your base class and define :class:`Field` objects inside. Don't
634
define any __init__() (unless you really, really have to have one) and
635
instead set appropriate attributes on the initializer of a particular field
638
What you get for *free* is, all the properties (for each field),
639
documentation, initializer, comparison methods (PODs have total ordering)
640
and the __repr__() method.
642
There are some additional methods, such as :meth:`as_tuple()` and
643
:meth:`as_dict()` that may be of use in some circumstances.
645
All fields in a single POD subclass are collected (including all of the
646
fields in the parent classes) and arranged in a list. That list is
647
available as ``POD.field_list``.
649
In addition each POD class has an unique named tuple that corresponds to
650
each field stored inside the POD, the named tuple is available as
651
``POD.namedtuple_cls``. The return value of :meth:`as_tuple()` actually
656
def modify_field_docstring(field_docstring_ext: str):
658
Decorator for altering field docstrings via assign filter functions.
660
A decorator for assign filter functions that allows them to declaratively
661
modify the docstring of the field they are used on.
663
:param field_docstring_ext:
664
A string compatible with python's str.format() method. The string
665
should be one line long (newlines will look odd) and may reference any
666
of the field attributes, as exposed by the {field} named format
671
>>> @modify_field_docstring("not even")
672
... def not_even(instance, field, old, new):
674
... raise ValueError("value cannot be even")
678
fn.field_docstring_ext = field_docstring_ext
683
@modify_field_docstring("constant (read-only after initialization)")
684
def read_only_assign_filter(
685
instance: POD, field: Field, old: "Any", new: "Any") -> "Any":
687
An assign filter that makes a field read-only.
689
The field can be only assigned if the old value is ``UNSET``, that is,
690
during the initial construction of a POD object.
693
A subclass of :class:`POD` that contains ``field``
695
The :class:`Field` being assigned to
697
The current value of the field
699
The proposed value of the field
702
:raises AttributeError:
703
if ``old`` is anything but the special object ``UNSET``
707
raise AttributeError(_(
709
).format(instance.__class__.__name__, field.name))
712
const = read_only_assign_filter
715
@modify_field_docstring(
716
"type-converted (value must be convertible to {field.type.__name__})")
717
def type_convert_assign_filter(
718
instance: POD, field: Field, old: "Any", new: "Any") -> "Any":
720
An assign filter that converts the value to the field type.
722
The field must have a valid python type object stored in the .type field.
725
A subclass of :class:`POD` that contains ``field``
727
The :class:`Field` being assigned to
729
The current value of the field
731
The proposed value of the field
733
``new`` type-converted to ``field.type``.
735
if ``new`` cannot be converted to ``field.type``
737
return field.type(new)
740
@modify_field_docstring(
741
"type-checked (value must be of type {field.type.__name__})")
742
def type_check_assign_filter(
743
instance: POD, field: Field, old: "Any", new: "Any") -> "Any":
745
An assign filter that type-checks the value according to the field type.
747
The field must have a valid python type object stored in the .type field.
750
A subclass of :class:`POD` that contains ``field``
752
The :class:`Field` being assigned to
754
The current value of the field
756
The proposed value of the field
760
if ``new`` is not an instance of ``field.type``
762
if isinstance(new, field.type):
764
raise TypeError("{}.{} requires objects of type {}".format(
765
instance.__class__.__name__, field.name, field.type.__name__))
768
typed = type_check_assign_filter
771
@modify_field_docstring(
772
"unset or type-checked (value must be of type {field.type.__name__})")
773
def unset_or_type_check_assign_filter(
774
instance: POD, field: Field, old: "Any", new: "Any") -> "Any":
776
An assign filter that type-checks the value according to the field type.
779
This filter allows (passes through) the special ``UNSET`` value as-is.
781
The field must have a valid python type object stored in the .type field.
784
A subclass of :class:`POD` that contains ``field``
786
The :class:`Field` being assigned to
788
The current value of the field
790
The proposed value of the field
794
if ``new`` is not an instance of ``field.type``
798
return type_check_assign_filter(instance, field, old, new)
801
unset_or_typed = unset_or_type_check_assign_filter
804
class sequence_type_check_assign_filter:
807
Assign filter for typed sequences.
809
An assign filter for typed sequences (lists or tuples) that must contain an
810
object of the given type.
813
def __init__(self, item_type: type):
815
Initialize the assign filter with the given sequence item type.
818
Desired type of each sequence item.
820
self.item_type = item_type
823
def field_docstring_ext(self) -> str:
824
return "type-checked sequence (items must be of type {})".format(
825
self.item_type.__name__)
828
self, instance: POD, field: Field, old: "Any", new: "Any"
831
An assign filter that type-checks the value of all sequence elements.
834
A subclass of :class:`POD` that contains ``field``
836
The :class:`Field` being assigned to
838
The current value of the field
840
The proposed value of the field
844
if ``new`` is not an instance of ``field.type``
847
if not isinstance(item, self.item_type):
849
"{}.{} requires all sequence elements of type {}".format(
850
instance.__class__.__name__, field.name,
851
self.item_type.__name__))
855
typed.sequence = sequence_type_check_assign_filter
858
class unset_or_sequence_type_check_assign_filter(typed.sequence):
861
Assign filter for typed sequences.
864
This filter allows (passes through) the special ``UNSET`` value as-is.
866
An assign filter for typed sequences (lists or tuples) that must contain an
867
object of the given type.
871
def field_docstring_ext(self) -> str:
873
"unset or type-checked sequence (items must be of type {})"
874
).format(self.item_type.__name__)
877
self, instance: POD, field: Field, old: "Any", new: "Any"
880
An assign filter that type-checks the value of all sequence elements.
883
This filter allows (passes through) the special ``UNSET`` value
887
A subclass of :class:`POD` that contains ``field``
889
The :class:`Field` being assigned to
891
The current value of the field
893
The proposed value of the field
897
if ``new`` is not an instance of ``field.type``
901
return super().__call__(instance, field, old, new)
904
unset_or_typed.sequence = unset_or_sequence_type_check_assign_filter
907
@modify_field_docstring("unique elements (sequence elements cannot repeat)")
908
def unique_elements_assign_filter(
909
instance: POD, field: Field, old: "Any", new: "Any") -> "Any":
911
An assign filter that ensures a sequence has non-repeating items.
914
A subclass of :class:`POD` that contains ``field``
916
The :class:`Field` being assigned to
918
The current value of the field
920
The proposed value of the field
924
if ``new`` contains any duplicates
929
raise ValueError("Duplicate element: {!r}".format(item))
933
unique = unique_elements_assign_filter
added
removed
plainbox/plainbox/impl/pod.py
