1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85
|
import sys
import requests
from .adapters import UnixAdapter
DEFAULT_SCHEME = "http+unix://"
class Session(requests.Session):
def __init__(self, url_scheme=DEFAULT_SCHEME, *args, **kwargs):
super().__init__(*args, **kwargs)
self.mount(url_scheme, UnixAdapter())
class monkeypatch:
def __init__(self, url_scheme=DEFAULT_SCHEME):
self.session = Session()
requests = self._get_global_requests_module()
# Methods to replace
self.methods = (
"request",
"get",
"head",
"post",
"patch",
"put",
"delete",
"options",
)
# Store the original methods
self.orig_methods = {m: requests.__dict__[m] for m in self.methods}
# Monkey patch
g = globals()
for m in self.methods:
requests.__dict__[m] = g[m]
def _get_global_requests_module(self):
return sys.modules["requests"]
def __enter__(self):
return self
def __exit__(self, *args):
requests = self._get_global_requests_module()
for m in self.methods:
requests.__dict__[m] = self.orig_methods[m]
# These are the same methods defined for the global requests object
def request(method, url, **kwargs):
session = Session()
return session.request(method=method, url=url, **kwargs)
def get(url, **kwargs):
kwargs.setdefault("allow_redirects", True)
return request("get", url, **kwargs)
def head(url, **kwargs):
kwargs.setdefault("allow_redirects", False)
return request("head", url, **kwargs)
def post(url, data=None, json=None, **kwargs):
return request("post", url, data=data, json=json, **kwargs)
def patch(url, data=None, **kwargs):
return request("patch", url, data=data, **kwargs)
def put(url, data=None, **kwargs):
return request("put", url, data=data, **kwargs)
def delete(url, **kwargs):
return request("delete", url, **kwargs)
def options(url, **kwargs):
kwargs.setdefault("allow_redirects", True)
return request("options", url, **kwargs)
|