现在学Python还有前景吗？后端开发与数据科学岗位需求分析

from fastapi import FastAPI, HTTPException, status, Depends
from pydantic import BaseModel, Field, EmailStr
from typing import Optional
import asyncio

app = FastAPI(title="UserTaskService", version="1.0.0")

# 1. 强类型请求模型定义
class CreateTaskRequest(BaseModel):
    title: str = Field(..., min_length=3, max_length=100, description="任务标题")
    contact_email: EmailStr = Field(..., description="通知邮箱")
    priority: int = Field(default=1, ge=1, le=5, description="优先级 1-5")

class TaskResponse(BaseModel):
    task_id: str
    status: str
    message: str

# 2. 模拟异步数据库持久化操作（非阻塞I/O）
async def save_task_to_storage(task_data: CreateTaskRequest) -> str:
    # 真实场景中此处调用 asyncpg / Tortoise-ORM 进行异步写入
    await asyncio.sleep(0.05)  # 模拟网络I/O延迟
    return f"tsk_{abs(hash(task_data.title)) % 1000000}"

# 3. 生产端点设计：依赖注入与结构化异常处理
@app.post("/api/v1/tasks", response_model=TaskResponse, status_code=status.HTTP_201_CREATED)
async def create_task_endpoint(payload: CreateTaskRequest):
    try:
        task_id = await save_task_to_storage(payload)
        return TaskResponse(
            task_id=task_id,
            status="pending",
            message="Task successfully queued for execution"
        )
    except Exception as exc:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Database persistence failure: {str(exc)}"
        )

# 4. 健康检查端点（供 Kubernetes 探针探测）
@app.get("/healthz", status_code=status.HTTP_200_OK)
async def health_check():
    return {"status": "healthy"}

# 启动生产就绪的 ASGI 服务（配置 4 个异步 Worker）
uvicorn backend_service:app --host 0.0.0.0 --port 8000 --workers 4

# 使用 curl 发起符合校验规则的 POST 请求
curl -X POST "http://127.0.0.1:8000/api/v1/tasks"      -H "Content-Type: application/json"      -d '{"title": "Sync Customer Profile", "contact_email": "ops@example.com", "priority": 3}'

import numpy as np
import pandas as pd
from typing import List, Dict

# 1. 模拟企业原始业务日志数据
raw_records = [
    {"doc_id": 101, "content": "FastAPI provides high performance async REST endpoints.", "category": "backend"},
    {"doc_id": 102, "content": "Pandas enables efficient tabular data manipulation and cleaning.", "category": "data"},
    {"doc_id": 103, "content": "PyTorch supports dynamic tensor computation and deep learning.", "category": "ai"},
    {"doc_id": 104, "content": "   ", "category": "invalid"},  # 脏数据需剔除
]

# 2. 数据清洗与格式标准化
df = pd.DataFrame(raw_records)
df["content"] = df["content"].str.strip()
clean_df = df[df["content"].str.len() > 0].reset_index(drop=True)

# 3. 简易特征向量化示例（基于预设维度的确定性语义映射，模拟 Embedding 输出）
def mock_embedding(text: str) -> np.ndarray:
    # 模拟 4 维语义空间：[后端, 数据, 智能AI, 系统网络]
    vec = np.zeros(4, dtype=np.float32)
    t = text.lower()
    if "api" in t or "rest" in t or "backend" in t: vec[0] += 0.8
    if "data" in t or "cleaning" in t or "tabular" in t: vec[1] += 0.9
    if "learning" in t or "ai" in t or "tensor" in t: vec[2] += 0.85
    if "network" in t or "async" in t: vec[3] += 0.7
    norm = np.linalg.norm(vec)
    return vec / norm if norm > 0 else vec

# 构建向量索引列
clean_df["vector"] = clean_df["content"].apply(mock_embedding)

# 4. 向量余弦相似度检索函数
def search_top_k(query: str, data: pd.DataFrame, top_k: int = 1) -> List[Dict]:
    q_vec = mock_embedding(query)
    scores = []
    for idx, row in data.iterrows():
        # 余弦相似度计算：dot(a, b) / (|a|*|b|)
        sim = float(np.dot(q_vec, row["vector"]))
        scores.append((sim, row["doc_id"], row["content"]))
    
    scores.sort(key=lambda x: x[0], reverse=True)
    return [{"score": round(s[0], 4), "doc_id": s[1], "content": s[2]} for s in scores[:top_k]]

# 5. 执行语义查询示例
if __name__ == "__main__":
    search_query = "How to train deep learning neural models?"
    results = search_top_k(search_query, clean_df, top_k=1)
    for res in results:
        print(f"匹配得分: {res['score']} | 文档ID: {res['doc_id']} | 匹配内容: {res['content']}")