added capabilities for making requests; changed hidden to required fields; docstrings

This commit is contained in:
Daniil Fajnberg 2021-12-26 14:41:01 +01:00
parent 9dbe472bbf
commit 7e10668cbc
2 changed files with 92 additions and 32 deletions

View File

@ -1,2 +1,3 @@
-r common.txt -r common.txt
aiohttp aiohttp
git+https://git.fajnberg.de/daniil/webutils-df.git

View File

@ -1,22 +1,30 @@
from types import SimpleNamespace
from pathlib import Path from pathlib import Path
from typing import Dict from typing import Dict, Callable, Union, Optional, TYPE_CHECKING
from yaml import safe_load 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: class FormField:
def __init__(self, name: str, default: str = None, options: Dict[str, str] = None, hidden: bool = False): def __init__(self, name: str, default: DefaultValueT = None, options: OptionsT = None, required: bool = False):
self.name = name self.name: str = name
self._default = default self._default: Optional[DefaultValueT] = default
self.options = options self.options: Optional[OptionsT] = options
self.hidden = hidden self.required: bool = required
def __repr__(self) -> str: def __repr__(self) -> str:
s = '<InputField' if self.options is None else '<SelectField' s = '<InputField' if self.options is None else '<SelectField'
s += f': name="{self.name}"' s += f': name="{self.name}"'
if self.hidden: if self.required:
s += ', hidden=True' s += ', required=True'
if self._default is not None: if self._default is not None:
s += f', default="{self.default}"' s += f', default="{self.default}"'
return s + '>' return s + '>'
@ -25,13 +33,9 @@ class FormField:
return repr(self) return repr(self)
@property @property
def default(self) -> str: def default(self) -> Optional[str]:
return self._default() if callable(self._default) else self._default 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: def valid_option(self, option: str) -> bool:
return self.options is None or option in self.options.keys() or option in self.options.values() 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}') raise ValueError(f'"{value}" is not a valid option for {self}')
class Form(SimpleNamespace): class Form:
def __init__(self, definition: Dict[str, Dict], full_payload: bool = True): 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: Args:
definition: definition:
@ -63,22 +67,23 @@ class Form(SimpleNamespace):
If True (default), the fields' predefined default values will be inserted into the payload, 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. 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. 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); The exception is required fields; these will cause an error, if no value is provided.
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 self._fields: Dict[str, FormField] = {alias: FormField(**field_def) for alias, field_def in definition.items()}
for alias, field_def in definition.items(): self.full_payload_always: bool = full_payload
setattr(self, alias, FormField(**field_def)) self.url: Optional[str] = url
@property @property
def fields(self) -> Dict[str, FormField]: 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 @property
def special_fields(self) -> Dict[str, FormField]: def required_fields(self) -> Dict[str, FormField]:
return {k: v for k, v in self.__dict__.items() if isinstance(v, FormField) if v.special} 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, 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. 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: Args:
kwargs (optional): 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). 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. 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. 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 payload[field.name] = field.default
else: else:
payload[field.name] = field.clean_value(value) payload[field.name] = field.clean_value(value)
elif field.special: elif field.required:
raise ValueError(f"`{key}` is a special field (hidden, no-defeault), but no argument was passed.") raise ValueError(f"`{key}` is a required field, but no argument was passed.")
elif self._full_payload_always: elif self.full_payload_always:
payload[field.name] = field.default payload[field.name] = field.default
return payload 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 = {} definition = {}
for path in def_paths: for path in def_paths:
with open(path, 'r') as f: with open(path, 'r') as f: