M rework/helper.py +22 -0
@@ 415,6 415,28 @@ def nary_pack(*bytestr):
)
+def convert_io(spec, args):
+ """Tries to convert string values into the well typed input values
+ needed for prepare or schedule calls.
+
+ May be usefull for user interfaces (or coming from json values).
+
+ """
+ if args is None and not len(spec):
+ return
+
+ typed = {}
+ for field in spec:
+ inp = _iobase.from_type(
+ field['type'], field['name'], field['required'], field['choices']
+ )
+ val = inp.from_string(args)
+ if val is not None:
+ typed[inp.name] = val
+
+ return typed
+
+
def pack_io(spec, args):
if args is None and not len(spec):
return
M rework/io.py +24 -0
@@ 48,6 48,9 @@ class _iobase:
class number(_iobase):
+ def from_string(self, args):
+ return self.binary_decode(args)
+
def binary_encode(self, args):
val = self.val(args)
if val is not None:
@@ 69,6 72,9 @@ class number(_iobase):
class string(_iobase):
+ def from_string(self, args):
+ return self.val(args)
+
def binary_encode(self, args):
val = self.val(args)
if val is not None:
@@ 86,6 92,13 @@ class string(_iobase):
class file(_iobase):
+ def from_string(self, args):
+ val = self.val(args)
+ if val is not None:
+ if not isinstance(val, bytes):
+ return val.encode('utf-8')
+ return val
+
def binary_encode(self, args):
val = self.val(args)
if val is not None:
@@ 101,6 114,11 @@ class file(_iobase):
class datetime(_iobase):
+ def from_string(self, args):
+ val = self.val(args)
+ if val is not None:
+ return defaultparse(val)
+
def binary_encode(self, args):
val = self.val(args)
if val is None:
@@ 156,6 174,12 @@ def _parsedatetime(strdt):
class moment(_iobase):
+ def from_string(self, args):
+ val = self.val(args)
+ if val is not None:
+ lisp.evaluate(val, env=_MOMENT_ENV)
+ return val
+
def binary_encode(self, args):
val = self.val(args)
if val is None:
M tests/test_api.py +24 -0
@@ 2,6 2,7 @@ import pytest
from datetime import datetime as dt
from rework.helper import (
+ convert_io,
filterio,
iospec,
host,
@@ 234,6 235,29 @@ def test_moment_input(engine, cleanup):
assert when == dt(2021, 1, 1, 9, 0)
+def test_convert_io(engine, cleanup):
+ reset_ops(engine)
+ register_tasks()
+ api.freeze_operations(engine)
+ args = {
+ 'myfile.txt': 'some file',
+ 'name': 'Babar',
+ 'weight': '65',
+ 'birthdate': '1973-5-20T09:00:00',
+ 'option': 'foo'
+ }
+ specs = iospec(engine)
+ spec = filterio(specs, 'yummy')
+ typed = convert_io(spec, args)
+ assert typed == {
+ 'myfile.txt': b'some file',
+ 'weight': 65,
+ 'birthdate': dt(1973, 5, 20, 9, 0),
+ 'name': 'Babar',
+ 'option': 'foo'
+ }
+
+
def test_prepare_with_inputs(engine, cleanup):
reset_ops(engine)
register_tasks()