FastAPI + MySQL + Redis 短链接生成与访问统计系统（含二维码）

FastAPI + MySQL + Redis 短链接生成与访问统计系统（含二维码）

涉及后端开发、缓存优化、数据统计和二维码生成等多个技术点。

🎯 核心功能模块

🗄️ 数据库设计 (MySQL)

-- 短链接主表 CREATE TABLE short_links ( id BIGINT AUTO_INCREMENT PRIMARY KEY, short_code VARCHAR(8) NOT NULL UNIQUE, -- 短码 original_url TEXT NOT NULL, -- 原始URL created_at DATETIME DEFAULT CURRENT_TIMESTAMP, expires_at DATETIME NULL, -- 过期时间，NULL表示永久 is_active TINYINT DEFAULT 1, -- 是否启用 total_clicks INT DEFAULT 0, -- 总点击数（冗余字段） INDEX idx_short_code (short_code), INDEX idx_expires (expires_at) ); -- 访问日志表（用于统计分析） CREATE TABLE click_logs ( id BIGINT AUTO_INCREMENT PRIMARY KEY, short_code VARCHAR(8) NOT NULL, ip_address VARCHAR(45), user_agent TEXT, referer VARCHAR(500), country VARCHAR(100), device_type VARCHAR(20), -- PC/Mobile/Tablet clicked_at DATETIME DEFAULT CURRENT_TIMESTAMP, INDEX idx_code_time (short_code, clicked_at) );

⚡ Redis缓存策略

Key设计： - short:{code} → original_url (字符串，TTL=1小时) - stats:{code}:hourly → ZSET (每小时点击数，member=时间戳) - stats:{code}:daily → ZSET (每天点击数) 流程： 1. 请求短链接 → 先查Redis 2. 命中 → 直接返回URL，异步写日志到消息队列 3. 未命中 → 查MySQL，写入Redis，返回URL

🔧 FastAPI 后端实现

项目结构

shortlink_service/ ├── main.py ├── models.py # SQLAlchemy模型 ├── schemas.py # Pydantic模型 ├── crud.py # 数据库操作 ├── utils.py # 工具函数（短码生成、二维码） ├── routers/ │ ├── link.py # 短链接CRUD │ └── stats.py # 统计数据接口 └── templates/ └── qrcode.html # 二维码展示页面

关键代码示例

1. 短码生成算法 (utils.py)

import string, random import hashlib import base62 def generate_short_code(url: str, length: int = 6) -> str: """ 方法一：基于MD5截取 """ hash_obj = hashlib.md5(url.encode()) hex_digest = hash_obj.hexdigest() # 将16进制转为62进制（0-9a-zA-Z） decimal_val = int(hex_digest[:8], 16) return base62.encode(decimal_val)[:length] def generate_random_code(length: int = 6) -> str: """ 方法二：随机生成（需检查冲突） """ chars = string.ascii_letters + string.digits return ''.join(random.choices(chars, k=length))

2. 创建短链接接口 (routers/link.py)

from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from .. import crud, schemas, utils from ..database import get_db from ..redis_client import redis_client router = APIRouter(prefix="/api/v1") @router.post("/shorten") async def create_short_link( req: schemas.ShortLinkCreate, db: Session = Depends(get_db) ): # 1. 生成短码 code = utils.generate_short_code(req.url) # 2. 检查冲突（极小概率） existing = crud.get_link_by_code(db, code) if existing: # 若冲突则重新生成（加盐） code = utils.generate_random_code() # 3. 写入数据库 link = crud.create_short_link(db, code, req.url, req.expire_days) # 4. 预热缓存 redis_client.setex(f"short:{code}", 3600, req.url) return { "short_code": code, "short_url": f"http://yourdomain.com/{code}", "original_url": req.url, "qrcode_url": f"/api/v1/qrcode/{code}" } @router.get("/{short_code}") async def redirect_to_url(short_code: str, request: Request, db: Session = Depends(get_db)): # 1. 从Redis获取 url = redis_client.get(f"short:{short_code}") if not url: # 2. 从DB获取 link = crud.get_active_link(db, short_code) if not link or (link.expires_at and link.expires_at < datetime.utcnow()): raise HTTPException(status_code=404, detail="Link expired or not found") url = link.original_url # 3. 写入缓存 redis_client.setex(f"short:{short_code}", 3600, url) # 4. 异步记录日志（使用后台任务） background_tasks.add_task(log_click, short_code, request) # 5. 更新总点击数（异步） background_tasks.add_task(crud.increment_clicks, db, short_code) # 6. 302重定向 return RedirectResponse(url=url, status_code=302)

3. 访问统计接口 (routers/stats.py)

@router.get("/stats/{short_code}") async def get_stats(short_code: str, period: str = "7d", db: Session = Depends(get_db)): """ period: 24h / 7d / 30d """ # 获取总点击数 link = crud.get_link_by_code(db, short_code) if not link: raise HTTPException(404) # 按时间粒度聚合 granularity = "hour" if period == "24h" else "day" raw_data = crud.get_click_timeline(db, short_code, period, granularity) # 设备分布 devices = crud.get_device_distribution(db, short_code) # 地域分布（需配合IP库） countries = crud.get_country_distribution(db, short_code) return { "total_clicks": link.total_clicks, "timeline": raw_data, "devices": devices, "countries": countries }

4. 二维码生成 (utils.py + router)

import qrcode from io import BytesIO from fastapi.responses import StreamingResponse def generate_qrcode(data: str) -> BytesIO: qr = qrcode.QRCode(box_size=10, border=4) qr.add_data(data) qr.make(fit=True) img = qr.make_image(fill_color="black", back_color="white") buf = BytesIO() img.save(buf, format="PNG") buf.seek(0) return buf @router.get("/qrcode/{short_code}") async def get_qrcode(short_code: str): short_url = f"http://yourdomain.com/{short_code}" buf = generate_qrcode(short_url) return StreamingResponse(buf, media_type="image/png")

📊 数据看板（ECharts示例）

你可以用FastAPI提供JSON数据，前端用ECharts展示：

// 每小时点击量柱状图 fetch(`/api/v1/stats/${code}?period=24h`) .then(res => res.json()) .then(data => { var chart = echarts.init(document.getElementById('chart')); chart.setOption({ title: { text: '过去24小时点击趋势' }, xAxis: { type: 'category', data: data.timeline.map(t => t.hour) }, yAxis: { type: 'value' }, series: [{ type: 'bar', data: data.timeline.map(t => t.count) }] }); });

🚀 部署与优化建议