Firepup650/bitbot-3.11-fork
2
0
Fork 0
Code Issues Pull requests Activity Actions

change utils.http.Request to be a dataclass

Browse source
This commit is contained in:
jesopo 2019-11-25 13:42:10 +00:00
parent 985e4704a0
commit ab8bc65cc9
1 changed files with 30 additions and 38 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

68
src/utils/http.py
View file

@ -1,5 +1,5 @@
import asyncio, codecs, ipaddress, re, signal, socket, traceback, typing import asyncio, codecs, dataclasses, ipaddress, re, signal, socket, traceback
import urllib.error, urllib.parse, uuid import typing, urllib.error, urllib.parse, uuid
import json as _json import json as _json
import bs4, netifaces, requests, tornado.httpclient import bs4, netifaces, requests, tornado.httpclient
from src import IRCBot, utils from src import IRCBot, utils
@ -53,46 +53,37 @@ class HTTPWrongContentTypeException(HTTPException):
def throw_timeout(): def throw_timeout():
raise HTTPTimeoutException() raise HTTPTimeoutException()
@dataclasses.dataclass
class Request(object): class Request(object):
def __init__(self, url: str, url: str
get_params: typing.Dict[str, str]={}, post_data: typing.Any=None, id: typing.Optional[str] = None
headers: typing.Dict[str, str]={}, method: str = "GET"
json: bool=False, json_body: bool=False, allow_redirects: bool=True, get_params: typing.Dict[str, str] = dataclasses.field(
check_content_type: bool=True, parse: bool=False, default_factory=dict)
detect_encoding: bool=True, post_data: typing.Any = None
headers: typing.Dict[str, str] = dataclasses.field(
default_factory=dict)
cookies: typing.Dict[str, str] = dataclasses.field(
default_factory=dict)
method: str="GET", parser: str="lxml", id: str=None, json: bool = False
fallback_encoding: str=None, content_type: str=None, json_body: typing.Any = None
proxy: str=None, useragent: str=None,
**kwargs): allow_redirects: bool = True
self.id = id or str(uuid.uuid4()) check_content_type: bool = True
parse: bool = False
detect_encoding: bool = True
parser: str = "lxml"
fallback_encoding: typing.Optional[str] = None
content_type: typing.Optional[str] = None
proxy: typing.Optional[str] = None
useragent: typing.Optional[str] = None
self.set_url(url) def validate(self):
self.method = method.upper() self.id = self.id or str(uuid.uuid4())
self.get_params = get_params self.set_url(self.url)
self.post_data = post_data self.method = self.method.upper()
self.headers = headers
self.json = json
self.json_body = json_body
self.allow_redirects = allow_redirects
self.check_content_type = check_content_type
self.parse = parse
self.detect_encoding = detect_encoding
self.parser = parser
self.fallback_encoding = fallback_encoding
self.content_type = content_type
self.proxy = proxy
self.useragent = useragent
if kwargs:
if method == "POST":
self.post_data = kwargs
else:
self.get_params.update(kwargs)
def set_url(self, url: str): def set_url(self, url: str):
parts = urllib.parse.urlparse(url) parts = urllib.parse.urlparse(url)
@ -166,7 +157,7 @@ def request(request_obj: typing.Union[str, Request], **kwargs) -> Response:
return _request(request_obj) return _request(request_obj)
def _request(request_obj: Request) -> Response: def _request(request_obj: Request) -> Response:
request_obj.validate()
def _wrap() -> Response: def _wrap() -> Response:
headers = request_obj.get_headers() headers = request_obj.get_headers()
response = requests.request( response = requests.request(
@ -241,6 +232,7 @@ def request_many(requests: typing.List[Request]) -> typing.Dict[str, Response]:
responses = {} responses = {}
async def _request(request): async def _request(request):
request.validate()
client = tornado.httpclient.AsyncHTTPClient() client = tornado.httpclient.AsyncHTTPClient()
url = request.url url = request.url
if request.get_params: if request.get_params: