diff --git a/requirements/aio.txt b/requirements/aio.txt index ad1fc4d..78d8351 100644 --- a/requirements/aio.txt +++ b/requirements/aio.txt @@ -1,2 +1,3 @@ -r common.txt -aiohttp \ No newline at end of file +aiohttp +git+https://git.fajnberg.de/daniil/webutils-df.git \ No newline at end of file diff --git a/src/yamlhttpforms/form.py b/src/yamlhttpforms/form.py index 81d40a0..22ec119 100644 --- a/src/yamlhttpforms/form.py +++ b/src/yamlhttpforms/form.py @@ -1,22 +1,30 @@ -from types import SimpleNamespace from pathlib import Path -from typing import Dict +from typing import Dict, Callable, Union, Optional, TYPE_CHECKING from yaml import safe_load +if TYPE_CHECKING: + from aiohttp import ClientSession as AioSession, ClientResponse as AioResponse + from requests import Session as ReqSession, Response as ReqResponse + + +PathT = Union[Path, str] +CallableDefaultT = Callable[[], Optional[str]] +DefaultValueT = Union[str, CallableDefaultT] +OptionsT = Dict[str, str] class FormField: - def __init__(self, name: str, default: str = None, options: Dict[str, str] = None, hidden: bool = False): - self.name = name - self._default = default - self.options = options - self.hidden = hidden + def __init__(self, name: str, default: DefaultValueT = None, options: OptionsT = None, required: bool = False): + self.name: str = name + self._default: Optional[DefaultValueT] = default + self.options: Optional[OptionsT] = options + self.required: bool = required def __repr__(self) -> str: s = '' @@ -25,13 +33,9 @@ class FormField: return repr(self) @property - def default(self) -> str: + def default(self) -> Optional[str]: return self._default() if callable(self._default) else self._default - @property - def special(self) -> bool: - return self.hidden and self._default is None - def valid_option(self, option: str) -> bool: return self.options is None or option in self.options.keys() or option in self.options.values() @@ -50,10 +54,10 @@ class FormField: raise ValueError(f'"{value}" is not a valid option for {self}') -class Form(SimpleNamespace): - def __init__(self, definition: Dict[str, Dict], full_payload: bool = True): +class Form: + def __init__(self, definition: Dict[str, Dict], full_payload: bool = True, url: str = None): """ - ... + Creates a form instance from a definition dictionary. Each element in the dictionary must define a field. Args: definition: @@ -63,22 +67,23 @@ class Form(SimpleNamespace): If True (default), the fields' predefined default values will be inserted into the payload, wherever no value (or `None`) is explicitly passed for that field. If False, fields for which no values were explicitly passed, will not be included in the payload. - The exception is special fields (i.e. hidden fields with no default value); - these will cause an error, if no value is provided. + The exception is required fields; these will cause an error, if no value is provided. + url (optional): + Can be set in advance to the url that requests using this form's payload should be made to. """ - self._full_payload_always = full_payload - for alias, field_def in definition.items(): - setattr(self, alias, FormField(**field_def)) + self._fields: Dict[str, FormField] = {alias: FormField(**field_def) for alias, field_def in definition.items()} + self.full_payload_always: bool = full_payload + self.url: Optional[str] = url @property def fields(self) -> Dict[str, FormField]: - return {k: v for k, v in self.__dict__.items() if isinstance(v, FormField)} + return self._fields.copy() @property - def special_fields(self) -> Dict[str, FormField]: - return {k: v for k, v in self.__dict__.items() if isinstance(v, FormField) if v.special} + def required_fields(self) -> Dict[str, FormField]: + return {alias: field for alias, field in self._fields.items() if field.required} - def get_payload(self, **kwargs) -> Dict[str, str]: + def get_payload(self, **kwargs: str) -> Dict[str, str]: """ Creates a request payload from the form's fields as a dictionary, where the keys represent the name attributes of the form fields and the values the actual values to be passed. @@ -88,7 +93,8 @@ class Form(SimpleNamespace): Args: kwargs (optional): - Every key must correspond to a key in the internal fields dictionary. + Every key must correspond to a key in the internal dictionary of fields; + otherwise that key-value-pair is ignored. Values will be passed into the payload (if they pass validation). Select fields with predefined options will only allow one of the options to be passed. If `None` is passed as a value, the corresponding field's default value will be used in the payload. @@ -104,14 +110,67 @@ class Form(SimpleNamespace): payload[field.name] = field.default else: payload[field.name] = field.clean_value(value) - elif field.special: - raise ValueError(f"`{key}` is a special field (hidden, no-defeault), but no argument was passed.") - elif self._full_payload_always: + elif field.required: + raise ValueError(f"`{key}` is a required field, but no argument was passed.") + elif self.full_payload_always: payload[field.name] = field.default return payload + async def post_aio(self, _aiohttp_session_obj: 'AioSession' = None, **kwargs: str) -> 'AioResponse': + """ + Uses `aiohttp` to perform a POST request to `.url` with the form's payload generated using `kwargs`. -def load_form(*def_paths: Path, full_payload: bool = True) -> Form: + Args: + _aiohttp_session_obj (optional): + Can be set to a pre-existing `aiohttp.ClientSession` instance that should be used for the request. + kwargs: + Passed directly into `.get_payload`. + + Returns: + The `aiohttp.ClientResponse` object from the request. + """ + if self.url is None: + raise AttributeError("`url` attribute not set") + from aiohttp import ClientSession, ClientResponse + from webutils import in_async_session + + @in_async_session + async def post(url: str, data: dict, session: ClientSession = None) -> ClientResponse: + async with session.post(url, data=data) as response: + return response + return await post(self.url, self.get_payload(**kwargs), session=_aiohttp_session_obj) + + def post_req(self, _requests_session_obj: 'ReqSession' = None, **kwargs: str) -> 'ReqResponse': + """ + Uses `requests` to perform a POST request to `.url` with the form's payload generated using `kwargs`. + + Args: + _requests_session_obj (optional): + Can be set to a pre-existing `requests.Session` instance that should be used for the request. + kwargs: + Passed directly into `.get_payload`. + + Returns: + The `requests.Response` object from the request. + """ + if self.url is None: + raise AttributeError("`url` attribute not set") + if _requests_session_obj is not None: + return _requests_session_obj.post(self.url, data=self.get_payload(**kwargs)) + from requests import post + return post(self.url, data=self.get_payload(**kwargs)) + + +def load_form(*def_paths: PathT, full_payload: bool = True) -> Form: + """ + Creates a form instance from an arbitrary number of definition files in YAML format. + + Args: + def_paths: + Each element must be a path to a readable YAML file containing a valid form definition. + full_payload (optional): + Passed directly into the Form constructor. + """ definition = {} for path in def_paths: with open(path, 'r') as f: