Compare commits
2 Commits
3001caf7ce
...
e6c5076e39
Author | SHA1 | Date | |
---|---|---|---|
e6c5076e39 | |||
2f506d7afd |
@ -1,6 +1,6 @@
|
|||||||
[metadata]
|
[metadata]
|
||||||
name = yamlhttpforms
|
name = yamlhttpforms
|
||||||
version = 0.0.6
|
version = 0.0.8
|
||||||
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
|
||||||
|
@ -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, ClientResponse as AioResponse
|
from aiohttp import ClientSession as AioSession
|
||||||
from requests import Session as ReqSession, Response as ReqResponse
|
from requests import Session as ReqSession
|
||||||
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,21 +17,7 @@ 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: Union[str, CallableDefaultT, None]
|
self.default = default
|
||||||
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()}
|
||||||
@ -55,6 +41,23 @@ 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()
|
||||||
|
|
||||||
@ -171,7 +174,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) -> '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`.
|
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`.
|
Passed directly into `.get_payload`.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
The `aiohttp.ClientResponse` object from the request.
|
The response text 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, ClientResponse
|
from aiohttp import ClientSession
|
||||||
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) -> ClientResponse:
|
async def post(url: str, data: dict, session: ClientSession = None) -> str:
|
||||||
async with session.post(url, data=data) as response:
|
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)
|
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`.
|
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`.
|
Passed directly into `.get_payload`.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
The `requests.Response` object from the request.
|
The response text 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))
|
return _requests_session_obj.post(self.url, data=self.get_payload(**kwargs)).text
|
||||||
from requests import post
|
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:
|
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
|
||||||
|
Loading…
x
Reference in New Issue
Block a user