Compare commits

..

No commits in common. "e6c5076e39d3750076519b1ff2cc923164c8d713" and "3001caf7ce44b0111fc1915f9ccdb5a5a01c1fd7" have entirely different histories.

2 changed files with 27 additions and 30 deletions

View File

@ -1,6 +1,6 @@
[metadata] [metadata]
name = yamlhttpforms name = yamlhttpforms
version = 0.0.8 version = 0.0.6
author = Daniil F. author = Daniil F.
author_email = mail@placeholder123.to author_email = mail@placeholder123.to
description = HTTP forms defined in YAML description = HTTP forms defined in YAML

View File

@ -2,8 +2,8 @@ from importlib import import_module
from typing import Dict, Callable, Union, Optional, Any, TYPE_CHECKING from typing import Dict, Callable, Union, Optional, Any, TYPE_CHECKING
if TYPE_CHECKING: if TYPE_CHECKING:
from aiohttp import ClientSession as AioSession from aiohttp import ClientSession as AioSession, ClientResponse as AioResponse
from requests import Session as ReqSession from requests import Session as ReqSession, Response as ReqResponse
from bs4.element import Tag as BS4Tag from bs4.element import Tag as BS4Tag
from .utils import PathT, yaml_overload from .utils import PathT, yaml_overload
@ -17,7 +17,21 @@ OptionsT = Dict[str, str]
class FormField: class FormField:
def __init__(self, name: str, default: DefaultInitT = None, options: OptionsT = None, required: bool = False): def __init__(self, name: str, default: DefaultInitT = None, options: OptionsT = None, required: bool = False):
self.name: str = name self.name: str = name
self.default = default self._default: Union[str, CallableDefaultT, None]
if isinstance(default, dict):
try:
module, function = default['module'], default['function']
except KeyError:
raise TypeError(f"Default for field '{name}' is invalid. The default must be either a string or a "
f"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)
self.options: Optional[OptionsT] = None self.options: Optional[OptionsT] = None
if options is not None: if options is not None:
self.options = {str(k): str(v) for k, v in options.items()} self.options = {str(k): str(v) for k, v in options.items()}
@ -41,23 +55,6 @@ class FormField:
def default(self) -> Optional[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
@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: 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()
@ -174,7 +171,7 @@ class Form:
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) -> str: 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`. Uses `aiohttp` to perform a POST request to `.url` with the form's payload generated using `kwargs`.
@ -185,20 +182,20 @@ class Form:
Passed directly into `.get_payload`. Passed directly into `.get_payload`.
Returns: Returns:
The response text from the request. The `aiohttp.ClientResponse` object from the request.
""" """
if self.url is None: if self.url is None:
raise AttributeError("`url` attribute not set") raise AttributeError("`url` attribute not set")
from aiohttp import ClientSession from aiohttp import ClientSession, ClientResponse
from webutils import in_async_session from webutils import in_async_session
@in_async_session @in_async_session
async def post(url: str, data: dict, session: ClientSession = None) -> str: async def post(url: str, data: dict, session: ClientSession = None) -> ClientResponse:
async with session.post(url, data=data) as response: async with session.post(url, data=data) as response:
return await response.text() return response
return await post(self.url, self.get_payload(**kwargs), session=_aiohttp_session_obj) 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: 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`. Uses `requests` to perform a POST request to `.url` with the form's payload generated using `kwargs`.
@ -209,14 +206,14 @@ class Form:
Passed directly into `.get_payload`. Passed directly into `.get_payload`.
Returns: Returns:
The response text from the request. The `requests.Response` object from the request.
""" """
if self.url is None: if self.url is None:
raise AttributeError("`url` attribute not set") raise AttributeError("`url` attribute not set")
if _requests_session_obj is not None: if _requests_session_obj is not None:
return _requests_session_obj.post(self.url, data=self.get_payload(**kwargs)).text return _requests_session_obj.post(self.url, data=self.get_payload(**kwargs))
from requests import post from requests import post
return post(self.url, data=self.get_payload(**kwargs)).text return post(self.url, data=self.get_payload(**kwargs))
def check_with_html(self, form_tag: 'BS4Tag', check_defaults: bool = True) -> None: def check_with_html(self, form_tag: 'BS4Tag', check_defaults: bool = True) -> None:
from .html import check_form_interface from .html import check_form_interface