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 的使用场景