Python requests 中如何处理不同的 HTTP 状态码

PythonBeginner
立即练习

介绍

本教程将指导你使用 Python requests 处理不同的 HTTP 状态码。HTTP 状态码对于理解 Web 请求是成功还是失败,以及如何正确响应不同的情况至关重要。通过完成这个实验(Lab),你将学习如何在与 Web 服务交互的 Python 应用程序中实现强大的错误处理。

你将从理解 HTTP 状态码开始,然后逐步构建你的技能,实现基本和高级的错误处理技术。这些知识对于开发能够优雅地处理各种服务器响应的可靠 Python Web 应用程序至关重要。

理解 HTTP 状态码和设置

什么是 HTTP 状态码?

HTTP 状态码是服务器响应客户端请求时返回的三位数字。它们指示特定的 HTTP 请求是否已成功完成。这些代码分为五类:

  • **1xx (信息性)**:请求已收到,并且正在处理
  • **2xx (成功)**:请求已成功接收、理解和接受
  • **3xx (重定向)**:必须采取进一步的操作才能完成请求
  • **4xx (客户端错误)**:请求包含错误的语法或无法完成
  • **5xx (服务器错误)**:服务器未能完成一个看似有效的请求

你将遇到的常见状态码包括:

状态码 名称 描述
200 OK 请求成功
201 Created 请求成功并创建了资源
400 Bad Request 服务器无法处理该请求
401 Unauthorized 需要身份验证
404 Not Found 找不到请求的资源
500 Internal Server Error 服务器遇到意外情况

设置你的 Python 环境

在开始处理 HTTP 状态码之前,让我们安装 Python requests 库并创建一个简单的测试脚本。

  1. 在 LabEx 环境中打开一个终端并运行:
pip install requests
  1. 为我们的项目文件创建一个新目录:
mkdir -p ~/project/http_status_lab
cd ~/project/http_status_lab
  1. 使用 WebIDE,在 http_status_lab 目录中创建一个名为 test_request.py 的新文件,其中包含以下基本代码:
import requests

def make_request(url):
    """Make a basic GET request to the specified URL."""
    response = requests.get(url)
    print(f"Status Code: {response.status_code}")
    print(f"Response: {response.text[:100]}...")  ## Print first 100 characters
    return response

## Test with a working URL
response = make_request("https://httpbin.org/status/200")
print(f"Request was successful: {response.ok}")
  1. 运行测试脚本以查看其运行情况:
python test_request.py

你应该看到类似这样的输出:

Status Code: 200
Response: ...
Request was successful: True

这确认你已成功设置了 Python 环境,并且可以发出基本的 HTTP 请求。在接下来的步骤中,你将学习如何处理不同的 HTTP 状态码并实现更高级的错误处理技术。

处理基本 HTTP 状态码

现在你已经了解了什么是 HTTP 状态码,并且设置好了你的环境,让我们为常见状态码实现基本的错误处理。

创建一个状态码测试工具

首先,让我们创建一个工具,它允许我们测试不同的 HTTP 状态码。网站 httpbin.org 提供了返回特定状态码的端点,这非常适合我们的测试。

在你的 http_status_lab 目录中创建一个名为 status_code_tester.py 的新文件,其中包含以下代码:

import requests
import sys

def test_status_code(status_code):
    """Test a specific HTTP status code using httpbin.org."""
    url = f"https://httpbin.org/status/{status_code}"
    try:
        response = requests.get(url)
        print(f"Status Code: {response.status_code}")
        print(f"Response OK: {response.ok}")
        return response
    except requests.exceptions.RequestException as e:
        print(f"Request failed: {e}")
        return None

if __name__ == "__main__":
    ## Get status code from command line or use default
    status_code = sys.argv[1] if len(sys.argv) > 1 else "200"
    test_status_code(status_code)

这个脚本使用特定的状态码向 httpbin.org 发送请求并打印结果。你可以使用命令行参数运行它来测试不同的状态码。

测试不同的状态码

让我们使用不同的状态码测试我们的脚本,看看 requests 库如何处理它们:

  1. 测试成功的请求(200 OK):
python status_code_tester.py 200

预期输出:

Status Code: 200
Response OK: True
  1. 测试客户端错误(404 Not Found):
python status_code_tester.py 404

预期输出:

Status Code: 404
Response OK: False
  1. 测试服务器错误(500 Internal Server Error):
python status_code_tester.py 500

预期输出:

Status Code: 500
Response OK: False

处理基本状态码类别

现在,让我们创建一个更全面的脚本,该脚本可以适当地处理不同类别的状态码。在你的 http_status_lab 目录中创建一个名为 basic_handler.py 的新文件:

import requests

def handle_response(url):
    """Handle different HTTP status codes with basic error handling."""
    try:
        response = requests.get(url)

        ## Check status code category
        if 200 <= response.status_code < 300:
            print(f"Success! Status code: {response.status_code}")
            return response
        elif 300 <= response.status_code < 400:
            print(f"Redirection! Status code: {response.status_code}")
            ## For redirection, you might want to follow the redirect
            return response
        elif 400 <= response.status_code < 500:
            print(f"Client error! Status code: {response.status_code}")
            ## Handle client errors
            return response
        elif 500 <= response.status_code < 600:
            print(f"Server error! Status code: {response.status_code}")
            ## Handle server errors
            return response
    except requests.exceptions.RequestException as e:
        print(f"Request failed: {e}")
        return None

## Test with different status codes
print("Testing 200 OK:")
handle_response("https://httpbin.org/status/200")

print("\nTesting 404 Not Found:")
handle_response("https://httpbin.org/status/404")

print("\nTesting 500 Internal Server Error:")
handle_response("https://httpbin.org/status/500")

运行脚本以查看它如何处理不同的状态码:

python basic_handler.py

预期输出:

Testing 200 OK:
Success! Status code: 200

Testing 404 Not Found:
Client error! Status code: 404

Testing 500 Internal Server Error:
Server error! Status code: 500

这个基本的处理程序将状态码分组为类别,并根据类别采取不同的操作。这是 Python Web 应用程序中的常见模式,允许你适当地处理不同类型的响应。

实现高级错误处理

现在你已经了解了处理 HTTP 状态码的基础知识,让我们为你的 Python 应用程序实现更高级的错误处理技术。

使用 raise_for_status()

requests 库提供了一个方便的方法,名为 raise_for_status(),它会为 4xx 和 5xx 状态码引发一个异常。这是一种在你的代码中处理错误的简单但有效的方法。

在你的 http_status_lab 目录中创建一个名为 raise_for_status_example.py 的新文件:

import requests

def fetch_data(url):
    """Fetch data from a URL with error handling using raise_for_status()."""
    try:
        response = requests.get(url)
        ## This will raise an HTTPError if the HTTP request returned an unsuccessful status code
        response.raise_for_status()
        ## If we get here, the request was successful
        print(f"Success! Status code: {response.status_code}")
        return response.json() if 'application/json' in response.headers.get('Content-Type', '') else response.text
    except requests.exceptions.HTTPError as e:
        print(f"HTTP Error: {e}")
        return None
    except requests.exceptions.ConnectionError as e:
        print(f"Connection Error: {e}")
        return None
    except requests.exceptions.Timeout as e:
        print(f"Timeout Error: {e}")
        return None
    except requests.exceptions.RequestException as e:
        print(f"Request Exception: {e}")
        return None

## Test with different URLs
print("Testing successful request:")
data = fetch_data("https://httpbin.org/json")
print(f"Received data type: {type(data)}")

print("\nTesting 404 error:")
data = fetch_data("https://httpbin.org/status/404")
print(f"Received data: {data}")

print("\nTesting 500 error:")
data = fetch_data("https://httpbin.org/status/500")
print(f"Received data: {data}")

运行脚本以查看 raise_for_status() 如何处理错误:

python raise_for_status_example.py

预期输出将显示该库自动为非 2xx 状态码引发异常,然后你的代码可以捕获并适当地处理这些异常。

创建自定义异常处理程序

对于更复杂的应用程序,你可能希望创建自定义异常处理程序以提供更具体的错误处理。让我们创建一个更高级的示例:

在你的 http_status_lab 目录中创建一个名为 custom_exception_handler.py 的新文件:

import requests
import logging

## Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

## Custom exceptions
class APIError(Exception):
    """Base class for API request exceptions."""
    pass

class ClientError(APIError):
    """Exception raised for client errors (4xx)."""
    pass

class ServerError(APIError):
    """Exception raised for server errors (5xx)."""
    pass

class NotFoundError(ClientError):
    """Exception raised for 404 Not Found errors."""
    pass

class UnauthorizedError(ClientError):
    """Exception raised for 401 Unauthorized errors."""
    pass

def api_request(url, method="get", **kwargs):
    """Make an API request with comprehensive error handling."""
    try:
        ## Make the request using the specified HTTP method
        response = getattr(requests, method.lower())(url, **kwargs)

        ## Log the response status
        logger.info(f"Request to {url} returned status code {response.status_code}")

        ## Handle different status codes
        if 200 <= response.status_code < 300:
            return response
        elif response.status_code == 401:
            raise UnauthorizedError(f"Authentication required: {response.text}")
        elif response.status_code == 404:
            raise NotFoundError(f"Resource not found: {url}")
        elif 400 <= response.status_code < 500:
            raise ClientError(f"Client error {response.status_code}: {response.text}")
        elif 500 <= response.status_code < 600:
            raise ServerError(f"Server error {response.status_code}: {response.text}")
        else:
            raise APIError(f"Unexpected status code {response.status_code}: {response.text}")

    except requests.exceptions.RequestException as e:
        logger.error(f"Request failed: {e}")
        raise APIError(f"Request failed: {e}")

## Test the custom exception handler
def test_api_request(url):
    """Test the api_request function with error handling."""
    try:
        response = api_request(url)
        print(f"Success! Status code: {response.status_code}")
        print(f"Response: {response.text[:100]}...")  ## Print first 100 characters
        return response
    except UnauthorizedError as e:
        print(f"Authentication Error: {e}")
    except NotFoundError as e:
        print(f"Not Found Error: {e}")
    except ClientError as e:
        print(f"Client Error: {e}")
    except ServerError as e:
        print(f"Server Error: {e}")
    except APIError as e:
        print(f"API Error: {e}")
    return None

## Test with different status codes
print("Testing 200 OK:")
test_api_request("https://httpbin.org/status/200")

print("\nTesting 404 Not Found:")
test_api_request("https://httpbin.org/status/404")

print("\nTesting 401 Unauthorized:")
test_api_request("https://httpbin.org/status/401")

print("\nTesting 500 Internal Server Error:")
test_api_request("https://httpbin.org/status/500")

运行脚本以查看自定义异常处理的工作方式:

python custom_exception_handler.py

这个高级示例演示了几个重要的概念:

  1. 扩展了基本 Exception 类的自定义异常类
  2. 用于不同类型错误的层次结构异常
  3. 用于调试和监视的详细日志记录
  4. 针对不同状态码的特定处理

这种方法允许你以更结构化和可维护的方式处理不同的错误类型,这对于更大的应用程序至关重要。

构建一个完整的 Web API 客户端

在最后一步中,我们将把你所学的一切整合起来,构建一个具有强大错误处理功能的完整 Web API 客户端。我们将为 JSONPlaceholder API 创建一个客户端,这是一个你可以用于测试的免费在线 REST API。

创建一个简单的 API 客户端

让我们在你的 http_status_lab 目录中创建一个名为 api_client.py 的文件:

import requests
import json
import logging

## Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class APIClient:
    """A simple API client with robust error handling."""

    def __init__(self, base_url):
        """Initialize the API client with a base URL."""
        self.base_url = base_url
        self.session = requests.Session()

    def request(self, endpoint, method="get", params=None, data=None, headers=None):
        """Make a request to the API with error handling."""
        url = f"{self.base_url}{endpoint}"

        ## Default headers for JSON APIs
        if headers is None:
            headers = {
                'Content-Type': 'application/json',
                'Accept': 'application/json'
            }

        ## Convert data to JSON if it's a dictionary
        json_data = None
        if data and isinstance(data, dict):
            json_data = data
            data = None

        try:
            ## Log the request
            logger.info(f"Making {method.upper()} request to {url}")

            ## Make the request
            response = self.session.request(
                method=method,
                url=url,
                params=params,
                data=data,
                json=json_data,
                headers=headers
            )

            ## Log the response status
            logger.info(f"Received response with status code {response.status_code}")

            ## Check for HTTP errors
            response.raise_for_status()

            ## Parse JSON response if applicable
            try:
                return response.json()
            except json.JSONDecodeError:
                return response.text

        except requests.exceptions.HTTPError as e:
            status_code = e.response.status_code
            error_message = f"HTTP Error: {status_code}"

            try:
                ## Try to get more details from the response
                error_data = e.response.json()
                error_message = f"{error_message} - {error_data.get('message', str(error_data))}"
            except (json.JSONDecodeError, AttributeError):
                error_message = f"{error_message} - {e.response.text if hasattr(e, 'response') else str(e)}"

            logger.error(error_message)

            ## Re-raise with more context
            if 400 <= status_code < 500:
                logger.error(f"Client error: {error_message}")
                raise Exception(f"Client error (HTTP {status_code}): {error_message}")
            elif 500 <= status_code < 600:
                logger.error(f"Server error: {error_message}")
                raise Exception(f"Server error (HTTP {status_code}): {error_message}")
            else:
                raise

        except requests.exceptions.ConnectionError as e:
            logger.error(f"Connection Error: {e}")
            raise Exception(f"Connection Error: Could not connect to {url}")

        except requests.exceptions.Timeout as e:
            logger.error(f"Timeout Error: {e}")
            raise Exception(f"Timeout Error: Request to {url} timed out")

        except requests.exceptions.RequestException as e:
            logger.error(f"Request Exception: {e}")
            raise Exception(f"Request Error: {str(e)}")

    def get(self, endpoint, params=None, headers=None):
        """Make a GET request to the API."""
        return self.request(endpoint, method="get", params=params, headers=headers)

    def post(self, endpoint, data=None, headers=None):
        """Make a POST request to the API."""
        return self.request(endpoint, method="post", data=data, headers=headers)

    def put(self, endpoint, data=None, headers=None):
        """Make a PUT request to the API."""
        return self.request(endpoint, method="put", data=data, headers=headers)

    def delete(self, endpoint, headers=None):
        """Make a DELETE request to the API."""
        return self.request(endpoint, method="delete", headers=headers)

## Test the API client with JSONPlaceholder
def test_api_client():
    """Test the API client with JSONPlaceholder."""
    client = APIClient("https://jsonplaceholder.typicode.com")

    try:
        ## Get a list of posts
        print("\nGetting posts:")
        posts = client.get("/posts")
        print(f"Retrieved {len(posts)} posts")
        print(f"First post: {posts[0]}")

        ## Get a single post
        print("\nGetting a single post:")
        post = client.get("/posts/1")
        print(f"Retrieved post: {post}")

        ## Get a non-existent post (404)
        print("\nTrying to get a non-existent post:")
        try:
            client.get("/posts/999999")
        except Exception as e:
            print(f"Expected error: {e}")

        ## Create a new post
        print("\nCreating a new post:")
        new_post = client.post("/posts", {
            "title": "My New Post",
            "body": "This is the content of my new post.",
            "userId": 1
        })
        print(f"Created post: {new_post}")

        ## Update a post
        print("\nUpdating a post:")
        updated_post = client.put("/posts/1", {
            "id": 1,
            "title": "Updated Title",
            "body": "Updated content.",
            "userId": 1
        })
        print(f"Updated post: {updated_post}")

        ## Delete a post
        print("\nDeleting a post:")
        delete_response = client.delete("/posts/1")
        print(f"Delete response: {delete_response}")

    except Exception as e:
        print(f"An error occurred: {e}")

if __name__ == "__main__":
    test_api_client()

运行脚本以查看我们完整的 API 客户端如何处理不同的场景:

python api_client.py

测试错误处理

让我们创建一个单独的文件来测试我们的 API 客户端的错误处理能力。在你的 http_status_lab 目录中创建一个名为 test_error_handling.py 的文件:

from api_client import APIClient
import time

def test_different_errors():
    """Test different error scenarios with our API client."""

    ## Test with a valid API
    print("\n1. Testing with a valid API:")
    valid_client = APIClient("https://jsonplaceholder.typicode.com")
    try:
        data = valid_client.get("/posts/1")
        print(f"Success! Retrieved data: {data}")
    except Exception as e:
        print(f"Unexpected error: {e}")

    ## Test with a 404 error
    print("\n2. Testing with a non-existent endpoint (404):")
    try:
        valid_client.get("/non_existent_endpoint")
    except Exception as e:
        print(f"Expected error: {e}")

    ## Test with an invalid host
    print("\n3. Testing with an invalid host (Connection Error):")
    invalid_client = APIClient("https://this-does-not-exist-123456789.com")
    try:
        invalid_client.get("/anything")
    except Exception as e:
        print(f"Expected error: {e}")

    ## Test with a timeout
    print("\n4. Testing with a timeout:")
    timeout_client = APIClient("https://httpbin.org")
    try:
        ## httpbin.org/delay/5 will delay the response for 5 seconds
        ## but we'll set the timeout to 2 seconds
        timeout_client.session.request = lambda **kwargs: timeout_client.session.request_original(
            **{**kwargs, 'timeout': 2}
        )
        timeout_client.session.request_original = timeout_client.session.request
        timeout_client.get("/delay/5")
    except Exception as e:
        print(f"Expected error: {e}")

if __name__ == "__main__":
    test_different_errors()

上面的脚本可能会由于超时处理而产生错误,因为它试图在运行时修改一个方法。让我们简化它以避免这些问题:

from api_client import APIClient

def test_different_errors():
    """Test different error scenarios with our API client."""

    ## Test with a valid API
    print("\n1. Testing with a valid API:")
    valid_client = APIClient("https://jsonplaceholder.typicode.com")
    try:
        data = valid_client.get("/posts/1")
        print(f"Success! Retrieved data: {data}")
    except Exception as e:
        print(f"Unexpected error: {e}")

    ## Test with a 404 error
    print("\n2. Testing with a non-existent endpoint (404):")
    try:
        valid_client.get("/non_existent_endpoint")
    except Exception as e:
        print(f"Expected error: {e}")

    ## Test with a server error
    print("\n3. Testing with a server error (500):")
    error_client = APIClient("https://httpbin.org")
    try:
        error_client.get("/status/500")
    except Exception as e:
        print(f"Expected error: {e}")

if __name__ == "__main__":
    test_different_errors()

运行错误测试脚本:

python test_error_handling.py

这个脚本演示了我们的 API 客户端如何处理不同的错误场景,为实际应用提供了坚实的基础。

主要收获

通过构建这个完整的 API 客户端,你已经学习了几种重要的技术:

  1. 创建可重用的 API 客户端类
  2. 实现全面的错误处理
  3. 记录请求和响应以进行调试
  4. 解析不同的响应格式
  5. 处理不同的 HTTP 方法(GET,POST,PUT,DELETE)

这些技能对于构建与 Web API 交互的强大 Python 应用程序至关重要,确保你的代码可以优雅地处理各种错误场景,并向用户提供有意义的错误消息。

总结

在这个实验中,你学习了如何在 Python 的 requests 库中处理 HTTP 状态码,从基本技术到高级技术。你现在具备以下技能:

  • 了解不同类别的 HTTP 状态码及其含义
  • 使用状态码检查实现基本的错误处理
  • 使用 raise_for_status() 进行简单的错误处理
  • 创建自定义异常类以进行更具体的错误处理
  • 构建一个具有全面错误处理功能的完整 API 客户端

这些技能对于开发与 Web 服务交互的强大 Python 应用程序至关重要。通过正确处理 HTTP 状态码和错误,你可以创建更可靠的应用程序,从而提供更好的用户体验,并且更易于维护。

请记住,在你的 Python 应用程序中使用 HTTP 请求时,始终要考虑错误处理。这是 Web 开发的一个关键方面,它区分了专业代码和业余实现。