Что такое сервисы сокращения ссылок?

Сервисы сокращения ссылок

Сервис сокращения ссылок (URL Shortener) — это веб-приложение, которое преобразует длинные URL в короткие, легко запоминающиеся ссылки. Примеры: bit.ly, tinyurl.com, короткие ссылки в твиттере.

Основные функции

1. Сокращение URL: длинная ссылка → короткая
2. Редирект: переход по короткой ссылке на оригинальную
3. Статистика: отслеживание количества кликов
4. Кастомизация: создание ссылок с нужным кодом

Архитектура сервиса сокращения ссылок

from fastapi import FastAPI, HTTPException
from sqlalchemy import Column, String, Integer, DateTime, create_engine
from sqlalchemy.orm import Session, declarative_base
from datetime import datetime
import secrets
import string

app = FastAPI()
Base = declarative_base()
engine = create_engine("postgresql://user:password@localhost/urlshortener")

class URLMapping(Base):
    """Модель для хранения сокращённых ссылок"""
    __tablename__ = "url_mappings"
    
    id = Column(Integer, primary_key=True)
    short_code = Column(String(10), unique=True, index=True)
    original_url = Column(String(2048))
    created_at = Column(DateTime, default=datetime.utcnow)
    clicks = Column(Integer, default=0)
    user_id = Column(Integer)  # Для отслеживания кто создал

Base.metadata.create_all(engine)

def generate_short_code(length: int = 6) -> str:
    """Генерирует уникальный короткий код"""
    characters = string.ascii_letters + string.digits
    return ''.join(secrets.choice(characters) for _ in range(length))

@app.post("/api/v1/shorten")
async def shorten_url(original_url: str, custom_code: str = None, db: Session = None):
    """API для сокращения ссылки"""
    
    # Генерируем уникальный код
    if custom_code:
        existing = db.query(URLMapping).filter_by(short_code=custom_code).first()
        if existing:
            raise HTTPException(status_code=400, detail="Code already taken")
        short_code = custom_code
    else:
        while True:
            short_code = generate_short_code()
            existing = db.query(URLMapping).filter_by(short_code=short_code).first()
            if not existing:
                break
    
    # Сохраняем в БД
    mapping = URLMapping(
        short_code=short_code,
        original_url=original_url
    )
    db.add(mapping)
    db.commit()
    
    return {
        "short_url": f"https://short.url/{short_code}",
        "original_url": original_url
    }

@app.get("/{short_code}")
async def redirect_url(short_code: str, db: Session = None):
    """Редирект с отслеживанием"""
    
    # Находим оригинальную ссылку
    mapping = db.query(URLMapping).filter_by(short_code=short_code).first()
    
    if not mapping:
        raise HTTPException(status_code=404, detail="URL not found")
    
    # Увеличиваем счётчик кликов
    mapping.clicks += 1
    db.commit()
    
    # Редиректим пользователя
    from fastapi.responses import RedirectResponse
    return RedirectResponse(url=mapping.original_url)

@app.get("/api/v1/stats/{short_code}")
async def get_stats(short_code: str, db: Session = None):
    """Получить статистику по ссылке"""
    
    mapping = db.query(URLMapping).filter_by(short_code=short_code).first()
    
    if not mapping:
        raise HTTPException(status_code=404)
    
    return {
        "short_code": mapping.short_code,
        "original_url": mapping.original_url,
        "clicks": mapping.clicks,
        "created_at": mapping.created_at
    }

Оптимизация: кэширование

Для популярных ссылок кэшируем редиректы:

import redis

redis_client = redis.Redis(host="localhost", port=6379, db=0)
CACHE_TTL = 3600  # 1 час

@app.get("/{short_code}")
async def redirect_url_cached(short_code: str, db: Session = None):
    """Редирект с кэшированием"""
    
    # Сначала проверяем Redis
    cached_url = redis_client.get(f"url:{short_code}")
    if cached_url:
        # Кэш найден, но считаем клики в БД
        mapping = db.query(URLMapping).filter_by(short_code=short_code).first()
        if mapping:
            mapping.clicks += 1
            db.commit()
        return RedirectResponse(url=cached_url.decode())
    
    # Если нет в кэше, ищем в БД
    mapping = db.query(URLMapping).filter_by(short_code=short_code).first()
    if not mapping:
        raise HTTPException(status_code=404)
    
    # Сохраняем в кэш
    redis_client.setex(f"url:{short_code}", CACHE_TTL, mapping.original_url)
    
    # Увеличиваем счётчик
    mapping.clicks += 1
    db.commit()
    
    return RedirectResponse(url=mapping.original_url)

Масштабирование: кэширование кликов

import asyncio

# Вместо обновления БД при каждом клике, батчим обновления
click_cache = {}  # {short_code: count}

async def update_clicks_in_background():
    """Периодически обновляет клики в БД"""
    while True:
        await asyncio.sleep(5)  # Обновляем каждые 5 секунд
        
        if not click_cache:
            continue
        
        db = SessionLocal()
        for short_code, count in click_cache.items():
            mapping = db.query(URLMapping).filter_by(short_code=short_code).first()
            if mapping:
                mapping.clicks += count
        
        db.commit()
        db.close()
        click_cache.clear()

@app.get("/{short_code}")
async def redirect_url_batched(short_code: str):
    """Батчинг кликов"""
    
    cached_url = redis_client.get(f"url:{short_code}")
    if not cached_url:
        # Логика получения из БД
        pass
    
    # Увеличиваем счётчик в памяти
    click_cache[short_code] = click_cache.get(short_code, 0) + 1
    
    return RedirectResponse(url=cached_url.decode())

Обработка уникальности кодов

import hashlib
import base64

def generate_short_code_from_url(url: str) -> str:
    """Детерминированное сокращение на основе URL"""
    hash_obj = hashlib.md5(url.encode())
    hash_bytes = hash_obj.digest()[:6]  # Берём первые 6 байт
    return base64.urlsafe_b64encode(hash_bytes).decode()[:6]

# Проблема: коллизии при длинных URL
# Решение: если код занят, добавляем суффикс

def get_unique_code(original_url: str, db: Session) -> str:
    """Получить уникальный код"""
    base_code = generate_short_code_from_url(original_url)
    code = base_code
    counter = 0
    
    while db.query(URLMapping).filter_by(short_code=code).first():
        counter += 1
        code = f"{base_code}{counter}"
    
    return code

Проблемы при масштабировании

1. Высокая нагрузка на редиректы

# Решение: использовать CDN для часто запрашиваемых ссылок
from cloudflare import Cloudflare

cdn = Cloudflare(api_token="token")

def purge_cdn_cache(short_code: str):
    """Обновить кэш CDN"""
    cdn.purge(f"https://short.url/{short_code}")

2. Растущая БД

# Решение: партиционирование по времени создания
# или использование sharding по хешу short_code

# Пример: распределение записей по нескольким БД
SHARD_COUNT = 10

def get_shard(short_code: str) -> int:
    return hash(short_code) % SHARD_COUNT

def save_url(original_url: str, custom_code: str):
    shard_id = get_shard(custom_code)
    db = get_db_connection(shard_id)
    # Сохраняем в нужную шарду

3. Одновременные запросы

# Решение: использовать Redis для генерации уникальных ID

class DistributedSequence:
    def __init__(self, redis_client):
        self.redis = redis_client
    
    def next_code(self) -> str:
        """Генерирует уникальный ID через Redis"""
        counter = self.redis.incr("url_counter")
        # Преобразуем число в буквы (base62)
        return self.to_base62(counter)
    
    @staticmethod
    def to_base62(num: int) -> str:
        chars = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"
        if num == 0:
            return chars[0]
        arr = []
        while num:
            arr.append(chars[num % 62])
            num //= 62
        return ''.join(reversed(arr))

Полный пример с быстрым редиректом

from fastapi import FastAPI
from fastapi.responses import RedirectResponse
import redis
from sqlalchemy import create_engine, Column, String, Integer, DateTime
from sqlalchemy.orm import Session
from datetime import datetime

app = FastAPI()
redis_client = redis.Redis()

class URLMapping(Base):
    __tablename__ = "url_mappings"
    short_code = Column(String(10), primary_key=True, index=True)
    original_url = Column(String(2048))
    created_at = Column(DateTime, default=datetime.utcnow)
    clicks = Column(Integer, default=0)

@app.get("/api/v1/shorten")
def shorten(url: str):
    code = generate_code()
    db.add(URLMapping(short_code=code, original_url=url))
    db.commit()
    return {"short_code": code}

@app.get("/{code}")
def redirect(code: str):
    # Кэш в Redis
    cached = redis_client.get(code)
    if cached:
        url = cached.decode()
    else:
        mapping = db.query(URLMapping).get(code)
        if not mapping:
            return {"error": "Not found"}, 404
        url = mapping.original_url
        redis_client.setex(code, 3600, url)
    
    return RedirectResponse(url=url)

Сервис сокращения ссылок — классический пример системного дизайна, требующий масштабируемости, надёжности и производительности.