def _do_req(self, path: str, *, method: str = 'GET',
data: Optional[Any] = None):
"""
Performs a request against the instance eureka server.
:param path: URL Path, the hostname is prepended automatically
:param method: request method (put/post/patch/get/etc)
:param data: Optional data to be sent with the request, must
already be encoded appropriately.
:return: optional[dict[str, any]]
"""
url = self._eureka_url + path
logger.debug('Performing %s on %s with payload: %s', method, path,
data)
async with _SESSION.request(method, url, data=data) as resp:
if 400 <= resp.status < 600:
# noinspection PyArgumentList
raise EurekaException(HTTPStatus(resp.status),
await resp.text())
logger.debug('Result: %s', resp.status)
return await resp.json()
python类HTTPStatus()的实例源码
def update_warc_response_from_item(record, item):
"""update a WARC response record from a scrapy Item"""
h = record.header
h['WARC-Target-URI'] = item['url']
h['WARC-Date'] = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime(item['retrieved']))
h['X-Spider-Name'] = item['spider_name']
h['X-Spider-Run-ID'] = item['spider_run_id']
# XXX Scrapy doesn't provide remote IP for WARC-IP-Address
# below based on WARCRecord.from_response()
# XXX scrapy doesn't provide human-readable status string
status = "HTTP/1.1 {} {}".format(item['status'],
http.HTTPStatus(item['status']).name).encode()
headers = [b': '.join((k, v)) for k, l in item['headers'].iteritems() for v in l]
record.update_payload(b"\r\n".join(itertools.chain((status, ),
headers,
(b'', ),
(item['content'], )
)))
def __init__(self, status):
from http import HTTPStatus
self.status = status
try:
phrase = HTTPStatus(status).phrase
self.message = '????? HTTP ?? {} ({})'.format(status, phrase)
except ValueError:
self.message = '????? HTTP ?? {}'.format(status)
def get_status_line(status_code):
try:
phrase = http.HTTPStatus(status_code).phrase.encode()
except ValueError:
phrase = b''
return b''.join([
b'HTTP/1.1 ', str(status_code).encode(), b' ', phrase, b'\r\n'
])
def __init__(self, status: HTTPStatus, *args, **kwargs):
self._status = status
super().__init__(*args, **kwargs)
def status(self) -> HTTPStatus:
return self._status
def log_request(self, code, message):
if hasattr(http, 'HTTPStatus') and isinstance(code, http.HTTPStatus) and code.value < 400:
return
elif code[0] < '4':
return
super().log_request(code, message)
def __init__(self, *args: Any, status: http.HTTPStatus,
text: str = None) -> None:
super().__init__(*args)
self.response = web.Response(status=status.value, text=text)
def process(cls, server: ServerHost,
request: web.Request,
client: aiohttp.ClientSession) -> "ContribHost":
"""Process a request into a contribution."""
# This method exists because __init__() cannot be a coroutine.
raise ResponseExit(status=http.HTTPStatus.NOT_IMPLEMENTED) # pragma: no cover
def response(self, status_code, response):
"""
Describe a possible response for a Flask handler.
Args:
status_code: The status code for which the response is
described. `str`, `int` or `http.HTTPStatus` are
accepted. The result will be exposed as a string. See
:swagger:`responsesCode`.
response: A description of the response object. This may be
a dict describing a response or a string referring to a
response added using `add_response`. See
:swagger:`responseObject`.
"""
if isinstance(response, str):
response = ref('responses', response)
if isinstance(status_code, HTTPStatus):
status_code = int(status_code)
def attach_response(fn):
if not hasattr(fn, 'responses'):
fn.responses = {}
fn.responses[str(status_code)] = response
return fn
return attach_response
def __init__(self, status, headers, data):
self.headers = headers
self.data = data
self.status = http.HTTPStatus(status)
def test_is_core_dev():
teams = [{"name": "not Python core"}]
gh = FakeGH(getiter={"https://api.github.com/orgs/python/teams": teams})
with pytest.raises(ValueError):
await util.is_core_dev(gh, "brett")
teams = [{"name": "Python core", "id": 42}]
getitem = {"https://api.github.com/teams/42/memberships/brett": True}
gh = FakeGH(getiter={"https://api.github.com/orgs/python/teams": teams},
getitem=getitem)
assert await util.is_core_dev(gh, "brett")
assert gh.getiter_url == "https://api.github.com/orgs/python/teams"
teams = [{"name": "Python core", "id": 42}]
getitem = {"https://api.github.com/teams/42/memberships/andrea":
gidgethub.BadRequest(status_code=http.HTTPStatus(404))}
gh = FakeGH(getiter={"https://api.github.com/orgs/python/teams": teams},
getitem=getitem)
assert not await util.is_core_dev(gh, "andrea")
teams = [{"name": "Python core", "id": 42}]
getitem = {"https://api.github.com/teams/42/memberships/andrea":
gidgethub.BadRequest(status_code=http.HTTPStatus(400))}
gh = FakeGH(getiter={"https://api.github.com/orgs/python/teams": teams},
getitem=getitem)
with pytest.raises(gidgethub.BadRequest):
await util.is_core_dev(gh, "andrea")
def __init__(self, body='', headers=None,
status_code=200, status_phrase=None, cookies=None,
**kwargs):
headers = QueryList(headers or [])
if status_phrase is None:
status_phrase = HTTPStatus(status_code).phrase
super().__init__(
body, headers,
status_code, status_phrase,
cookies, **kwargs
)
def test_5XX(self):
status_code = 502
with pytest.raises(GitHubBroken) as exc_info:
sansio.decipher_response(status_code, {}, b'')
assert exc_info.value.status_code == http.HTTPStatus(status_code)
def test_4XX_no_message(self):
status_code = 400
with pytest.raises(BadRequest) as exc_info:
sansio.decipher_response(status_code, {}, b'')
assert exc_info.value.status_code == http.HTTPStatus(status_code)
def test_4XX_message(self):
status_code = 400
message = json.dumps({"message": "it went bad"}).encode("UTF-8")
headers = {"content-type": "application/json; charset=utf-8"}
with pytest.raises(BadRequest) as exc_info:
sansio.decipher_response(status_code, headers, message)
assert exc_info.value.status_code == http.HTTPStatus(status_code)
assert str(exc_info.value) == "it went bad"
def test_404(self):
status_code = 404
headers, body = sample("pr_not_found", status_code)
with pytest.raises(BadRequest) as exc_info:
sansio.decipher_response(status_code, headers, body)
assert exc_info.value.status_code == http.HTTPStatus(status_code)
assert str(exc_info.value) == "Not Found"
def test_403_rate_limit_exceeded(self):
status_code = 403
headers = {"content-type": "application/json; charset=utf-8",
"x-ratelimit-limit": "2",
"x-ratelimit-remaining": "0",
"x-ratelimit-reset": "1",
}
body = json.dumps({"message": "oops"}).encode("UTF-8")
with pytest.raises(RateLimitExceeded) as exc_info:
sansio.decipher_response(status_code, headers, body)
assert exc_info.value.status_code == http.HTTPStatus(status_code)
def test_422(self):
status_code = 422
errors = [{"resource": "Issue", "field": "title",
"code": "missing_field"}]
body = json.dumps({"message": "it went bad", "errors": errors})
body = body.encode("utf-8")
headers = {"content-type": "application/json; charset=utf-8"}
with pytest.raises(InvalidField) as exc_info:
sansio.decipher_response(status_code, headers, body)
assert exc_info.value.status_code == http.HTTPStatus(status_code)
assert str(exc_info.value) == "it went bad for 'title'"
def test_3XX(self):
status_code = 301
with pytest.raises(RedirectionException) as exc_info:
sansio.decipher_response(status_code, {}, b'')
assert exc_info.value.status_code == http.HTTPStatus(status_code)
def test_2XX_error(self):
status_code = 205
with pytest.raises(HTTPException) as exc_info:
sansio.decipher_response(status_code, {}, b'')
assert exc_info.value.status_code == http.HTTPStatus(status_code)
def from_http(cls, headers: Mapping, body: bytes,
*, secret: Optional[str] = None) -> "Event":
"""Construct an event from HTTP headers and JSON body data.
The mapping providing the headers is expected to support lowercase keys.
Since this method assumes the body of the HTTP request is JSON, a check
is performed for a content-type of "application/json" (GitHub does
support other content-types). If the content-type does not match,
BadRequest is raised.
If the appropriate headers are provided for event validation, then it
will be performed unconditionally. Any failure in validation
(including not providing a secret) will lead to ValidationFailure being
raised.
"""
if "x-hub-signature" in headers:
if secret is None:
raise ValidationFailure("secret not provided")
validate_event(body, signature=headers["x-hub-signature"],
secret=secret)
elif secret is not None:
raise ValidationFailure("signature is missing")
try:
data = _decode_body(headers["content-type"], body, strict=True)
except (KeyError, ValueError) as exc:
raise BadRequest(http.HTTPStatus(415),
"expected a content-type of "
"'application/json' or "
"'application/x-www-form-urlencoded'") from exc
return cls(data, event=headers["x-github-event"],
delivery_id=headers["x-github-delivery"])
def __init__(self, rate_limit: Any, *args: Any) -> None:
self.rate_limit = rate_limit
if not args:
super().__init__(http.HTTPStatus.FORBIDDEN,
"rate limit exceeded")
else:
super().__init__(http.HTTPStatus.FORBIDDEN, *args)
def __init__(self, errors: Any, *args: Any) -> None:
"""Store the error details."""
self.errors = errors
super().__init__(http.HTTPStatus.UNPROCESSABLE_ENTITY, *args)
def describe_http_status(code):
"""Return a string describing the given HTTP status code."""
try:
code = HTTPStatus(code)
except ValueError:
return "HTTP {code}".format(code=code)
else:
return "HTTP {code.value:d} {code.name}".format(code=code)
def test_opened_pr():
username = "brettcannon"
issue_url = "https://api.github.com/issue/42"
data = {
"action": "opened",
"pull_request": {
"user": {
"login": username,
},
"issue_url": issue_url,
}
}
event = sansio.Event(data, event="pull_request", delivery_id="12345")
teams = [
{"name": "python core", "id": 6}
]
items = {
f"https://api.github.com/teams/6/memberships/{username}": "OK",
issue_url: {"labels": [], "labels_url": "https://api.github.com/labels"}
}
gh = FakeGH(getiter={"https://api.github.com/orgs/python/teams": teams},
getitem=items)
await awaiting.router.dispatch(event, gh)
assert len(gh.post_) == 1
post_ = gh.post_[0]
assert post_[0] == "https://api.github.com/labels"
assert post_[1] == [awaiting.Blocker.merge.value]
username = "andreamcinnes"
issue_url = "https://api.github.com/issue/42"
data = {
"action": "opened",
"pull_request": {
"user": {
"login": username,
},
"issue_url": issue_url,
}
}
event = sansio.Event(data, event="pull_request", delivery_id="12345")
teams = [
{"name": "python core", "id": 6}
]
items = {
f"https://api.github.com/teams/6/memberships/{username}":
gidgethub.BadRequest(status_code=http.HTTPStatus(404)),
issue_url: {"labels": [], "labels_url": "https://api.github.com/labels"}
}
gh = FakeGH(getiter={"https://api.github.com/orgs/python/teams": teams},
getitem=items)
await awaiting.router.dispatch(event, gh)
assert len(gh.post_) == 1
post_ = gh.post_[0]
assert post_[0] == "https://api.github.com/labels"
assert post_[1] == [awaiting.Blocker.review.value]
def test_change_requested_for_non_core_dev():
data = {
"action": "submitted",
"review": {
"user": {
"login": "gvanrossum",
},
"state": "changes_requested".upper(),
},
"pull_request": {
"url": "https://api.github.com/pr/42",
"issue_url": "https://api.github.com/issue/42",
"comments_url": "https://api.github.com/comment/42",
"user": {
"login": "miss-islington"
}
},
}
event = sansio.Event(data, event="pull_request_review", delivery_id="12345")
teams = [
{"name": "python core", "id": 6}
]
items = {
f"https://api.github.com/teams/6/memberships/gvanrossum": True,
"https://api.github.com/teams/6/memberships/miss-islington":
gidgethub.BadRequest(status_code=http.HTTPStatus(404)),
"https://api.github.com/issue/42": {
"labels": [],
"labels_url": "https://api.github.com/labels/42",
}
}
iterators = {
"https://api.github.com/orgs/python/teams": teams,
"https://api.github.com/pr/42/reviews":
[{"user": {"login": "brettcannon"}, "state": "changes_requested"}],
}
gh = FakeGH(getiter=iterators, getitem=items)
await awaiting.router.dispatch(event, gh)
assert len(gh.post_) == 2
labeling = gh.post_[0]
assert labeling[0] == "https://api.github.com/labels/42"
assert labeling[1] == [awaiting.Blocker.changes.value]
message = gh.post_[1]
assert message[0] == "https://api.github.com/comment/42"
change_requested_message = awaiting.CHANGES_REQUESTED_MESSAGE.replace(
"{easter_egg}", "").strip()
assert change_requested_message in message[1]["body"]
def decipher_response(status_code: int, headers: Mapping,
body: bytes) -> Tuple[Any, RateLimit, Optional[str]]:
"""Decipher an HTTP response for a GitHub API request.
The mapping providing the headers is expected to support lowercase keys.
The parameters of this function correspond to the three main parts
of an HTTP response: the status code, headers, and body. Assuming
no errors which lead to an exception being raised, a 3-item tuple
is returned. The first item is the decoded body (typically a JSON
object, but possibly None or a string depending on the content
type of the body). The second item is an instance of RateLimit
based on what the response specified.
The last item of the tuple is the URL where to request the next
part of results. If there are no more results then None is
returned. Do be aware that the URL can be a URI template and so
may need to be expanded.
If the status code is anything other than 200, 201, or 204, then
an HTTPException is raised.
"""
data = _decode_body(headers.get("content-type"), body)
if status_code in {200, 201, 204}:
return data, RateLimit.from_http(headers), _next_link(headers.get("link"))
else:
try:
message = data["message"]
except (TypeError, KeyError):
message = None
exc_type: Type[HTTPException]
if status_code >= 500:
exc_type = GitHubBroken
elif status_code >= 400:
exc_type = BadRequest
if status_code == 403:
rate_limit = RateLimit.from_http(headers)
if not rate_limit.remaining:
raise RateLimitExceeded(rate_limit, message)
elif status_code == 422:
errors = data["errors"]
fields = ", ".join(repr(e["field"]) for e in errors)
message = f"{message} for {fields}"
raise InvalidField(errors, message)
elif status_code >= 300:
exc_type = RedirectionException
else:
exc_type = HTTPException
status_code_enum = http.HTTPStatus(status_code)
args: Tuple
if message:
args = status_code_enum, message
else:
args = status_code_enum,
raise exc_type(*args)
def reducedWebLogFormatter(timestamp, request):
"""Return a reduced formatted log line for the given request.
The `timestamp` argument is ignored. The line returned is expected to be
sent out by a logger which will add its own timestamp, so this one is
superfluous.
:see: `IAccessLogFormatter`
:see: `combinedLogFormatter`
"""
template = (
"{origin} {method} {uri} {proto} --> {status} "
"(referrer: {referrer}; agent: {agent})"
)
def field(value, default):
if value is None or len(value) == 0 or value.isspace():
return default
elif isinstance(value, bytes):
return value.decode("ascii", "replace")
else:
return value
def normaliseAddress(address):
"""Normalise an IP address."""
try:
address = IPAddress(address)
except AddrFormatError:
return address # Hostname?
else:
if address.is_ipv4_mapped():
return address.ipv4()
else:
return address
def describeHttpStatus(code):
try:
code = HTTPStatus(code)
except ValueError:
if isinstance(code, int):
return str(code)
else:
return "???"
else:
return "{code.value:d} {code.name}".format(code=code)
origin = field(request.getClientIP(), None)
origin = "-" if origin is None else normaliseAddress(origin)
return template.format(
referrer=field(request.getHeader(b"referer"), "-"),
agent=field(request.getHeader(b"user-agent"), "-"),
status=describeHttpStatus(request.code), origin=origin,
method=field(request.method, "???"), uri=field(request.uri, "-"),
proto=field(request.clientproto, "-"),
)
def search(query, web = False, useragent = "DuckDuckGo Python3 Api", skip_disambig = True, pretty = True, **kwargs):
'''
Searching the DuckDuckGo search engine and returning a Result object.
Keyworld Arguments:
query : Straightforward, the search string.
web : If true, opens the web browser and skips and command line. Default: False
useragent : Useragent to use while searching. Default: "DuckDuckGo Python3 Api". Can be overridden.
skip_disambig : Used to skip disambiguation. Default: 1 (True)
pretty : To make JSON look pretty. Default: 1 (True)
**kwargs : Any other parameters passed. As of writing this API no other parameters exists, helpful for future scaling.
'''
parameters = {'q': query, 'format': 'json', 'pretty': pretty, 'no_redirect': 1, 'no_html': 1, 'skip_disambig': skip_disambig}
encoded_parameters = urllib.parse.urlencode(parameters)
url = 'http://api.duckduckgo.com/?' + encoded_parameters
# Performs the Web browser search instead of instant answer API.
if web:
url = 'http://duckduckgo.com/?ia=web&' + urllib.parse.urlencode({'q': query})
webbrowser.open(url)
return
request = urllib.request.Request(url, headers = {'User-Agent': useragent})
try:
response = urllib.request.urlopen(request)
except urllib.error.HTTPError as e:
print("The server couldn't fulfill the request.")
print("Error code: {}, Response: {}.".format(e.code, http.HTTPStatus(e.code)))
except urllib.error.URLError as e:
print("We failed to reach the server.")
print("Reason: {}", e.reason)
else:
response = response.read().decode()
response = json.loads(response)
return Result(response)
# Below are the return fields as class objects.
