File: __init__.py

package info (click to toggle)
python-requests-unixsocket 1.0.1-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 256 kB
  • sloc: python: 316; sh: 23; makefile: 14
file content (85 lines) | stat: -rw-r--r-- 2,121 bytes parent folder | download
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)