Django 调用外部http服务的实现

 更新时间:2026年10月16日 09:12:35   作者:alden_ygq  
本文介绍在Django中调用外部HTTP服务的方法,涵盖使用Python内置库、第三方库,异步请求方式,以及在视图中集成的示例,感兴趣的可以了解一下

在 Django 中调用外部 HTTP 服务获取数据是常见需求,可通过多种方式实现。以下是详细的实现方法、最佳实践及完整示例:

一、使用 Python 内置库(urllib)

适合简单请求,无需额外依赖:

from urllib import request
import json
def fetch_data():
    try:
        url = "https://api.example.com/data"
        response = request.urlopen(url)
        data = json.loads(response.read().decode("utf-8"))
        return data
    except Exception as e:
        print(f"请求失败: {e}")
        return None

二、使用第三方库(推荐requests)

1. 安装依赖

pip install requests 

2. 基础用法

import requests
def fetch_data():
    try:
        url = "https://api.example.com/data"
        response = requests.get(url)
        response.raise_for_status()  # 检查请求是否成功
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"请求失败: {e}")
        return None

3. 带参数的请求

def fetch_data_with_params():
    url = "https://api.example.com/search"
    params = {"query": "python", "page": 1}
    headers = {"Authorization": "Bearer your_token"}
    response = requests.get(url, params=params, headers=headers)
    return response.json()

4. POST 请求示例

def post_data():
    url = "https://api.example.com/submit"
    data = {"name": "John", "age": 30}
    headers = {"Content-Type": "application/json"}
    response = requests.post(url, json=data, headers=headers)
    return response.json()

三、异步请求(使用aiohttp或requests-futures)

1. 使用aiohttp(异步非阻塞)

pip install aiohttp
import asyncio
import aiohttp
async def fetch_data_async():
    async with aiohttp.ClientSession() as session:
        async with session.get("https://api.example.com/data") as response:
            return await response.json()
# 在视图中调用(需使用 async/await)
async def my_view(request):
    data = await fetch_data_async()
    return JsonResponse(data)

2. 使用requests-futures(线程池异步)

pip install requests-futures
from requests_futures.sessions import FuturesSession
def fetch_data_async():
    session = FuturesSession(max_workers=5)
    future = session.get("https://api.example.com/data")
    response = future.result()  # 阻塞直到获取结果
    return response.json()

四、在 Django 视图中集成

1. 同步视图示例

from django.http import JsonResponse
import requests
def api_view(request):
    try:
        response = requests.get("https://api.example.com/data")
        response.raise_for_status()
        return JsonResponse(response.json())
    except requests.exceptions.RequestException as e:
        return JsonResponse({"error": str(e)}, status=500)

2. 异步视图示例(Django 3.1+)

from django.http import JsonResponse
import aiohttp
import asyncio
async def async_api_view(request):
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get("https://api.example.com/data") as response:
                data = await response.json()
                return JsonResponse(data)
    except Exception as e:
        return JsonResponse({"error": str(e)}, status=500)

五、错误处理与重试机制

1. 完整错误处理

import requests
from requests.exceptions import Timeout, HTTPError, RequestException
def fetch_data_safe():
    try:
        response = requests.get(
            "https://api.example.com/data",
            timeout=5,  # 设置超时时间
            headers={"User-Agent": "MyDjangoApp/1.0"}
        )
        response.raise_for_status()  # 检查HTTP状态码
        return response.json()
    except Timeout:
        print("请求超时")
        return None
    except HTTPError as e:
        print(f"HTTP错误: {e.response.status_code}")
        return None
    except RequestException as e:
        print(f"请求异常: {e}")
        return None

2. 自动重试机制(使用requests+tenacity)

pip install tenacity
from tenacity import retry, stop_after_attempt, wait_fixed
@retry(stop=stop_after_attempt(3), wait=wait_fixed(2))
def fetch_data_with_retry():
    response = requests.get("https://api.example.com/data")
    response.raise_for_status()
    return response.json()

六、性能优化

1. 连接池管理

from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
def get_session():
    session = requests.Session()
    retries = Retry(total=3, backoff_factor=1)
    session.mount("https://", HTTPAdapter(max_retries=retries))
    return session
# 在视图中使用
def my_view(request):
    session = get_session()
    response = session.get("https://api.example.com/data")
    return JsonResponse(response.json())

2. 缓存机制(减少重复请求)

from django.core.cache import cache
def cached_fetch_data():
    data = cache.get("external_api_data")
    if data is None:
        response = requests.get("https://api.example.com/data")
        data = response.json()
        cache.set("external_api_data", data, 60 * 10)  # 缓存10分钟
    return data

七、安全注意事项

  1. 敏感信息保护:

    • 不要在代码中硬编码 API 密钥或密码
    • 使用环境变量或 Django 配置文件存储敏感信息
  2. 输入验证:

    • 验证外部 API 返回的数据格式
    • 使用 serializers 或 pydantic 进行数据验证
  3. 防止 DOS 攻击:

    • 设置合理的超时时间(建议 5-10 秒)
    • 限制并发请求数量
    • 实现请求频率限制

八、生产环境建议

1)使用专用服务层:

# services/external_api.py
import requests
class ExternalAPIService:
    def __init__(self):
        self.base_url = "https://api.example.com"
        self.session = requests.Session()
        self.session.headers.update({"Authorization": "Bearer your_token"})
    def get_data(self, endpoint):
        response = self.session.get(f"{self.base_url}/{endpoint}")
        response.raise_for_status()
        return response.json()

2)异步任务处理:

  • 使用 Celery 将耗时的 API 调用放入后台队列
  • 避免阻塞 Django 请求处理线程
  1. 监控与日志:

    • 记录 API 请求响应时间和错误
    • 集成 Sentry 等监控工具捕获异常

九、完整示例:天气 API 调用

# views.py
from django.http import JsonResponse
import requests
from django.views.decorators.cache import cache_page
@cache_page(60 * 15)  # 缓存15分钟
def weather_view(request, city):
    api_key = "your_api_key"  # 从环境变量获取
    url = f"https://api.openweathermap.org/data/2.5/weather"
    params = {
        "q": city,
        "appid": api_key,
        "units": "metric"
    }
    try:
        response = requests.get(url, params=params)
        response.raise_for_status()
        data = response.json()
        # 提取需要的数据
        weather_data = {
            "city": data["name"],
            "temperature": data["main"]["temp"],
            "description": data["weather"][0]["description"],
            "humidity": data["main"]["humidity"]
        }
        return JsonResponse(weather_data)
    except requests.exceptions.RequestException as e:
        return JsonResponse({"error": str(e)}, status=500)

总结

  • 简单请求:使用 requests 库
  • 异步需求:使用 aiohttp 或 requests-futures
  • 重试机制:结合 tenacity 库实现
  • 性能优化:使用连接池和缓存
  • 安全保障:验证输入、保护敏感信息、设置超时

合理选择请求方式并做好错误处理,可以在 Django 中高效、安全地调用外部 HTTP 服务。

到此这篇关于Django 调用外部http服务的实现的文章就介绍到这了,更多相关Django 调用外部http服务内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

相关文章

  • Python手机号码归属地查询代码

    Python手机号码归属地查询代码

    这篇文章主要介绍了Python手机号码归属地查询代码的相关资料,需要的朋友可以参考下
    2016-05-05
  • Python实现的检测网站挂马程序

    Python实现的检测网站挂马程序

    这篇文章主要介绍了Python实现的检测网站挂马程序,需要的朋友可以参考下
    2014-11-11
  • python图像加噪声的实现示例

    python图像加噪声的实现示例

    图像加噪声就是其中一种常见的处理方式,噪声可以帮助提高图像的真实性和复杂性,使得处理后的图像更加接近真实的场景,本文主要介绍了python图像加噪声的实现示例,感兴趣的可以了解一下
    2023-08-08
  • Python 画出来六维图

    Python 画出来六维图

    这篇文章主要介绍了Python 画出来六维图,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
    2019-07-07
  • python实现机器学习之多元线性回归

    python实现机器学习之多元线性回归

    这篇文章主要为大家详细介绍了python实现机器学习之多元线性回归,文中示例代码介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们可以参考一下
    2018-09-09
  • Python Django给admin添加Action的方法实例详解

    Python Django给admin添加Action的方法实例详解

    这篇文章主要介绍了Django给admin添加Action的方法,本文通过实例代码给大家介绍的非常详细,具有一定的参考借鉴价值,需要的朋友可以参考下
    2019-04-04
  • Pandas探索之高性能函数eval和query解析

    Pandas探索之高性能函数eval和query解析

    这篇文章主要介绍了Pandas探索之高性能函数eval和query解析,小编觉得还是挺不错的,这里分享给大家,供需要的朋友参考。
    2017-10-10
  • OpenCV:imwrite函数保存图片问题

    OpenCV:imwrite函数保存图片问题

    这篇文章主要介绍了关于OpenCV:imwrite函数保存图片问题,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教
    2022-11-11
  • python读取raw binary图片并提取统计信息的实例

    python读取raw binary图片并提取统计信息的实例

    今天小编就为大家分享一篇python读取raw binary图片并提取统计信息的实例,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2020-01-01
  • 详解Python文本操作相关模块

    详解Python文本操作相关模块

    这篇文章主要介绍了详解Python文本操作相关模块的相关资料,需要的朋友可以参考下
    2017-06-06

最新评论