253 lines
11 KiB
Python
253 lines
11 KiB
Python
from importlib import import_module
|
|
from typing import Dict, Callable, Union, Optional, Any, TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
from aiohttp import ClientSession as AioSession
|
|
from requests import Session as ReqSession
|
|
from bs4.element import Tag as BS4Tag
|
|
|
|
from .utils import PathT, yaml_overload
|
|
|
|
|
|
CallableDefaultT = Callable[[], Optional[str]]
|
|
DefaultInitT = Union[str, Dict[str, str]]
|
|
OptionsT = Dict[str, str]
|
|
|
|
|
|
class FormField:
|
|
def __init__(self, name: str, default: DefaultInitT = None, options: OptionsT = None, required: bool = False):
|
|
self.name: str = name
|
|
self.default = default
|
|
self.options: Optional[OptionsT] = None
|
|
if options is not None:
|
|
self.options = {str(k): str(v) for k, v in options.items()}
|
|
self.required: bool = required
|
|
|
|
def __repr__(self) -> str:
|
|
s = '<InputField' if self.options is None else '<SelectField'
|
|
s += f': name="{self.name}"'
|
|
if self._default is not None:
|
|
s += f', default="{self.default}"'
|
|
if self.required:
|
|
s += ', required=True'
|
|
if self.options is not None:
|
|
s += f', options={self.options}'
|
|
return s + '>'
|
|
|
|
def __str__(self) -> str:
|
|
return repr(self)
|
|
|
|
@property
|
|
def default(self) -> Optional[str]:
|
|
return self._default() if callable(self._default) else self._default
|
|
|
|
@default.setter
|
|
def default(self, default: DefaultInitT) -> None:
|
|
if isinstance(default, dict):
|
|
try:
|
|
module, function = default['module'], default['function']
|
|
except KeyError:
|
|
raise TypeError(f"Default for field '{self.name}' is invalid. The default must be either a string or "
|
|
f"`None` or a dictionary with the special keys 'module' and 'function'.")
|
|
obj = import_module(module)
|
|
for attr in function.split('.'):
|
|
obj = getattr(obj, attr)
|
|
self._default = obj
|
|
elif default is None:
|
|
self._default = None
|
|
else:
|
|
self._default = str(default)
|
|
|
|
def valid_option(self, option: str) -> bool:
|
|
return self.options is None or option in self.options.keys() or option in self.options.values()
|
|
|
|
def clean(self, value: Any) -> str:
|
|
value = str(value)
|
|
if self.options is None or value in self.options.keys():
|
|
return value
|
|
# Try to find an unambiguous match in the visible option texts:
|
|
key = None
|
|
for option_key, option_text in self.options.items():
|
|
if option_text == value:
|
|
if key is not None:
|
|
raise LookupError(f'More than one options key has the text "{value}" in {self}')
|
|
key = option_key
|
|
if key is not None:
|
|
return key
|
|
raise ValueError(f'"{value}" is not a valid option for {self}')
|
|
|
|
def check_with_html(self, field_tag: 'BS4Tag', check_defaults: bool = True) -> None:
|
|
from .html import check_field_interface
|
|
check_field_interface(field_tag, self, check_defaults=check_defaults)
|
|
|
|
def get_value_from_html_form(self, form_tag: 'BS4Tag') -> Optional[str]:
|
|
from .html import get_field_value
|
|
try:
|
|
return get_field_value(form_tag, self.name)
|
|
except AttributeError:
|
|
return None
|
|
|
|
|
|
class Form:
|
|
|
|
@staticmethod
|
|
def fields_from_dict(definition: Dict[str, Optional[dict]]) -> Dict[str, FormField]:
|
|
"""
|
|
Takes a dictionary defining form fields and creates `FormFields` objects from it.
|
|
Every key in `definition` is interpreted as the field's name.
|
|
The corresponding value can be `None` or a dictionary that can be unpacked into the FormField constructor call
|
|
alongside the name.
|
|
The special key `alias` in a field's dictionary is also allowed. If it is present, the corresponding value
|
|
will be used as the key in the output dictionary; otherwise the field's name is used as the key.
|
|
The constructed `FormField` objects are the values in the output dictionary.
|
|
"""
|
|
field_dict = {}
|
|
for name, field_def in definition.items():
|
|
if field_def is None:
|
|
field_def = {}
|
|
if isinstance(field_def, dict):
|
|
alias = field_def.pop('alias', name)
|
|
field_dict[alias] = FormField(name, **field_def)
|
|
else:
|
|
raise TypeError("Field definitions must be either dictionaries or `None`")
|
|
return field_dict
|
|
|
|
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:
|
|
The field definition dictionary
|
|
full_payload (optional):
|
|
Defines the behavior during payload validation/creation.
|
|
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 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.fields: Dict[str, FormField] = self.fields_from_dict(definition)
|
|
self.full_payload_always: bool = full_payload
|
|
self.url: Optional[str] = url
|
|
|
|
def __repr__(self) -> str:
|
|
fields = ', '.join(f"'{alias}': {field}" for alias, field in self.fields.items())
|
|
return f'Form({fields})'
|
|
|
|
@property
|
|
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: 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.
|
|
|
|
This payload can then be used for example in a GET request in the form of url parameters (after encoding it
|
|
in the appropriate string form) or in a regular POST request.
|
|
|
|
Args:
|
|
kwargs (optional):
|
|
Every key must correspond to an alias or name of a field in the internal dictionary of fields;
|
|
otherwise that key-value-pair is ignored.
|
|
If both a field's alias and name are different and both are present as keys in `kwargs`,
|
|
the alias-key takes precedence.
|
|
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.
|
|
|
|
Returns:
|
|
Validated name-value-mapping to be used for HTTP requests from the form's fields.
|
|
"""
|
|
payload = {}
|
|
for alias, field in self.fields.items():
|
|
if alias in kwargs.keys():
|
|
value = kwargs[alias]
|
|
payload[field.name] = field.default if value is None else field.clean(value)
|
|
elif alias != field.name and field.name in kwargs.keys():
|
|
value = kwargs[field.name]
|
|
payload[field.name] = field.default if value is None else field.clean(value)
|
|
elif field.required:
|
|
raise ValueError(f"`{alias}` 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) -> str:
|
|
"""
|
|
Uses `aiohttp` to perform a POST request to `.url` with the form's payload generated using `kwargs`.
|
|
|
|
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 response text from the request.
|
|
"""
|
|
if self.url is None:
|
|
raise AttributeError("`url` attribute not set")
|
|
from aiohttp import ClientSession
|
|
from webutils_df import in_async_session
|
|
|
|
@in_async_session
|
|
async def post(url: str, data: dict, session: ClientSession = None) -> str:
|
|
async with session.post(url, data=data) as response:
|
|
return await response.text()
|
|
return await post(self.url, self.get_payload(**kwargs), session=_aiohttp_session_obj)
|
|
|
|
def post_req(self, _requests_session_obj: 'ReqSession' = None, **kwargs: str) -> str:
|
|
"""
|
|
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 response text 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)).text
|
|
from requests import post
|
|
return post(self.url, data=self.get_payload(**kwargs)).text
|
|
|
|
def check_with_html(self, form_tag: 'BS4Tag', check_defaults: bool = True) -> None:
|
|
from .html import check_form_interface
|
|
check_form_interface(form_tag, self, check_defaults=check_defaults)
|
|
|
|
def get_values_from_html_form(self, form_tag: 'BS4Tag', required_fields_only: bool = True) -> Dict[str, str]:
|
|
from .html import get_field_values
|
|
return get_field_values(form_tag, self, required_fields_only=required_fields_only)
|
|
|
|
|
|
def load_form(*def_paths: PathT, dir_sort_key: Callable[[PathT], Any] = None, dir_sort_reverse: bool = False,
|
|
full_payload: bool = True, url: str = None) -> Form:
|
|
"""
|
|
Creates a form instance from an arbitrary number of definition files in YAML format.
|
|
|
|
Args:
|
|
def_paths:
|
|
Paths to the YAML files containing the form definitions; see the `paths` parameter in `yaml_overload`
|
|
dir_sort_key (optional):
|
|
See `yaml_overload`
|
|
dir_sort_reverse (optional):
|
|
See `yaml_overload`
|
|
full_payload (optional):
|
|
See the `Form` constructor
|
|
url (optional):
|
|
See the `Form` constructor
|
|
"""
|
|
return Form(
|
|
definition=yaml_overload(*def_paths, dir_sort_key=dir_sort_key, dir_sort_reverse=dir_sort_reverse),
|
|
full_payload=full_payload,
|
|
url=url
|
|
)
|