Requests 传递参数

import requests

url = "https://httpbin.org/get"

# 1. 基本参数
params = {
    "search": "python requests",
    "page": 1,
    "sort": "desc"
}

# 2. 列表参数（自动转换为多个同名参数）
params["tags"] = ["python", "http", "tutorial"]

# 3. 嵌套参数（需要手动处理）
# 对于嵌套结构，可以序列化为JSON字符串
import json
params["filters"] = json.dumps({
    "category": "programming",
    "level": "beginner",
    "year": 2023
})

# 4. 特殊字符和中文（自动编码）
params["query"] = "Python请求库&教程"

response = requests.get(url, params=params)
print(f"实际请求URL:\n{response.url}")

if response.status_code == 200:
    data = response.json()
    print("\n服务器接收到的参数:")
    print(json.dumps(data['args'], indent=2, ensure_ascii=False))

import requests

url = "https://httpbin.org/post"

# 表单数据
form_data = {
    "username": "john_doe",
    "password": "secret123",
    "email": "john@example.com",
    "remember_me": True,
    "newsletter": False
}

# 发送POST请求
response = requests.post(url, data=form_data)

print(f"状态码: {response.status_code}")
print(f"Content-Type: {response.headers.get('content-type')}")

if response.status_code == 200:
    result = response.json()
    print("\n服务器接收到的表单数据:")
    print(result['form'])

import requests
from requests_toolbelt.multipart.encoder import MultipartEncoder

url = "https://httpbin.org/post"

# 1. 包含列表的表单数据
form_data = {
    "name": "张三",
    "skills": ["Python", "JavaScript", "SQL"],  # 列表会被正确处理
    "preferences": {
        "theme": "dark",
        "language": "zh-CN"
    }
}

# 2. 嵌套数据（需要手动序列化）
import json
form_data["preferences_json"] = json.dumps(form_data["preferences"])

# 3. 发送请求
response = requests.post(url, data=form_data)

if response.status_code == 200:
    result = response.json()
    print("服务器接收到的数据:")

    # 查看原始表单数据
    print("\n表单数据:")
    for key, value in result['form'].items():
        print(f"  {key}: {value}")

    # 查看请求头
    print("\n请求头Content-Type:")
    print(result['headers']['Content-Type'])

import requests
from datetime import datetime

url = "https://httpbin.org/post"

# 复杂JSON数据结构
json_data = {
    "user": {
        "id": 12345,
        "name": "张三",
        "email": "zhangsan@example.com",
        "profile": {
            "age": 25,
            "city": "北京",
            "interests": ["编程", "读书", "旅行"]
        }
    },
    "order": {
        "id": "ORD-2023-001",
        "items": [
            {"product_id": 1, "quantity": 2, "price": 99.99},
            {"product_id": 2, "quantity": 1, "price": 49.99}
        ],
        "total": 249.97,
        "timestamp": datetime.now().isoformat()
    },
    "metadata": {
        "api_version": "1.0",
        "request_id": "req_123456789",
        "tags": ["important", "urgent"]
    }
}

# 发送JSON数据
response = requests.post(url, json=json_data)

print(f"状态码: {response.status_code}")
print(f"Content-Type: {response.headers.get('content-type')}")

if response.status_code == 200:
    result = response.json()

    # 查看发送的JSON数据
    print("\n发送的JSON数据:")
    import json
    print(json.dumps(result['json'], indent=2, ensure_ascii=False))

    # 查看请求头
    print("\n请求头:")
    print(f"Content-Type: {result['headers']['Content-Type']}")
    print(f"Content-Length: {result['headers']['Content-Length']}")

import requests
import json
from datetime import datetime
from decimal import Decimal

url = "https://httpbin.org/post"

# 自定义JSON编码器
class CustomJSONEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, datetime):
            return obj.isoformat()
        if isinstance(obj, Decimal):
            return float(obj)
        if hasattr(obj, '__dict__'):
            return obj.__dict__
        return super().default(obj)

# 包含自定义对象的数据
data = {
    "timestamp": datetime.now(),
    "price": Decimal("99.99"),
    "user": type('User', (), {
        'name': '张三',
        'age': 25
    })()
}

# 方法1：使用自定义编码器
response = requests.post(
    url,
    data=json.dumps(data, cls=CustomJSONEncoder),
    headers={'Content-Type': 'application/json'}
)

# 方法2：先转换为可序列化的字典
def to_serializable(obj):
    if isinstance(obj, datetime):
        return obj.isoformat()
    if isinstance(obj, Decimal):
        return float(obj)
    return obj

serializable_data = {
    k: to_serializable(v) for k, v in data.items()
}
response = requests.post(url, json=serializable_data)

import requests

url = "https://httpbin.org/post"

# 准备要上传的文件
# 假设当前目录下有这些文件
file_paths = {
    "document": "report.pdf",
    "image": "photo.jpg",
    "data": "data.csv"
}

# 打开文件
files = {}
for key, file_path in file_paths.items():
    try:
        files[key] = open(file_path, 'rb')
    except FileNotFoundError:
        print(f"文件不存在: {file_path}")
        # 创建示例文件
        with open(file_path, 'w') as f:
            f.write(f"这是{file_path}的示例内容")
        files[key] = open(file_path, 'rb')

# 其他表单字段
data = {
    "description": "项目文档和图片",
    "category": "documents",
    "uploader": "张三"
}

# 发送文件上传请求
response = requests.post(url, files=files, data=data)

# 关闭文件
for file in files.values():
    file.close()

print(f"状态码: {response.status_code}")

if response.status_code == 200:
    result = response.json()
    print("\n上传的文件信息:")
    for file_key in file_paths.keys():
        if file_key in result['files']:
            print(f"{file_key}: {len(result['files'][file_key])} 字节")

    print("\n表单字段:")
    print(result['form'])

    print("\n请求头Content-Type:")
    print(result['headers']['Content-Type'][:50] + "...")

import requests
import io

url = "https://httpbin.org/post"

# 1. 上传内存中的文件（不保存到磁盘）
text_file = io.BytesIO(b"这是内存中的文本文件内容")
image_data = b"fake image binary data"  # 模拟图片数据
image_file = io.BytesIO(image_data)

files = {
    'text_file': ('memory.txt', text_file, 'text/plain'),
    'image_file': ('memory.jpg', image_file, 'image/jpeg'),
    'custom_file': ('data.bin', io.BytesIO(b"custom binary data"), 'application/octet-stream')
}

# 2. 设置自定义文件名和MIME类型
files['document'] = ('年度报告.pdf', open('report.pdf', 'rb'), 'application/pdf')

# 3. 使用元组指定文件信息
# 格式: (filename, fileobj, content_type, headers)
files['with_headers'] = (
    'data.json',
    io.BytesIO(b'{"key": "value"}'),
    'application/json',
    {'X-Custom-Header': 'custom-value'}
)

# 4. 发送请求
response = requests.post(url, files=files)

print(f"状态码: {response.status_code}")

if response.status_code == 200:
    result = response.json()
    print("\n上传成功!")
    print(f"文件数量: {len(result['files'])}")

    # 查看请求头中的boundary
    content_type = result['headers']['Content-Type']
    print(f"Content-Type: {content_type[:80]}...")

import requests
import os
from tqdm import tqdm  # 需要安装: pip install tqdm

class FileWithProgress:
    """包装文件对象，添加进度显示"""
    def __init__(self, file_path):
        self.file_path = file_path
        self.file_size = os.path.getsize(file_path)
        self.file_obj = open(file_path, 'rb')
        self.progress_bar = tqdm(
            total=self.file_size,
            unit='B',
            unit_scale=True,
            desc=os.path.basename(file_path)
        )

    def read(self, size=-1):
        data = self.file_obj.read(size)
        if data:
            self.progress_bar.update(len(data))
        return data

    def close(self):
        self.file_obj.close()
        self.progress_bar.close()

# 大文件上传示例
def upload_large_file(file_path, url):
    """上传大文件并显示进度"""
    file_wrapper = FileWithProgress(file_path)

    files = {
        'file': (os.path.basename(file_path), file_wrapper, 'application/octet-stream')
    }

    data = {
        'description': '大文件上传测试',
        'chunked': True
    }

    try:
        response = requests.post(url, files=files, data=data)
        return response
    finally:
        file_wrapper.close()

# 使用示例
if __name__ == "__main__":
    # 创建一个测试大文件
    test_file = "large_test_file.bin"
    with open(test_file, 'wb') as f:
        f.write(os.urandom(10 * 1024 * 1024))  # 10MB随机数据

    print(f"开始上传 {test_file}...")
    response = upload_large_file(test_file, "https://httpbin.org/post")

    if response.status_code == 200:
        print("上传成功!")
        result = response.json()
        print(f"接收到的文件大小: {len(result['files']['file'])} 字节")

    # 清理测试文件
    os.remove(test_file)

import requests
import uuid

url = "https://httpbin.org/headers"

# 完整的请求头配置
headers = {
    # 认证信息
    'Authorization': 'Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...',

    # 内容协商
    'Accept': 'application/json',
    'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
    'Accept-Encoding': 'gzip, deflate',

    # 客户端信息
    'User-Agent': 'MyPythonApp/1.0.0 (Windows NT 10.0; Win64; x64)',
    'Referer': 'https://example.com/dashboard',

    # 自定义头部
    'X-API-Key': '12345-abcde-67890-fghij',
    'X-Request-ID': str(uuid.uuid4()),
    'X-Correlation-ID': str(uuid.uuid4()),
    'X-Client-Version': '1.2.3',

    # 缓存控制
    'Cache-Control': 'no-cache',
    'Pragma': 'no-cache',

    # 连接管理
    'Connection': 'keep-alive',
    'Keep-Alive': 'timeout=5, max=100'
}

# 发送请求
response = requests.get(url, headers=headers)

if response.status_code == 200:
    result = response.json()
    print("服务器接收到的请求头:")

    # 过滤显示我们设置的头部
    our_headers = {k: v for k, v in result['headers'].items()
                   if any(h.lower() in k.lower() for h in headers.keys())}

    for key, value in our_headers.items():
        print(f"{key}: {value}")

import requests

url = "https://httpbin.org/cookies"

# 1. 发送Cookie
cookies = {
    'session_id': 'abc123def456',
    'user_id': '12345',
    'preferences': 'theme=dark&lang=zh'
}

response = requests.get(url, cookies=cookies)
print("发送Cookie的响应:")
print(response.json())

# 2. 接收和保存Cookie
response = requests.get('https://httpbin.org/cookies/set/session/xyz789')
print("\n服务器设置的Cookie:")
print(response.cookies)

# 3. 使用Session自动处理Cookie
session = requests.Session()

# 首次请求，服务器设置cookie
response1 = session.get('https://httpbin.org/cookies/set/user/john_doe')
print(f"\n第一次请求，服务器设置cookie: {response1.cookies}")

# 后续请求自动携带cookie
response2 = session.get('https://httpbin.org/cookies')
print("第二次请求，自动携带的Cookie:")
print(response2.json())

import requests
from http.cookies import SimpleCookie

url = "https://httpbin.org/cookies"

# 1. 从字符串解析Cookie
cookie_string = "session_id=abc123; user_id=456; theme=dark"
cookie = SimpleCookie()
cookie.load(cookie_string)

# 转换为字典
cookies_dict = {k: v.value for k, v in cookie.items()}
print(f"解析的Cookie: {cookies_dict}")

# 2. 设置Cookie属性（如domain, path, expires）
cookies = {
    'session': {
        'value': 'xyz789',
        'domain': 'httpbin.org',
        'path': '/',
        'expires': 3600  # 1小时后过期
    }
}

# RequestsCookieJar对象可以存储带属性的cookie
from requests.cookies import RequestsCookieJar

jar = RequestsCookieJar()
jar.set('session_id', 'abc123', domain='httpbin.org', path='/cookies')
jar.set('user_pref', 'dark_mode', domain='httpbin.org', path='/')

response = requests.get(url, cookies=jar)
print(f"\n使用CookieJar的响应: {response.json()}")

# 3. Cookie的持久化保存
import pickle

# 保存cookies
def save_cookies(session, filename):
    with open(filename, 'wb') as f:
        pickle.dump(session.cookies, f)

# 加载cookies
def load_cookies(filename):
    with open(filename, 'rb') as f:
        return pickle.load(f)

# 使用示例
session = requests.Session()
response = session.get('https://httpbin.org/cookies/set/persistent/cookie_value')

# 保存
save_cookies(session, 'cookies.pkl')

# 新会话中加载
new_session = requests.Session()
new_session.cookies = load_cookies('cookies.pkl')

response = new_session.get('https://httpbin.org/cookies')
print(f"\n加载持久化Cookie后的响应: {response.json()}")

import requests

url = "https://httpbin.org/post"

# 混合使用多种参数
params = {'action': 'upload'}  # URL参数
data = {'description': '文件描述'}  # 表单数据
json_data = {'metadata': {'source': 'python', 'version': '1.0'}}  # JSON数据
files = {'file': open('example.txt', 'rb')}  # 文件
headers = {'X-API-Key': 'secret123'}  # 请求头
cookies = {'session': 'abc123'}  # Cookie

# 注意：不能同时使用data和json参数
# 正确做法：如果需要混合，手动处理JSON部分
import json

# 将JSON数据合并到表单数据中
data['metadata_json'] = json.dumps(json_data['metadata'])

response = requests.post(
    url,
    params=params,
    data=data,
    files=files,
    headers=headers,
    cookies=cookies
)

print(f"状态码: {response.status_code}")
if response.status_code == 200:
    result = response.json()
    print("\n服务器接收到的数据:")
    print(f"URL参数: {result['args']}")
    print(f"表单数据: {result['form']}")
    print(f"文件: {list(result['files'].keys())}")
    print(f"请求头: {{k: v for k, v in result['headers'].items() if 'api' in k.lower()}}")
    print(f"Cookie: {result['headers'].get('Cookie', '无')}")

import requests
import re
from typing import Dict, Any

def validate_and_clean_params(params: Dict[str, Any]) -> Dict[str, Any]:
    """验证和清理参数"""
    cleaned = {}

    for key, value in params.items():
        # 跳过None值
        if value is None:
            continue

        # 清理字符串（去除首尾空格）
        if isinstance(value, str):
            value = value.strip()
            if not value:  # 跳过空字符串
                continue

        # 验证键名（只允许字母、数字、下划线）
        if not re.match(r'^[a-zA-Z0-9_]+$', key):
            raise ValueError(f"无效的参数名: {key}")

        # 限制值长度
        if isinstance(value, str) and len(value) > 1000:
            value = value[:1000]

        cleaned[key] = value

    return cleaned

def send_request_with_validation(url: str, **kwargs):
    """发送请求前验证参数"""

    # 验证和清理参数
    if 'params' in kwargs and kwargs['params']:
        kwargs['params'] = validate_and_clean_params(kwargs['params'])

    if 'data' in kwargs and kwargs['data']:
        kwargs['data'] = validate_and_clean_params(kwargs['data'])

    # 发送请求
    response = requests.post(url, **kwargs)
    return response

# 使用示例
url = "https://httpbin.org/post"

try:
    params = {
        "valid_key": "正常值",
        "key with spaces": "值",  # 无效键名
        "empty": "",  # 空字符串
        "none_value": None,  # None值
        "long_value": "a" * 2000  # 超长值
    }

    response = send_request_with_validation(
        url,
        params=params,
        json={"test": "data"},
        headers={"User-Agent": "TestClient"}
    )

    print(f"状态码: {response.status_code}")

except ValueError as e:
    print(f"参数验证失败: {e}")
except Exception as e:
    print(f"请求失败: {e}")

import requests
from typing import Optional, Dict, Any

class RequestBuilder:
    """请求构建器，支持链式调用"""

    def __init__(self, method: str = 'GET', url: str = ''):
        self.method = method.upper()
        self.url = url
        self.params = {}
        self.data = {}
        self.json_data = None
        self.files = {}
        self.headers = {}
        self.cookies = {}
        self.timeout = None

    def set_url(self, url: str) -> 'RequestBuilder':
        self.url = url
        return self

    def add_param(self, key: str, value: Any) -> 'RequestBuilder':
        self.params[key] = value
        return self

    def add_header(self, key: str, value: str) -> 'RequestBuilder':
        self.headers[key] = value
        return self

    def set_json(self, data: Dict) -> 'RequestBuilder':
        self.json_data = data
        return self

    def set_timeout(self, seconds: float) -> 'RequestBuilder':
        self.timeout = seconds
        return self

    def build(self):
        """构建请求参数"""
        kwargs = {}

        if self.params:
            kwargs['params'] = self.params

        if self.json_data:
            kwargs['json'] = self.json_data
        elif self.data:
            kwargs['data'] = self.data

        if self.files:
            kwargs['files'] = self.files

        if self.headers:
            kwargs['headers'] = self.headers

        if self.cookies:
            kwargs['cookies'] = self.cookies

        if self.timeout:
            kwargs['timeout'] = self.timeout

        return kwargs

    def execute(self):
        """执行请求"""
        kwargs = self.build()

        method_func = getattr(requests, self.method.lower())
        return method_func(self.url, **kwargs)

# 使用示例
builder = (RequestBuilder('POST', 'https://httpbin.org/post')
    .add_param('action', 'create')
    .add_header('User-Agent', 'RequestBuilder/1.0')
    .add_header('X-Request-ID', 'req_123')
    .set_json({
        'user': {'name': '张三', 'age': 25},
        'action': 'register'
    })
    .set_timeout(10.0))

response = builder.execute()

print(f"状态码: {response.status_code}")
if response.status_code == 200:
    result = response.json()
    print(f"URL: {result['url']}")
    print(f"JSON数据: {result['json']}")
    print(f"请求头User-Agent: {result['headers'].get('User-Agent')}")