Python 全栈技术生态深度实践：从 Web 后端到 AI 工程的完整工具链解析

from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker, declarative_base
from pydantic import BaseModel
import uvicorn

# 数据库配置
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)

Base = declarative_base()
app = FastAPI()

# 定义模型和 Pydantic 模式
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50), nullable=False)
    email = Column(String(100), unique=True)

class UserCreate(BaseModel):
    name: str
    email: str

# 依赖项：获取数据库会话
async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

@app.post("/users")
async def create_user(user: UserCreate, db: AsyncSession = Depends(get_db)):
    new_user = User(name=user.name, email=user.email)
    db.add(new_user)
    await db.commit()
    await db.refresh(new_user)
    return new_user

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

import polars as pl
import pandas as pd

# 使用 Polars 高效读取大型 CSV
df = pl.read_csv("sales_2025.csv", try_parse_dates=True)

# 数据预处理：过滤、分组、聚合
result = (df
    .filter(pl.col("amount") > 0)
    .group_by("category")
    .agg([
        pl.sum("amount").alias("total_sales"),
        pl.mean("quantity").alias("avg_qty")
    ])
    .sort("total_sales", descending=True)
)

# 转换为 Pandas 进行复杂统计分析（如使用 scipy）
pandas_df = result.to_pandas()
print(pandas_df.describe())

from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.model_selection import train_test_split
from xgboost import XGBClassifier
from sklearn.metrics import classification_report

# 数据准备（假设使用上述 Polars 结果）
X = pandas_df[["total_sales", "avg_qty", "category"]]  # 简化示例
y = pandas_df["target"]

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)

# 预处理管道
preprocessor = ColumnTransformer([
    ("num", StandardScaler(), ["total_sales", "avg_qty"]),
    ("cat", OneHotEncoder(), ["category"])
])

pipeline = Pipeline([
    ("pre", preprocessor),
    ("clf", XGBClassifier(n_estimators=100, learning_rate=0.1, use_label_encoder=False))
])

pipeline.fit(X_train, y_train)
y_pred = pipeline.predict(X_test)
print(classification_report(y_test, y_pred))

import click
from loguru import logger
import polars as pl

@click.command()
@click.option("--input", required=True, help="输入 CSV 路径")
@click.option("--output", required=True, help="输出 Parquet 路径")
@click.option("--threshold", default=0, help="销售额阈值")
def process_data(input, output, threshold):
    """清洗并转换销售数据"""
    logger.info(f"正在读取 {input}")
    df = pl.read_csv(input)
    df_clean = df.filter(pl.col("amount") > threshold)
    df_clean.write_parquet(output)
    logger.success(f"已保存至 {output}, 共 {df_clean.height} 行")

if __name__ == "__main__":
    process_data()

from celery import Celery

celery_app = Celery(
    "tasks",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/0"
)

@celery_app.task
def train_model_task(data_url):
    # 执行训练逻辑
    return {"status": "completed", "accuracy": 0.92}

@app.post("/train")
async def start_training(data_url: str):
    task = train_model_task.delay(data_url)
    return {"task_id": task.id}