Compare commits

...

2 Commits

Author SHA1 Message Date
e6c5076e39 post requests return text only 2022-01-02 00:40:23 +01:00
2f506d7afd default value setter 2022-01-02 00:17:47 +01:00
2 changed files with 30 additions and 27 deletions

View File

@ -1,6 +1,6 @@
[metadata]
name = yamlhttpforms
version = 0.0.6
version = 0.0.8
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, ClientResponse as AioResponse
from requests import Session as ReqSession, Response as ReqResponse
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
@ -17,21 +17,7 @@ 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: 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.default = default
self.options: Optional[OptionsT] = None
if options is not None:
self.options = {str(k): str(v) for k, v in options.items()}
@ -55,6 +41,23 @@ 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()
@ -171,7 +174,7 @@ class Form:
payload[field.name] = field.default
return payload
async def post_aio(self, _aiohttp_session_obj: 'AioSession' = None, **kwargs: str) -> 'AioResponse':
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`.
@ -182,20 +185,20 @@ class Form:
Passed directly into `.get_payload`.
Returns:
The `aiohttp.ClientResponse` object from the request.
The response text from the request.
"""
if self.url is None:
raise AttributeError("`url` attribute not set")
from aiohttp import ClientSession, ClientResponse
from aiohttp import ClientSession
from webutils import in_async_session
@in_async_session
async def post(url: str, data: dict, session: ClientSession = None) -> ClientResponse:
async def post(url: str, data: dict, session: ClientSession = None) -> str:
async with session.post(url, data=data) as response:
return 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) -> 'ReqResponse':
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`.
@ -206,14 +209,14 @@ class Form:
Passed directly into `.get_payload`.
Returns:
The `requests.Response` object from the request.
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))
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))
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