Python数据处理与API实战指南

# 这就是一个"JSON对象"的Python形态
user = {
    "name": "蓝胖",
    "age": 22,
    "skills": ["Python", "Java", "SQL"],
    "address": {
        "city": "北京",
        "zip": "100000"
    }
}

# 访问方式和JSON的键值对一模一样
print(user["name"])              # "蓝胖"
print(user["skills"][0])         # "Python"
print(user["address"]["city"])   # "北京"

// Java对比：需要定义实体类或用Map
// 方式1：实体类（繁琐）
public class User {
    private String name;
    private int age;
    private List<String> skills;
    // getter/setter...
}

// 方式2：Map（不够直观）
Map<String, Object> user = new HashMap<>();
user.put("name", "蓝胖");
user.put("age", 22);

import json

user = {"name": "蓝胖", "age": 22}

# 字典 → JSON字符串（序列化）
json_str = json.dumps(user)
print(json_str)  # {"name": "蓝胖", "age": 22}

# 格式化输出（缩进2空格，面试常用！）
json_pretty = json.dumps(user, indent=2, ensure_ascii=False)
print(json_pretty)
# {
#   "name": "蓝胖",
#   "age": 22
# }

// Java对比（Jackson）
ObjectMapper mapper = new ObjectMapper();
String json = mapper.writeValueAsString(user);
// 或 Gson
String json = new Gson().toJson(user);

import json

json_str = '{"name": "蓝胖", "age": 22, "active": true}'

# JSON字符串 → 字典（反序列化）
user = json.loads(json_str)
print(user["name"])     # "蓝胖"
print(user["active"])   # True（注意：JSON的true变成Python的True）

# 安全解析（键不存在时不会报错）
city = user.get("city", "未知")  # "未知"

import json

data = {"name": "蓝胖", "age": 22}

# 写入JSON文件
with open("user.json", "w", encoding="utf-8") as f:
    json.dump(data, f, ensure_ascii=False, indent=2)

# 读取JSON文件
with open("user.json", "r", encoding="utf-8") as f:
    loaded = json.load(f)
    print(loaded["name"])  # "蓝胖"

// Java对比（Jackson读文件）
ObjectMapper mapper = new ObjectMapper();
User user = mapper.readValue(new File("user.json"), User.class);

# 坑1：dumps和loads搞混
json.dumps({"key": "value"})  # dict → str（序列化）
json.loads('{"key": "value"}')  # str → dict（反序列化）
# 记忆技巧：s = string，loads = load string

# 坑2：单引号不是合法JSON
json.loads("{'key': 'value'}")  # ❌ 报错！JSON必须用双引号
json.loads('{"key": "value"}')  # ✅ 正确

# 坑3：中文乱码
json.dumps({"name": "蓝胖"})           # {"name": "\u84dd\u80d6"} ← 乱码
json.dumps({"name": "蓝胖"}, ensure_ascii=False)  # {"name": "蓝胖"} ← 正确

# 先安装
pip install requests

// Java对比：HttpClient 需要多行代码
HttpClient client = HttpClient.newHttpClient();
HttpRequest request = HttpRequest.newBuilder()
    .uri(URI.create("https://api.example.com/data"))
    .header("Content-Type", "application/json")
    .GET()
    .build();
HttpResponse<String> response = client.send(request, 
    HttpResponse.BodyHandlers.ofString());
String body = response.body();

import requests

# 最简单的GET请求
response = requests.get("https://api.example.com/users")

# 带查询参数
params = {"page": 1, "size": 10, "keyword": "Python"}
response = requests.get("https://api.example.com/users", params=params)
# 实际请求URL：https://api.example.com/users?page=1&size=10&keyword=Python

# 带请求头
headers = {
    "Authorization": "Bearer token123",
    "User-Agent": "MyApp/1.0"
}
response = requests.get("https://api.example.com/users", headers=headers)

// Java对比（OkHttp）
OkHttpClient client = new OkHttpClient();
HttpUrl url = HttpUrl.parse("https://api.example.com/users").newBuilder()
    .addQueryParameter("page", "1")
    .addQueryParameter("size", "10")
    .build();
Request request = new Request.Builder()
    .url(url)
    .addHeader("Authorization", "Bearer token123")
    .build();
Response response = client.newCall(request).execute();

import requests

# 发送JSON数据（最常用！）
payload = {
    "name": "蓝胖",
    "age": 22,
    "skills": ["Python", "Java"]
}
headers = {"Content-Type": "application/json"}

response = requests.post("https://api.example.com/users", json=payload, headers=headers)
# 注意：json= 参数会自动设置 Content-Type 并序列化字典为JSON
# 所以上面的headers可以省略！

# 简化写法
response = requests.post("https://api.example.com/users", json=payload)

# 发送表单数据（类比 form-urlencoded）
form_data = {"username": "lanpang", "password": "123456"}
response = requests.post("https://api.example.com/login", data=form_data)

// Java对比（OkHttp POST JSON）
MediaType JSON = MediaType.parse("application/json; charset=utf-8");
RequestBody body = RequestBody.create(JSON, new Gson().toJson(payload));
Request request = new Request.Builder()
    .url("https://api.example.com/users")
    .post(body)
    .build();
Response response = client.newCall(request).execute();

response = requests.get("https://api.example.com/users")

# 状态码（类比 response.statusLine().statusCode()）
print(response.status_code)      # 200, 404, 500...

# 响应体
print(response.text)             # 原始字符串
data = response.json()           # 自动解析JSON为字典！（超级方便）
print(data["name"])              # 直接用

# 响应头
print(response.headers)                    # 所有响应头
print(response.headers["Content-Type"])    # 特定响应头

# 请求的URL（看最终请求地址）
print(response.url)

// Java对比
int code = response.code();
String body = response.body().string();
// Java需要手动解析JSON：
JsonObject json = JsonParser.parseString(body).getAsJsonObject();

import requests

try:
    response = requests.get("https://api.example.com/data", timeout=5)
    
    # 检查HTTP状态码，非2xx会抛出异常
    response.raise_for_status()
    
    # 处理数据
    data = response.json()
    
except requests.exceptions.Timeout:
    print("请求超时")
except requests.exceptions.ConnectionError:
    print("连接失败")
except requests.exceptions.HTTPError as e:
    print(f"HTTP错误：{e}")
except requests.exceptions.RequestException as e:
    print(f"请求异常：{e}")

# 需要多次请求同一个站点时，使用Session复用连接
session = requests.Session()
session.headers.update({"Authorization": "Bearer token123"})

# 后续请求自动带上headers
resp1 = session.get("https://api.example.com/users")
resp2 = session.get("https://api.example.com/orders")

# 类比Java OkHttp的OkHttpClient复用

1. 准备参数 → 2. 发送请求 → 3. 解析响应 → 4. 提取内容

import requests
import os
from dotenv import load_dotenv

load_dotenv()

def get_weather(city: str) -> dict:
    """查询天气信息"""
    # 1. 准备参数
    url = "https://api.weather.com/v1/current"
    params = {"city": city, "key": os.getenv("WEATHER_API_KEY")}
    headers = {"Accept": "application/json"}
    
    # 2. 发送请求
    response = requests.get(url, params=params, headers=headers, timeout=10)
    
    # 3. 解析响应
    response.raise_for_status()
    data = response.json()
    
    # 4. 提取内容
    return {
        "city": city,
        "temperature": data["current"]["temp"],
        "humidity": data["current"]["humidity"],
        "description": data["current"]["weather_desc"]
    }

# 使用
weather = get_weather("北京")
print(f"{weather['city']}：{weather['temperature']}°C，{weather['description']}")

messages = [
    {"role": "system", "content": "你是一个专业的Python老师，擅长用Java对比来讲解Python"},
    {"role": "user", "content": "Python的列表和Java的ArrayList有什么区别？"},
    {"role": "assistant", "content": "Python列表是动态类型的，可以混合存放不同类型的元素..."},
    {"role": "user", "content": "能给个代码例子吗？"}
]

import requests
import os
from dotenv import load_dotenv

load_dotenv()

def call_qwen(prompt: str, system_prompt: str = "你是一个有帮助的AI助手。") -> str:
    """调用通义千问大模型"""
    # 1. 准备参数
    url = "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions"
    
    headers = {
        "Authorization": f"Bearer {os.getenv('DASHSCOPE_API_KEY')}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "model": "qwen-turbo",           # 模型选择
        "messages": [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": prompt}
        ],
        "temperature": 0.7               # 创造性参数
    }
    
    # 2. 发送请求
    response = requests.post(url, json=payload, headers=headers, timeout=30)
    
    # 3. 解析响应
    response.raise_for_status()
    result = response.json()
    
    # 4. 剥洋葱取内容
    # 结构：result["choices"][0]["message"]["content"]
    content = result["choices"][0]["message"]["content"]
    return content

# 使用
answer = call_qwen("用Python写一个快速排序")
print(answer)

# temperature = 0：确定性输出，每次结果几乎一样
# 适合：代码生成、数据分析、事实查询
payload = {"model": "qwen-turbo", "messages": messages, "temperature": 0}

# temperature = 0.7：平衡模式（默认推荐）
# 适合：一般对话、内容创作

# temperature = 1.0+：创造性输出，结果更多样
# 适合：头脑风暴、创意写作、广告文案
payload = {"model": "qwen-turbo", "messages": messages, "temperature": 1.0}

// Java对比：没有直接对应的概念
// 类比：生成验证码时的随机种子
// temperature=0 像是用固定种子的Random，每次结果一致
// temperature=1 像是完全随机的Random

# 响应结构
{
    "id": "chatcmpl-xxx",
    "model": "qwen-turbo",
    "choices": [                          # 可能有多个候选回复
        {
            "index": 0,
            "message": {                  # 消息对象
                "role": "assistant",
                "content": "这是AI的回复"   # ← 我们最终要拿的内容
            },
            "finish_reason": "stop"
        }
    ],
    "usage": {                            # token用量
        "prompt_tokens": 50,
        "completion_tokens": 100,
        "total_tokens": 150
    }
}
# 逐层提取（像剥洋葱一样）
content = response["choices"][0]["message"]["content"]
# 安全写法（防止键不存在）
content = response.get("choices", [{}])[0].get("message", {}).get("content", "")

// Java对比（用Jackson解析同样的结构）
JsonNode root = mapper.readTree(responseBody);
String content = root.path("choices").path(0).path("message").path("content").asText();
// Java的path()方法也是类似的逐层访问，但更啰嗦

def chat_with_qwen():
    """支持多轮对话的大模型交互"""
    messages = [
        {"role": "system", "content": "你是Python老师，用Java对比教学"}
    ]
    
    while True:
        user_input = input("你：")
        if user_input.lower() in ["quit", "exit", "q"]:
            break
        
        # 把用户输入加入对话历史
        messages.append({"role": "user", "content": user_input})
        
        # 调用API
        response = requests.post(
            "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions",
            headers={"Authorization": f"Bearer {os.getenv('DASHSCOPE_API_KEY')}"},
            json={"model": "qwen-turbo", "messages": messages, "temperature": 0.7},
            timeout=30
        )
        response.raise_for_status()
        
        # 提取回复
        reply = response.json()["choices"][0]["message"]["content"]
        
        # 把AI回复也加入历史（实现上下文记忆）
        messages.append({"role": "assistant", "content": reply})
        
        print(f"AI：{reply}\n")

# chat_with_qwen()

pip install fastapi uvicorn
# uvicorn是ASGI服务器，类比Tomcat

# main.py
from fastapi import FastAPI

app = FastAPI()

# 路由装饰器（类比 @GetMapping）
@app.get("/")
def root():
    return {"message": "Hello World"}  # return字典自动转JSON！

@app.get("/users/{user_id}")
def get_user(user_id: int):  # 路径参数自动类型转换
    return {"user_id": user_id, "name": "蓝胖"}

@app.get("/users")
def list_users(page: int = 1, size: int = 10):  # 查询参数
    return {"page": page, "size": size, "users": []}

// Java对比（Spring Boot）
@RestController
public class UserController {
    @GetMapping("/")
    public Map<String, String> root() {
        return Map.of("message", "Hello World");
    }
    @GetMapping("/users/{user_id}")
    public Map<String, Object> getUser(@PathVariable int userId) {
        return Map.of("user_id", userId, "name", "蓝胖");
    }
    @GetMapping("/users")
    public Map<String, Object> listUsers(
            @RequestParam(defaultValue = "1") int page,
            @RequestParam(defaultValue = "10") int size) {
        return Map.of("page", page, "size", size, "users", List.of());
    }
}

from fastapi import FastAPI
from pydantic import BaseModel  # 数据验证（类比Java的@Valid + DTO）

app = FastAPI()

# 定义数据模型（类比Java的DTO/VO）
class UserCreate(BaseModel):
    name: str
    age: int
    skills: list[str] = []  # 可选字段，默认空列表

# POST路由（类比 @PostMapping）
@app.post("/users")
def create_user(user: UserCreate):  # 自动解析JSON请求体
    # user已经是UserCreate对象了！
    return {
        "id": 1,
        "name": user.name,
        "age": user.age,
        "skills": user.skills
    }

// Java对比
@PostMapping("/users")
public UserVO createUser(@RequestBody @Valid UserCreateDTO dto) {
    UserVO vo = new UserVO();
    vo.setId(1);
    vo.setName(dto.getName());
    vo.setAge(dto.getAge());
    vo.setSkills(dto.getSkills());
    return vo;
}

# 路径参数：在URL路径中
@app.get("/users/{user_id}")
def get_user(user_id: int):  # /users/123
    return {"user_id": user_id}

# 查询参数：不在路径中的参数
@app.get("/users")
def list_users(page: int = 1, size: int = 10):  # /users?page=1&size=10
    return {"page": page, "size": size}

# 混合使用
@app.get("/users/{user_id}/orders")
def get_user_orders(user_id: int, status: str = "all"):  
    # /users/123/orders?status=pending
    return {"user_id": user_id, "status": status}

// Java对比
@GetMapping("/users/{userId}")
public User getUser(@PathVariable Long userId, 
                    @RequestParam(defaultValue = "all") String status) {
    // ...
}

# 启动开发服务器（类比 java -jar 或 mvn spring-boot:run）
uvicorn main:app --reload --port 8000

# 参数说明：
# main:app → main.py文件中的app对象
# --reload → 代码修改自动重载（开发模式）
# --port 8000 → 端口号

# 访问：
# http://localhost:8000          → 根路由
# http://localhost:8000/docs     → 自动生成的API文档（Swagger UI）
# http://localhost:8000/redoc    → 另一种API文档

# json= → 发送JSON（Content-Type: application/json）
requests.post(url, json={"key": "value"})

# data= → 发送表单（Content-Type: application/x-www-form-urlencoded）
requests.post(url, data={"key": "value"})

# 面试常问：调大模型API用哪个？→ json=

response = requests.get("https://api.example.com/data")
data = response.json()  # 这已经是dict了，不是字符串！

# ❌ 常见错误
data = response.json()
data_str = json.dumps(data)  # 多此一举！
data_again = json.loads(data_str)  # 更没必要！

# ✅ 直接用
name = data["name"]

# 同步写法（简单场景足够）
@app.get("/data")
def get_data():
    return {"result": "ok"}

# 异步写法（需要await其他异步操作时）
@app.get("/data")
async def get_data():
    result = await some_async_function()
    return {"result": result}

# 注意：如果函数内部没有await，不要用async def
# 否则可能阻塞事件循环

Python数据处理与API实战
├── JSON处理
│   ├── json.dumps()：dict → JSON字符串（序列化）
│   ├── json.loads()：JSON字符串 → dict（反序列化）
│   ├── json.dump() / json.load()：文件读写
│   ├── 字典就是JSON的Python形态
│   └── 对比Java的Jackson/Gson
├── requests库
│   ├── GET：requests.get(url, params={}, headers={})
│   ├── POST：requests.post(url, json={}, headers={})
│   ├── 响应：status_code / text / json() / headers
│   ├── 异常：raise_for_status() + try/except
│   ├── Session：复用连接
│   └── 对比Java的HttpClient/OkHttp
├── API调用实战
│   ├── 四步法：准备参数→发送请求→解析响应→提取内容
│   ├── 天气API示例
│   ├── Qwen大模型调用
│   │   ├── 三种角色：system / user / assistant
│   │   ├── payload结构：model / messages / temperature
│   │   ├── temperature：0=确定性，1=创造性
│   │   └── 剥洋葱：response["choices"][0]["message"]["content"]
│   └── 多轮对话：维护messages列表
├── FastAPI
│   ├── @app.get / @app.post（类比@GetMapping / @PostMapping）
│   ├── return字典自动转JSON
│   ├── 路径参数：{param} + 函数参数
│   ├── 查询参数：带默认值的函数参数
│   ├── Pydantic BaseModel：请求体自动解析验证
│   └── 自动API文档：/docs（Swagger UI）
└── 常见坑点
    ├── json= vs data=
    ├── response.json()已经是dict
    └── async def 的使用场景