url_shortener/

│

├── app.py # Flask application entry point

├── database.py # Database operations (context manager)

├── models.py # Data models and business logic

├── utils.py # Helper functions (code generation, validation)

├── decorators.py # Authentication and rate limiting decorators

├── config.py # Configuration settings

│

├── tests/

│ ├── __init__.py

│ ├── test_database.py

│ ├── test_models.py

│ └── test_api.py

│

├── logs/ # Log files directory

├── shortener.db # SQLite database (auto-created)

├── requirements.txt

└── README.md

import sqlite3

import logging

from contextlib import contextmanager

from datetime import datetime

logger = logging.getLogger(__name__)

DATABASE_PATH = “shortener.db”

@contextmanager

def get_db():

“””Context manager for database connections.”””

conn = sqlite3.connect(DATABASE_PATH)

conn.row_factory = sqlite3.Row

try:

yield conn

conn.commit()

except Exception as e:

conn.rollback()

logger.error(f”Database error: {e}”)

raise

finally:

conn.close()

def init_db():

“””Create tables if they don’t exist.”””

with get_db() as conn:

conn.execute(“””

CREATE TABLE IF NOT EXISTS urls (

id INTEGER PRIMARY KEY AUTOINCREMENT,

code TEXT UNIQUE NOT NULL,

original_url TEXT NOT NULL,

created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,

click_count INTEGER DEFAULT 0,

last_clicked TIMESTAMP

)

“””)

conn.execute(“””

CREATE INDEX IF NOT EXISTS idx_code ON urls(code)

“””)

logger.info(“Database initialized”)

import hashlib

import random

import string

import re

from urllib.parse import urlparse

def generate_code(url: str, length: int = 6) -> str:

“””Generate a short code from a URL using hashing.”””

hash_obj = hashlib.md5(url.encode())

# Take first `length` characters of hex digest

return hash_obj.hexdigest()[:length]

def generate_random_code(length: int = 6) -> str:

“””Generate a random short code (for custom URLs).”””

characters = string.ascii_letters + string.digits

return “”.join(random.choices(characters, k=length))

def is_valid_url(url: str) -> bool:

“””Validate URL format.”””

try:

result = urlparse(url)

return all([result.scheme, result.netloc])

except Exception:

return False

def normalize_url(url: str) -> str:

“””Add https:// if no scheme is present.”””

if not url.startswith((“http://”, “https://”)):

url = “https://” + url

return url

from database import get_db

from utils import generate_code, is_valid_url, normalize_url

import logging

from datetime import datetime

logger = logging.getLogger(__name__)

class URLShortener:

@staticmethod

def create_short_url(original_url: str, custom_code: str = None) -> dict:

“””Create a new short URL.”””

# Validate URL

if not is_valid_url(original_url):

raise ValueError(“Invalid URL format”)

original_url = normalize_url(original_url)

# Generate code

if custom_code:

code = custom_code

else:

code = generate_code(original_url)

with get_db() as conn:

try:

cursor = conn.execute(

“INSERT INTO urls (code, original_url) VALUES (?, ?)”,

(code, original_url)

)

return {

“code”: code,

“original_url”: original_url,

“short_url”: f”/{code}”

}

except sqlite3.IntegrityError:

# Code already exists, try with random code

if not custom_code:

from utils import generate_random_code

code = generate_random_code()

return URLShortener.create_short_url(original_url, custom_code=code)

raise ValueError(“Custom code already exists”)

@staticmethod

def get_original_url(code: str) -> str:

“””Get the original URL and increment click count.”””

with get_db() as conn:

cursor = conn.execute(

“SELECT original_url, click_count FROM urls WHERE code = ?”,

(code,)

)

row = cursor.fetchone()

if not row:

return None

# Increment click count

conn.execute(

“UPDATE urls SET click_count = click_count + 1, last_clicked = CURRENT_TIMESTAMP WHERE code = ?”,

(code,)

)

return row[“original_url”]

@staticmethod

def get_stats(code: str) -> dict:

“””Get statistics for a short URL.”””

with get_db() as conn:

cursor = conn.execute(

“SELECT code, original_url, created_at, click_count, last_clicked FROM urls WHERE code = ?”,

(code,)

)

row = cursor.fetchone()

return dict(row) if row else None

@staticmethod

def delete_url(code: str, api_key: str = None) -> bool:

# Simple authentication (in production, use proper auth)

if api_key != “secret-key-123”:

raise PermissionError(“Invalid API key”)

with get_db() as conn:

cursor = conn.execute(“DELETE FROM urls WHERE code = ?”, (code,))

return cursor.rowcount > 0

@staticmethod

def get_all_urls(limit: int = 100) -> list:

“””Get all shortened URLs (for admin).”””

with get_db() as conn:

cursor = conn.execute(

“SELECT code, original_url, click_count, created_at FROM urls ORDER BY created_at DESC LIMIT ?”,

(limit,)

)

return [dict(row) for row in cursor.fetchall()]

from functools import wraps

from flask import request, jsonify

import time

from collections import defaultdict

# Simple in-memory rate limiter (production would use Redis)

rate_limit_store = defaultdict(list)

def rate_limit(max_requests: int = 10, window_seconds: int = 60):

“””Decorator that limits request rate per IP.”””

def decorator(func):

@wraps(func)

def wrapper(*args, **kwargs):

ip = request.remote_addr or “unknown”

now = time.time()

# Clean old entries

rate_limit_store[ip] = [t for t in rate_limit_store[ip] if now – t < window_seconds]

if len(rate_limit_store[ip]) >= max_requests:

return jsonify({“error”: “Rate limit exceeded”}), 429

rate_limit_store[ip].append(now)

return func(*args, **kwargs)

return wrapper

return decorator

def require_api_key(func):

“””Decorator that requires a valid API key.”””

@wraps(func)

def wrapper(*args, **kwargs):

api_key = request.headers.get(“X-API-Key”)

if not api_key or api_key != “secret-key-123”:

return jsonify({“error”: “Invalid or missing API key”}), 401

return func(*args, **kwargs)

return wrapper

import logging

from flask import Flask, request, jsonify, redirect

from logging.handlers import RotatingFileHandler

import os

from database import init_db

from models import URLShortener

from decorators import rate_limit, require_api_key

from utils import is_valid_url, normalize_url

# Setup logging

def setup_logging():

if not os.path.exists(“logs”):

os.makedirs(“logs”)

file_handler = RotatingFileHandler(“logs/shortener.log”, maxBytes=10485760, backupCount=5)

file_handler.setLevel(logging.INFO)

formatter = logging.Formatter(“%(asctime)s – %(name)s – %(levelname)s – %(message)s”)

file_handler.setFormatter(formatter)

logger = logging.getLogger(__name__)

logger.setLevel(logging.INFO)

logger.addHandler(file_handler)

return logger

logger = setup_logging()

# Create Flask app

app = Flask(__name__)

# Initialize database

init_db()

logger.info(“URL Shortener API started”)

# Routes

@app.route(“/shorten”, methods=[“POST”])

@rate_limit(max_requests=20, window_seconds=60)

def shorten_url():

“””Create a shortened URL.”””

data = request.get_json()

if not data or “url” not in data:

return jsonify({“error”: “Missing ‘url’ field”}), 400

original_url = data[“url”]

custom_code = data.get(“custom_code”)

try:

result = URLShortener.create_short_url(original_url, custom_code)

logger.info(f”Created short URL: {result[‘code’]} -> {original_url}”)

return jsonify(result), 201

except ValueError as e:

return jsonify({“error”: str(e)}), 400

@app.route(“/<code>”, methods=[“GET”])

def redirect_to_url(code):

“””Redirect short code to original URL.”””

original_url = URLShortener.get_original_url(code)

if not original_url:

logger.warning(f”Short code not found: {code}”)

return jsonify({“error”: “Short URL not found”}), 404

logger.info(f”Redirected {code} -> {original_url}”)

return redirect(original_url)

@app.route(“/stats/<code>”, methods=[“GET”])

def get_stats(code):

“””Get statistics for a short URL.”””

stats = URLShortener.get_stats(code)

if not stats:

return jsonify({“error”: “Short URL not found”}), 404

return jsonify(stats)

@app.route(“/shorten/<code>”, methods=[“DELETE”])

@require_api_key

def delete_url(code):

“””Delete a short URL (requires API key).”””

try:

success = URLShortener.delete_url(code, request.headers.get(“X-API-Key”))

if success:

logger.info(f”Deleted short URL: {code}”)

return jsonify({“message”: “Deleted successfully”}), 200

return jsonify({“error”: “Short URL not found”}), 404

except PermissionError as e:

return jsonify({“error”: str(e)}), 401

@app.route(“/shorten”, methods=[“GET”])

@require_api_key

def list_all_urls():

“””List all shortened URLs (admin only).”””

limit = request.args.get(“limit”, 100, type=int)

urls = URLShortener.get_all_urls(limit)

return jsonify({“urls”: urls, “count”: len(urls)})

@app.route(“/health”, methods=[“GET”])

def health_check():

“””Health check endpoint.”””

return jsonify({“status”: “ok”, “service”: “URL Shortener”})

@app.errorhandler(404)

def not_found(error):

return jsonify({“error”: “Endpoint not found”}), 404

if __name__ == “__main__”:

app.run(debug=True, host=”0.0.0.0″, port=5000)

import unittest

import json

import sys

import os

sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))

from app import app

from database import get_db, init_db

class URLShortenerTestCase(unittest.TestCase):

def setUp(self):

self.app = app.test_client()

self.app.testing = True

# Reinitialize database for tests

with get_db() as conn:

conn.execute(“DROP TABLE IF EXISTS urls”)

init_db()

def test_shorten_url(self):

response = self.app.post(

“/shorten”,

data=json.dumps({“url”: “https://example.com/very/long/url”}),

content_type=”application/json”

)

self.assertEqual(response.status_code, 201)

data = json.loads(response.data)

self.assertIn(“code”, data)

self.assertIn(“short_url”, data)

def test_shorten_invalid_url(self):

response = self.app.post(

“/shorten”,

data=json.dumps({“url”: “not-a-valid-url”}),

content_type=”application/json”

)

self.assertEqual(response.status_code, 400)

data = json.loads(response.data)

self.assertIn(“error”, data)

def test_redirect(self):

# First, create a short URL

create_resp = self.app.post(

“/shorten”,

data=json.dumps({“url”: “https://example.com/test”}),

content_type=”application/json”

)

code = json.loads(create_resp.data)[“code”]

# Then test redirect

redirect_resp = self.app.get(f”/{code}”, follow_redirects=False)

self.assertEqual(redirect_resp.status_code, 302)

self.assertEqual(redirect_resp.location, “https://example.com/test”)

def test_stats(self):

# Create URL

create_resp = self.app.post(

“/shorten”,

data=json.dumps({“url”: “https://example.com/stats-test”}),

content_type=”application/json”

)

code = json.loads(create_resp.data)[“code”]

# Get stats

stats_resp = self.app.get(f”/stats/{code}”)

self.assertEqual(stats_resp.status_code, 200)

stats = json.loads(stats_resp.data)

self.assertEqual(stats[“click_count”], 0)

def test_health_check(self):

response = self.app.get(“/health”)

self.assertEqual(response.status_code, 200)

data = json.loads(response.data)

self.assertEqual(data[“status”], “ok”)

if __name__ == “__main__”:

unittest.main()

flask==2.3.0

pytest==7.4.0

coverage==7.3.0

# Create and activate virtual environment

python -m venv venv

source venv/bin/activate # On Windows: venv\Scripts\activate

# Install dependencies

pip install -r requirements.txt

# Run the application

python app.py

# In another terminal, test the API

curl -X POST http://localhost:5000/shorten \\

-H “Content-Type: application/json” \\

-d ‘{“url”: “https://www.python.org”}’

# Response example:

# {“code”: “abc123”, “original_url”: “https://www.python.org”, “short_url”: “/abc123”}

# Visit the short URL in your browser

open http://localhost:5000/abc123

# Get statistics

curl http://localhost:5000/stats/abc123

# Run tests

python -m unittest discover tests

# Run tests with coverage

coverage run -m unittest discover

coverage report