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 = '' 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 )