吴极实战：从入门到精通搞定全栈项目

刚学会写 if-else 和 for 循环，却面对空白的 main.py 发呆？别慌，这是绝大多数转行编程新人的通病。我们常陷入“语法孤岛”，记住了 API 长什么样，却不知道如何把它们拼成能跑的砖块。

真正的【吴极】式开发，不是背诵文档，而是构建系统。今天带你走一遍从【入门到精通】的路径，用 Python 搭建一个高可用的短链接生成器。这不是玩具，而是具备缓存、并发处理和日志审计的生产级雏形。

项目目标与核心逻辑

很多新手写项目，上来就堆代码，结果改一处崩全局。我们要先定规矩。这个短链接服务旨在将冗长的 URL 压缩为短码，支持点击统计，并具备防刷能力。

核心痛点在于状态管理。如果每次点击都直接查数据库，高并发下数据库必挂。我们的方案是：

生成短码：使用 Base62 编码，保证唯一性。
缓存层：利用 Redis 存储短码与长链接的映射，命中缓存直接返回，未命中再查库。
异步落盘：点击计数通过消息队列异步写入数据库，解耦读与写。

这套架构是后端开发的基石。理解它，你就跨过了“写脚本”到“做工程”的门槛。

目录结构设计

工程化思维的第一步，是目录结构。混乱的文件是项目腐烂的开始。我们采用标准分层架构：

project_wuji/
├── app/
│   ├── __init__.py
│   ├── main.py          # FastAPI 入口
│   ├── api/
│   │   ├── __init__.py
│   │   └── v1/
│   │       ├── __init__.py
│   │       └── routes.py # 路由定义
│   ├── core/
│   │   ├── __init__.py
│   │   ├── config.py     # 配置管理
│   │   └── security.py   # 签名验证
│   ├── services/
│   │   ├── __init__.py
│   │   └── shortener.py  # 核心业务逻辑
│   └── db/
│       ├── __init__.py
│       ├── models.py     # SQLAlchemy 模型
│       └── session.py    # 数据库会话
├── tests/
│   ├── __init__.py
│   └── test_api.py
├── requirements.txt
├── .env.example
└── README.md

关键点解析：

app/core：存放不依赖具体业务的通用逻辑，如配置加载。
app/services：纯业务逻辑层，不直接操作 HTTP 请求，方便单元测试。
app/db：数据访问层，隔离 ORM 细节。

这种分离让你修改数据库引擎时，只需动 db 层，业务代码纹丝不动。

核心代码实现

1. 短码生成算法

不要直接用 uuid，它太长了。我们要生成 6-8 位的短码。参考 MDN Web Docs 中关于 Base64 的编码原理，我们定制 Base62 字符集。

import random
import stringCHAR_SET = string.ascii_letters + string.digits  # 62个字符def generate_short_code(length: int = 6) -> str:"""生成指定长度的随机短码:param length: 短码长度:return: 短码字符串"""# 使用 random.choices 保证字符随机且可重复code = ''.join(random.choices(CHAR_SET, k=length))return code

避坑指南：

碰撞处理：随机生成必然存在碰撞概率。必须在数据库中设置唯一索引，并在生成时加入重试机制。
可读性：避免使用易混淆字符（如 0/O, 1/I），在 CHAR_SET 中剔除它们。

2. 数据库模型定义

使用 SQLAlchemy 2.0 风格定义模型，清晰映射数据库表结构。

from sqlalchemy import Column, String, Integer, DateTime
from sqlalchemy.orm import declarative_base
from datetime import datetimeBase = declarative_base()class ShortLink(Base):__tablename__ = 'short_links'id = Column(Integer, primary_key=True, index=True)short_code = Column(String(10), unique=True, index=True, nullable=False)original_url = Column(String(2048), nullable=False)created_at = Column(DateTime, default=datetime.utcnow)click_count = Column(Integer, default=0)def __repr__(self):return f"<ShortLink(code={self.short_code}, url={self.original_url[:30]}...)>"

细节关注：

index=True：对 short_code 建索引，加速查询。
nullable=False：强制约束，防止脏数据入库。

3. 业务逻辑层

这是项目的灵魂。我们将缓存逻辑封装在 Service 层。

import redis
from typing import Optional
from .db.models import ShortLink
from .db.session import get_db_session
from .core.config import settingsclass ShortenerService:def __init__(self):self.redis_client = redis.Redis(host=settings.REDIS_HOST,port=settings.REDIS_PORT,decode_responses=True)self.db_session = get_db_session()async def create_short_link(self, url: str) -> dict:"""创建短链接，返回短码"""# 1. 检查缓存是否已有该 URLcache_key = f"link:reverse:{url}"cached_code = self.redis_client.get(cache_key)if cached_code:return {"short_code": cached_code, "url": url}# 2. 生成短码，处理碰撞max_retries = 5for _ in range(max_retries):code = generate_short_code()# 检查数据库唯一性existing = self.db_session.query(ShortLink).filter(ShortLink.short_code == code).first()if not existing:# 3. 入库new_link = ShortLink(short_code=code, original_url=url)self.db_session.add(new_link)self.db_session.commit()# 4. 写入缓存self.redis_client.set(f"link:code:{code}", url, ex=86400)self.redis_client.set(cache_key, code, ex=86400)return {"short_code": code, "url": url}raise Exception("Failed to generate unique short code")async def get_redirect(self, short_code: str) -> Optional[str]:"""获取重定向 URL"""cache_key = f"link:code:{short_code}"url = self.redis_client.get(cache_key)if url:# 异步增加点击计数 (这里简化为同步，生产环境用 MQ)self._increment_click(short_code)return url# 缓存未命中，查库db_link = self.db_session.query(ShortLink).filter(ShortLink.short_code == short_code).first()if db_link:self.redis_client.set(cache_key, db_link.original_url, ex=86400)self._increment_click(short_code)return db_link.original_urlreturn Nonedef _increment_click(self, code: str):"""自增点击数"""self.redis_client.incr(f"clicks:{code}")

逻辑剖析：

双重检查：先查 Redis，再查 DB。这是典型的 Cache-Aside 模式。
TTL 设置：缓存过期时间设为 24 小时，平衡内存占用与数据新鲜度。
异常处理：生成失败抛出明确异常，便于上层捕获并返回 500 错误。

4. API 路由层

FastAPI 自动处理序列化与验证，我们只需定义接口。

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, HttpUrl
from .services.shortener import ShortenerServicerouter = APIRouter()
service = ShortenerService()class URLCreate(BaseModel):url: HttpUrlclass URLResponse(BaseModel):short_code: strurl: str@router.post("/shorten", response_model=URLResponse)
async def shorten_url(payload: URLCreate):"""创建短链接"""try:result = await service.create_short_link(str(payload.url))return resultexcept Exception as e:raise HTTPException(status_code=500, detail=str(e))@router.get("/r/{short_code}")
async def redirect_url(short_code: str):"""重定向"""url = await service.get_redirect(short_code)if not url:raise HTTPException(status_code=404, detail="Link not found")return {"redirect": url}

注意：

HttpUrl：Pydantic 自动校验 URL 格式，非法输入直接返回 422。
async def：FastAPI 原生支持异步，提升并发性能。

运行与测试

环境准备

创建虚拟环境，安装依赖：

python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate
pip install fastapi uvicorn sqlalchemy redis pydantic python-dotenv

配置管理

使用 .env 文件管理敏感配置，切勿硬编码。

# app/core/config.py
from pydantic_settings import BaseSettings
from functools import lru_cacheclass Settings(BaseSettings):DATABASE_URL: str = "postgresql://user:pass@localhost/db"REDIS_HOST: str = "localhost"REDIS_PORT: int = 6379SECRET_KEY: str = "your-secret-key"class Config:env_file = ".env"@lru_cache()
def get_settings():return Settings()settings = get_settings()

启动服务

# app/main.py
from fastapi import FastAPI
from .api.v1.routes import router as v1_routerapp = FastAPI(title="Wuji Shortener API")app.include_router(v1_router, prefix="/api/v1")if __name__ == "__main__":import uvicornuvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=True)

测试用例

使用 httpx 进行异步测试，模拟真实请求。

# tests/test_api.py
import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app@pytest.mark.anyio
async def test_create_and_redirect():transport = ASGITransport(app=app)async with AsyncClient(transport=transport, base_url="http://test") as client:# 1. 创建短链resp = await client.post("/api/v1/shorten", json={"url": "https://example.com/very/long/url"})assert resp.status_code == 200data = resp.json()code = data["short_code"]# 2. 访问短链resp2 = await client.get(f"/api/v1/r/{code}")assert resp2.status_code == 200assert resp2.json()["redirect"] == "https://example.com/very/long/url"

运行测试：

pytest tests/ -v

看到 passed 绿灯，说明核心链路已通。

优化扩展方向

项目跑通了，但距离生产级还有距离。以下是进阶方向：

限流保护：在 Nginx 或 FastAPI 中间件中引入令牌桶算法，防止恶意刷接口。

# 伪代码：简单的内存限流
from collections import defaultdict
import timerate_limit = defaultdict(list)def check_rate_limit(ip: str, limit: int = 10, window: int = 60) -> bool:now = time.time()rate_limit[ip] = [t for t in rate_limit[ip] if now - t < window]if len(rate_limit[ip]) >= limit:return Falserate_limit[ip].append(now)return True

监控与日志：接入 Prometheus，暴露 /metrics 接口，监控 QPS、延迟和错误率。使用 structlog 替代标准 logging，输出 JSON 格式日志，便于 ELK 收集。
安全性加固：
- URL 白名单：禁止短链指向内网地址（SSRF 攻击防御）。
- HTTPS 强制：在 Nginx 层配置 301 重定向。
- 签名验证：对敏感操作增加 HMAC-SHA256 签名，防止篡改。
性能优化：
- 连接池：配置 SQLAlchemy 连接池大小，避免数据库连接耗尽。
- 压缩：启用 Gzip 压缩，减少传输体积。