1
0
Fork 0
platypush/docs/source/_ext/sphinx_marshmallow.py
Fabio Manganiello 6db070db1c
- Fixed switchbot.status to handle virtual devices
- Fixed StrippedString schema field serialize handler

- Fixed rendering of lists in documentation schemas
2021-10-26 00:48:05 +02:00

87 lines
2.9 KiB
Python

import importlib
import json
import os
import re
import sys
from random import randint
from typing import Union, List
from docutils import nodes
from docutils.parsers.rst import Directive
from marshmallow import fields
class SchemaDirective(Directive):
"""
Support for response/message schemas in the docs. Format: ``.. schema:: rel_path.SchemaClass(arg1=value1, ...)``,
where ``rel_path`` is the path of the schema relative to ``platypush/schemas``.
"""
has_content = True
_schema_regex = re.compile(r'^\s*(.+?)\s*(\((.+?)\))?\s*$')
_schemas_path = os.path.abspath(
os.path.join(
os.path.dirname(os.path.relpath(__file__)), '..', '..', '..', 'platypush', 'schemas'))
sys.path.insert(0, _schemas_path)
@classmethod
def _get_field_value(cls, field):
metadata = getattr(field, 'metadata', {})
if metadata.get('example'):
return metadata['example']
if metadata.get('description'):
return metadata['description']
if isinstance(field, fields.Number):
return randint(1, 99)
if isinstance(field, fields.Boolean):
return bool(randint(0, 1))
if isinstance(field, fields.URL):
return 'https://example.org'
if isinstance(field, fields.List):
return [cls._get_field_value(field.inner)]
if isinstance(field, fields.Dict):
return {
cls._get_field_value(field.key_field) if field.key_field else 'key':
cls._get_field_value(field.value_field) if field.value_field else 'value'
}
if isinstance(field, fields.Nested):
ret = {
name: cls._get_field_value(f)
for name, f in field.nested().fields.items()
}
return [ret] if field.many else ret
return str(field.__class__.__name__).lower()
def _parse_schema(self) -> Union[dict, List[dict]]:
m = self._schema_regex.match('\n'.join(self.content))
schema_module_name = '.'.join(['platypush.schemas', *(m.group(1).split('.')[:-1])])
schema_module = importlib.import_module(schema_module_name)
schema_class = getattr(schema_module, m.group(1).split('.')[-1])
schema_args = eval(f'dict({m.group(3)})') if m.group(3) else {}
schema = schema_class(**schema_args)
output = {
name: self._get_field_value(field)
for name, field in schema.fields.items()
if not field.load_only
}
return [output] if schema.many else output
def run(self):
content = json.dumps(self._parse_schema(), sort_keys=True, indent=2)
block = nodes.literal_block(content, content)
block['language'] = 'json'
return [block]
def setup(app):
app.add_directive('schema', SchemaDirective)
return {
'version': '0.1',
'parallel_read_safe': True,
'parallel_write_safe': True,
}