http

FetchedResponse dataclass

Result object for fetch_response.

Source code in m/core/http.py

@dataclass
class FetchedResponse:
    """Result object for fetch_response."""

    response: httplib.HTTPResponse
    body: str

HttpMethod

Bases: str, Enum

Python 3.10 does not have this enum.

https://docs.python.org/3/library/http.html#http-methods

Source code in m/core/http.py

class HttpMethod(str, Enum):  # noqa: WPS600
    """Python 3.10 does not have this enum.

    https://docs.python.org/3/library/http.html#http-methods
    """

    get = 'GET'
    post = 'POST'
    put = 'PUT'

    def __str__(self) -> str:
        """Return the value of the enum.

        Returns:
            The value of the enum.
        """
        return self.value

__str__()

Return the value of the enum.

Returns:

Type	Description
str	The value of the enum.

Source code in m/core/http.py

def __str__(self) -> str:
    """Return the value of the enum.

    Returns:
        The value of the enum.
    """
    return self.value

fetch(url, headers, method=HttpMethod.get, body=None)

Send an http(s) request.

Parameters:

Name	Type	Description	Default
url	str	The url to request.	required
headers	Mapping[str, str]	The headers for the request. By default it sets the user-agent to "m".	required
method	HttpMethod	The request method type. Defaults to GET.	get
body	Optional[str]	The body of the request.	None

Returns:

Type	Description
OneOf[Issue, str]	A OneOf containing the raw response from the server or an Issue.

Source code in m/core/http.py

def fetch(
    url: str,
    headers: Mapping[str, str],
    method: HttpMethod = HttpMethod.get,
    body: Optional[str] = None,
) -> OneOf[Issue, str]:
    """Send an http(s) request.

    Args:
        url:
            The url to request.
        headers:
            The headers for the request. By default it sets the `user-agent`
            to "m".
        method:
            The request method type. Defaults to `GET`.
        body:
            The body of the request.

    Returns:
        A `OneOf` containing the raw response from the server or an Issue.
    """
    return one_of(lambda: [
        fetch_res.body
        for fetch_res in fetch_response(url, headers, method, body)
    ])

fetch_json(url, headers, method=HttpMethod.get, body_json=None)

Specialized fetch to deal with json data.

Parameters:

Name	Type	Description	Default
url	str	The url to request.	required
headers	Mapping[str, str]	Additional headers for the request. By default it will add proper accept and content-type headers for json requests.	required
method	HttpMethod	The request method type. Defaults to GET.	get
body_json	Any	The data to send to the server (python object).	None

Returns:

Type	Description
OneOf[Issue, Any]	A OneOf containing a json parsed response from the server or an
OneOf[Issue, Any]	Issue.

Source code in m/core/http.py

def fetch_json(
    url: str,
    headers: Mapping[str, str],
    method: HttpMethod = HttpMethod.get,
    body_json: Any = None,
) -> OneOf[Issue, Any]:
    """Specialized fetch to deal with json data.

    Args:
        url:
            The url to request.
        headers:
            Additional headers for the request. By default it will add
            proper accept and content-type headers for json requests.
        method:
            The request method type. Defaults to `GET`.
        body_json:
            The data to send to the server (python object).

    Returns:
        A `OneOf` containing a json parsed response from the server or an
        Issue.
    """
    body = builtin_json.dumps(body_json) if body_json else None
    fetch_headers = {
        'accept': 'application/json',
        'content-type': 'application/json',
        **headers,
    }
    return one_of(lambda: [
        response
        for payload in fetch(url, fetch_headers, method, body)
        for response in parse_json(payload)
    ])

fetch_response(url, headers, method=HttpMethod.get, body=None)

Send an http(s) request.

Parameters:

Name	Type	Description	Default
url	str	The url to request.	required
headers	Mapping[str, str]	The headers for the request. By default it sets the user-agent to "m".	required
method	HttpMethod	The request method type. Defaults to GET.	get
body	Optional[str]	The body of the request.	None

Returns:

Type	Description
OneOf[Issue, FetchedResponse]	A OneOf containing the response object from the server or an Issue.

Source code in m/core/http.py

def fetch_response(
    url: str,
    headers: Mapping[str, str],
    method: HttpMethod = HttpMethod.get,
    body: Optional[str] = None,
) -> OneOf[Issue, FetchedResponse]:
    """Send an http(s) request.

    Args:
        url:
            The url to request.
        headers:
            The headers for the request. By default it sets the `user-agent`
            to "m".
        method:
            The request method type. Defaults to `GET`.
        body:
            The body of the request.

    Returns:
        A `OneOf` containing the response object from the server or an Issue.
    """
    parts = urlparse(url)
    protocol, hostname, path = [parts.scheme, parts.netloc, parts.path]
    path = f'{path}?{parts.query}' if parts.query else path
    fetch_headers = {'user-agent': 'm', **headers}
    ctxt: dict[str, str] = {'url': f'{hostname}{path}', 'method': f'{method}'}
    if body:
        fetch_headers['content-length'] = str(len(body))
        if 'DEBUG_HTTP_INCLUDE_BODY' in os.environ:
            ctxt['body'] = body
    connection = _get_connection(protocol, hostname)
    # See the next link for explanation disabling WPS440:
    #  https://github.com/wemake-services/wemake-python-styleguide/issues/1416
    try:
        connection.request(f'{method}', path, body, fetch_headers)
    except Exception as ex:
        return issue(f'{protocol} request failure', cause=ex, context=ctxt)
    try:
        res = connection.getresponse()
    except Exception as ex:  # noqa: WPS440
        return issue(f'{protocol} response failure', cause=ex, context=ctxt)
    try:
        res_body = res.read().decode()
    except Exception as ex:  # noqa: WPS440
        return issue(f'{protocol} read failure', cause=ex, context=ctxt)
    code = res.getcode()
    if HTTPStatus.OK <= code < HTTPStatus.MULTIPLE_CHOICES:
        return Good(FetchedResponse(
            response=res,
            body=res_body,
        ))
    return issue(
        f'{protocol} request failure ({code})',
        context={
            'body': body,
            'code': code,
            'res_body': res_body,
            **ctxt,
        },
    )