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]
name = yamlhttpforms
version = 0.0.8
version = 0.0.6
author = Daniil F.
author_email = mail@placeholder123.to
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
if TYPE_CHECKING:
from aiohttp import ClientSession as AioSession
from requests import Session as ReqSession
from aiohttp import ClientSession as AioSession, ClientResponse as AioResponse
from requests import Session as ReqSession, Response as ReqResponse
from bs4.element import Tag as BS4Tag
from .utils import PathT, yaml_overload
@ -17,7 +17,21 @@ 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._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
if options is not None:
self.options = {str(k): str(v) for k, v in options.items()}
@ -41,23 +55,6 @@ class FormField:
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()
@ -174,7 +171,7 @@ class Form:
payload[field.name] = field.default
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`.
@ -185,20 +182,20 @@ class Form:
Passed directly into `.get_payload`.
Returns:
The response text from the request.
The `aiohttp.ClientResponse` object from the request.
"""
if self.url is None:
raise AttributeError("`url` attribute not set")
from aiohttp import ClientSession
from aiohttp import ClientSession, ClientResponse
from webutils import 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:
return await response.text()
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) -> 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`.
@ -209,14 +206,14 @@ class Form:
Passed directly into `.get_payload`.
Returns:
The response text from the request.
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)).text
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)).text
return post(self.url, data=self.get_payload(**kwargs))
def check_with_html(self, form_tag: 'BS4Tag', check_defaults: bool = True) -> None:
from .html import check_form_interface