我们从Python开源项目中，提取了以下30个代码示例，用于说明如何使用http.HTTPStatus()。
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()
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.
