Refactoring file structure, not tested yet
This commit is contained in:
@@ -1 +1,3 @@
|
||||
DEBUG=True
|
||||
DATABASE_URL=postgresql+psycopg2://user:password@localhost:5432/url_db
|
||||
BASE_URL=http://localhost:8000
|
||||
@@ -24,4 +24,4 @@ http://127.0.0.1:8000/shorten
|
||||
PYTHONPATH=./ pytest
|
||||
export PYTHONPATH=$(pwd)
|
||||
pytest
|
||||
```
|
||||
```
|
||||
@@ -0,0 +1,70 @@
|
||||
# app/api/routes.py
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from fastapi.responses import RedirectResponse
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.db.session import get_db
|
||||
from app.schemas.url import (
|
||||
URLCreate,
|
||||
URLResponse,
|
||||
URLStatsResponse,
|
||||
MessageResponse,
|
||||
)
|
||||
from app.services import url_service
|
||||
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.get("/", response_model=MessageResponse)
|
||||
def home():
|
||||
return {"message": "URL Shortener API"}
|
||||
|
||||
|
||||
@router.post("/shorten", response_model=URLResponse)
|
||||
def shorten_url(
|
||||
request: URLCreate,
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
short_url = url_service.create_short_url(db, str(request.url))
|
||||
return {"short_url": short_url}
|
||||
|
||||
|
||||
@router.get("/{short_code}")
|
||||
def redirect_to_url(
|
||||
short_code: str,
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
original_url = url_service.get_original_url(db, short_code)
|
||||
|
||||
if not original_url:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Shortened URL not found",
|
||||
)
|
||||
|
||||
url_service.increment_clicks(db, short_code)
|
||||
|
||||
return RedirectResponse(url=original_url)
|
||||
|
||||
|
||||
@router.get("/stats/{short_code}", response_model=URLStatsResponse)
|
||||
def get_url_stats(
|
||||
short_code: str,
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
stats = url_service.get_stats(db, short_code)
|
||||
|
||||
if not stats:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="URL stats not found",
|
||||
)
|
||||
|
||||
return stats
|
||||
|
||||
|
||||
@router.get("/health", response_model=MessageResponse)
|
||||
def health_check():
|
||||
return {"message": "healthy"}
|
||||
@@ -0,0 +1,36 @@
|
||||
# app/core/config.py
|
||||
|
||||
from functools import lru_cache
|
||||
from pydantic_settings import BaseSettings, SettingsConfigDict
|
||||
|
||||
|
||||
class Settings(BaseSettings):
|
||||
# App
|
||||
app_name: str = "URL Shortener API"
|
||||
debug: bool = False
|
||||
|
||||
# Server
|
||||
host: str = "0.0.0.0"
|
||||
port: int = 8000
|
||||
base_url: str = "http://localhost:8000"
|
||||
|
||||
# Database
|
||||
database_url: str = "sqlite:///./urls.db"
|
||||
|
||||
# Optional future extensions
|
||||
allowed_hosts: list[str] = ["*"]
|
||||
rate_limit_per_minute: int = 60
|
||||
|
||||
model_config = SettingsConfigDict(
|
||||
env_file=".env",
|
||||
env_file_encoding="utf-8",
|
||||
case_sensitive=False,
|
||||
extra="ignore",
|
||||
)
|
||||
|
||||
@lru_cache
|
||||
def get_settings() -> Settings:
|
||||
return Settings()
|
||||
|
||||
# Singleton-style access
|
||||
settings = get_settings()
|
||||
@@ -0,0 +1,6 @@
|
||||
# app/db/base.py
|
||||
|
||||
from sqlalchemy.orm import DeclarativeBase
|
||||
|
||||
class Base(DeclarativeBase):
|
||||
pass
|
||||
@@ -0,0 +1,57 @@
|
||||
# app/db/models.py
|
||||
|
||||
from sqlalchemy import String, Integer, DateTime, func, Index
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
from datetime import datetime
|
||||
from app.db.base import Base
|
||||
|
||||
|
||||
class URL(Base):
|
||||
__tablename__ = "urls"
|
||||
|
||||
id: Mapped[int] = mapped_column(
|
||||
Integer,
|
||||
primary_key=True,
|
||||
index=True
|
||||
)
|
||||
|
||||
short_code: Mapped[str] = mapped_column(
|
||||
String(10),
|
||||
unique=True,
|
||||
nullable=False,
|
||||
index=True
|
||||
)
|
||||
|
||||
original_url: Mapped[str] = mapped_column(
|
||||
String(2048),
|
||||
nullable=False,
|
||||
index=True
|
||||
)
|
||||
|
||||
clicks: Mapped[int] = mapped_column(
|
||||
Integer,
|
||||
default=0,
|
||||
nullable=False
|
||||
)
|
||||
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True),
|
||||
server_default=func.now(),
|
||||
nullable=False
|
||||
)
|
||||
|
||||
last_accessed: Mapped[datetime | None] = mapped_column(
|
||||
DateTime(timezone=True),
|
||||
nullable=True
|
||||
)
|
||||
|
||||
# Additional compound index for faster lookups
|
||||
__table_args__ = (
|
||||
Index("idx_short_original", "short_code", "original_url"),
|
||||
)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return (
|
||||
f"<URL(id={self.id}, short_code='{self.short_code}', "
|
||||
f"original_url='{self.original_url}', clicks={self.clicks})>"
|
||||
)
|
||||
@@ -0,0 +1,26 @@
|
||||
# app/db/session.py
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from app.core.config import settings
|
||||
|
||||
# Create SQLAlchemy engine
|
||||
engine = create_engine(
|
||||
settings.database_url,
|
||||
pool_pre_ping=True, # helps prevent stale connections
|
||||
)
|
||||
|
||||
# Session factory
|
||||
SessionLocal = sessionmaker(
|
||||
autocommit=False,
|
||||
autoflush=False,
|
||||
bind=engine,
|
||||
)
|
||||
|
||||
# Dependency for FastAPI
|
||||
def get_db():
|
||||
db = SessionLocal()
|
||||
try:
|
||||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
+17
-113
@@ -1,124 +1,28 @@
|
||||
from fastapi import FastAPI, HTTPException
|
||||
from fastapi.responses import RedirectResponse
|
||||
# app/main.py
|
||||
|
||||
from fastapi import FastAPI
|
||||
from contextlib import asynccontextmanager
|
||||
from app.utils import generate_short_code
|
||||
from pydantic import BaseModel, HttpUrl
|
||||
from pydantic_settings import BaseSettings
|
||||
import sqlite3
|
||||
import logging
|
||||
from app.core.config import settings
|
||||
from app.db.session import engine
|
||||
from app.db.base import Base
|
||||
from app.api.routes import router
|
||||
|
||||
class AppSettings(BaseSettings):
|
||||
base_url: str = "http://localhost:8000"
|
||||
database_url: str = "urls.db"
|
||||
|
||||
settings = AppSettings() # Load environment variables
|
||||
|
||||
# Set up logging configuration
|
||||
logging.basicConfig(level=logging.INFO,
|
||||
format="%(asctime)s - %(levelname)s - %(message)s")
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Database connection setup
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
conn = sqlite3.connect(settings.database_url)
|
||||
conn.execute("""
|
||||
CREATE TABLE IF NOT EXISTS urls (
|
||||
id INTEGER PRIMARY KEY,
|
||||
short_code TEXT UNIQUE,
|
||||
original_url TEXT NOT NULL,
|
||||
clicks INTEGER DEFAULT 0
|
||||
);
|
||||
""")
|
||||
conn.execute("""
|
||||
CREATE INDEX IF NOT EXISTS idx_short_code ON urls(short_code);
|
||||
""")
|
||||
conn.commit()
|
||||
logger.info("Database initialized or already exists.")
|
||||
|
||||
Base.metadata.create_all(bind=engine)
|
||||
yield
|
||||
conn.close()
|
||||
logger.info("Database connection closed.")
|
||||
|
||||
app = FastAPI(lifespan=lifespan)
|
||||
|
||||
def get_db_connection():
|
||||
return sqlite3.connect(settings.database_url, check_same_thread=False)
|
||||
|
||||
class URLRequest(BaseModel):
|
||||
url: HttpUrl
|
||||
|
||||
@app.get("/")
|
||||
def home():
|
||||
logger.info("Home endpoint accessed.")
|
||||
return {"message": "URL Shortener API"}
|
||||
|
||||
@app.post("/shorten")
|
||||
def shorten_url(request: URLRequest):
|
||||
conn = get_db_connection()
|
||||
|
||||
existing = conn.execute(
|
||||
"SELECT short_code FROM urls WHERE original_url = ?",
|
||||
(str(request.url),)
|
||||
).fetchone()
|
||||
|
||||
if existing:
|
||||
conn.close()
|
||||
return {"short_url": f"{settings.base_url}/{existing[0]}"}
|
||||
|
||||
short_code = generate_short_code()
|
||||
while conn.execute(
|
||||
"SELECT 1 FROM urls WHERE short_code = ?",
|
||||
(short_code,)
|
||||
).fetchone():
|
||||
short_code = generate_short_code()
|
||||
|
||||
conn.execute(
|
||||
"INSERT INTO urls (short_code, original_url, clicks) VALUES (?, ?, 0)",
|
||||
(short_code, str(request.url))
|
||||
def create_app() -> FastAPI:
|
||||
app = FastAPI(
|
||||
title=settings.app_name,
|
||||
debug=settings.debug,
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
return {"short_url": f"{settings.base_url}/{short_code}"}
|
||||
app.include_router(router)
|
||||
|
||||
@app.get("/{short_code}")
|
||||
def redirect_to_url(short_code: str):
|
||||
conn = get_db_connection()
|
||||
url_data = conn.execute(
|
||||
"SELECT original_url FROM urls WHERE short_code = ?",
|
||||
(short_code,)
|
||||
).fetchone()
|
||||
return app
|
||||
|
||||
if url_data is None:
|
||||
conn.close()
|
||||
raise HTTPException(status_code=404, detail="Shortened URL not found")
|
||||
|
||||
original_url = url_data[0]
|
||||
|
||||
conn.execute(
|
||||
"UPDATE urls SET clicks = clicks + 1 WHERE short_code = ?",
|
||||
(short_code,)
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
return RedirectResponse(url=original_url)
|
||||
|
||||
@app.get("/stats/{short_code}")
|
||||
def get_url_stats(short_code: str):
|
||||
logger.info(f"Stats request received for short_code: {short_code}")
|
||||
conn = get_db_connection()
|
||||
url_data = conn.execute("SELECT original_url, clicks FROM urls WHERE short_code = ?", (short_code,)).fetchone()
|
||||
|
||||
if url_data is None:
|
||||
logger.warning(f"Stats not found for {short_code}.")
|
||||
conn.close()
|
||||
raise HTTPException(status_code=404, detail="URL stats not found")
|
||||
|
||||
original_url, clicks = url_data
|
||||
|
||||
conn.close()
|
||||
|
||||
logger.info(f"Returning stats for {original_url}. Total clicks: {clicks}")
|
||||
|
||||
return {"original_url": original_url, "clicks": clicks}
|
||||
app = create_app()
|
||||
app.router.lifespan_context = lifespan
|
||||
@@ -0,0 +1,24 @@
|
||||
# app/schemas/url.py
|
||||
|
||||
from pydantic import BaseModel, HttpUrl
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
|
||||
class URLCreate(BaseModel):
|
||||
url: HttpUrl
|
||||
|
||||
|
||||
class URLResponse(BaseModel):
|
||||
short_url: str
|
||||
|
||||
|
||||
class URLStatsResponse(BaseModel):
|
||||
original_url: HttpUrl
|
||||
clicks: int
|
||||
created_at: datetime
|
||||
last_accessed: Optional[datetime] = None
|
||||
|
||||
|
||||
class MessageResponse(BaseModel):
|
||||
message: str
|
||||
@@ -0,0 +1,104 @@
|
||||
# app/services/url_service.py
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy import select
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
from app.db.models import URL
|
||||
from app.utils.short_code import generate_short_code
|
||||
from app.core.config import settings
|
||||
|
||||
|
||||
SHORT_CODE_LENGTH = 6
|
||||
MAX_GENERATION_ATTEMPTS = 5
|
||||
|
||||
|
||||
def create_short_url(db: Session, original_url: str) -> str:
|
||||
"""
|
||||
Create a shortened URL or return existing one if already present.
|
||||
"""
|
||||
|
||||
# 1️⃣ Check if URL already exists
|
||||
existing = db.execute(
|
||||
select(URL).where(URL.original_url == original_url)
|
||||
).scalar_one_or_none()
|
||||
|
||||
if existing:
|
||||
return f"{settings.base_url}/{existing.short_code}"
|
||||
|
||||
# 2️⃣ Generate unique short code with collision handling
|
||||
for _ in range(MAX_GENERATION_ATTEMPTS):
|
||||
short_code = generate_short_code(SHORT_CODE_LENGTH)
|
||||
|
||||
collision = db.execute(
|
||||
select(URL).where(URL.short_code == short_code)
|
||||
).scalar_one_or_none()
|
||||
|
||||
if not collision:
|
||||
new_url = URL(
|
||||
short_code=short_code,
|
||||
original_url=original_url,
|
||||
)
|
||||
db.add(new_url)
|
||||
db.commit()
|
||||
db.refresh(new_url)
|
||||
|
||||
return f"{settings.base_url}/{new_url.short_code}"
|
||||
|
||||
# If we somehow fail multiple times
|
||||
raise RuntimeError("Failed to generate unique short code")
|
||||
|
||||
|
||||
def get_original_url(db: Session, short_code: str) -> Optional[str]:
|
||||
"""
|
||||
Retrieve original URL for redirection.
|
||||
"""
|
||||
|
||||
url_entry = db.execute(
|
||||
select(URL).where(URL.short_code == short_code)
|
||||
).scalar_one_or_none()
|
||||
|
||||
if not url_entry:
|
||||
return None
|
||||
|
||||
return url_entry.original_url
|
||||
|
||||
|
||||
def increment_clicks(db: Session, short_code: str) -> bool:
|
||||
"""
|
||||
Increment click counter and update last_accessed timestamp.
|
||||
"""
|
||||
|
||||
url_entry = db.execute(
|
||||
select(URL).where(URL.short_code == short_code)
|
||||
).scalar_one_or_none()
|
||||
|
||||
if not url_entry:
|
||||
return False
|
||||
|
||||
url_entry.clicks += 1
|
||||
url_entry.last_accessed = datetime.utcnow()
|
||||
|
||||
db.commit()
|
||||
return True
|
||||
|
||||
|
||||
def get_stats(db: Session, short_code: str) -> Optional[dict]:
|
||||
"""
|
||||
Return URL statistics.
|
||||
"""
|
||||
|
||||
url_entry = db.execute(
|
||||
select(URL).where(URL.short_code == short_code)
|
||||
).scalar_one_or_none()
|
||||
|
||||
if not url_entry:
|
||||
return None
|
||||
|
||||
return {
|
||||
"original_url": url_entry.original_url,
|
||||
"clicks": url_entry.clicks,
|
||||
"created_at": url_entry.created_at,
|
||||
"last_accessed": url_entry.last_accessed,
|
||||
}
|
||||
@@ -1,7 +0,0 @@
|
||||
# app/settings.py
|
||||
|
||||
import os
|
||||
|
||||
# Example configuration
|
||||
database_url = os.getenv("DATABASE_URL", "sqlite:///default.db")
|
||||
debug = os.getenv("DEBUG", "True") == "True"
|
||||
@@ -0,0 +1,64 @@
|
||||
# tests/conftest.py
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.main import app
|
||||
from app.db.base import Base
|
||||
from app.db.session import get_db
|
||||
|
||||
|
||||
# Use in-memory SQLite for tests
|
||||
TEST_DATABASE_URL = "sqlite+pysqlite:///:memory:"
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def db_session():
|
||||
"""
|
||||
Creates a new database session for a test.
|
||||
Rolls back everything after test finishes.
|
||||
"""
|
||||
|
||||
engine = create_engine(
|
||||
TEST_DATABASE_URL,
|
||||
connect_args={"check_same_thread": False},
|
||||
)
|
||||
|
||||
TestingSessionLocal = sessionmaker(
|
||||
autocommit=False,
|
||||
autoflush=False,
|
||||
bind=engine,
|
||||
)
|
||||
|
||||
# Create tables
|
||||
Base.metadata.create_all(bind=engine)
|
||||
|
||||
db = TestingSessionLocal()
|
||||
|
||||
try:
|
||||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
Base.metadata.drop_all(bind=engine)
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def client(db_session):
|
||||
"""
|
||||
Overrides get_db dependency to use test database.
|
||||
"""
|
||||
|
||||
def override_get_db():
|
||||
try:
|
||||
yield db_session
|
||||
finally:
|
||||
pass
|
||||
|
||||
app.dependency_overrides[get_db] = override_get_db
|
||||
|
||||
with TestClient(app) as c:
|
||||
yield c
|
||||
|
||||
app.dependency_overrides.clear()
|
||||
@@ -0,0 +1,93 @@
|
||||
# tests/test_routes.py
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
def test_home(client):
|
||||
response = client.get("/")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"message": "URL Shortener API"}
|
||||
|
||||
|
||||
def test_health(client):
|
||||
response = client.get("/health")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"message": "healthy"}
|
||||
|
||||
|
||||
def test_shorten_url(client):
|
||||
response = client.post(
|
||||
"/shorten",
|
||||
json={"url": "https://google.com"},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
|
||||
assert "short_url" in data
|
||||
assert data["short_url"].startswith("http://")
|
||||
|
||||
|
||||
def test_shorten_invalid_url(client):
|
||||
response = client.post(
|
||||
"/shorten",
|
||||
json={"url": "not-a-valid-url"},
|
||||
)
|
||||
|
||||
assert response.status_code == 422 # Pydantic validation error
|
||||
|
||||
|
||||
def test_duplicate_url_returns_same_code(client):
|
||||
r1 = client.post("/shorten", json={"url": "https://example.com"})
|
||||
r2 = client.post("/shorten", json={"url": "https://example.com"})
|
||||
|
||||
assert r1.status_code == 200
|
||||
assert r2.status_code == 200
|
||||
|
||||
assert r1.json()["short_url"] == r2.json()["short_url"]
|
||||
|
||||
|
||||
def test_redirect_success(client):
|
||||
shorten = client.post(
|
||||
"/shorten",
|
||||
json={"url": "https://redirect-test.com"},
|
||||
)
|
||||
|
||||
short_url = shorten.json()["short_url"]
|
||||
code = short_url.split("/")[-1]
|
||||
|
||||
response = client.get(f"/{code}", follow_redirects=False)
|
||||
|
||||
assert response.status_code in (302, 307)
|
||||
|
||||
|
||||
def test_redirect_404(client):
|
||||
response = client.get("/nonexistent", follow_redirects=False)
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_stats_success(client):
|
||||
shorten = client.post(
|
||||
"/shorten",
|
||||
json={"url": "https://stats-test.com"},
|
||||
)
|
||||
|
||||
short_url = shorten.json()["short_url"]
|
||||
code = short_url.split("/")[-1]
|
||||
|
||||
# Trigger one click
|
||||
client.get(f"/{code}", follow_redirects=False)
|
||||
|
||||
stats = client.get(f"/stats/{code}")
|
||||
|
||||
assert stats.status_code == 200
|
||||
data = stats.json()
|
||||
|
||||
assert data["original_url"] == "https://stats-test.com"
|
||||
assert data["clicks"] == 1
|
||||
assert "created_at" in data
|
||||
|
||||
|
||||
def test_stats_404(client):
|
||||
response = client.get("/stats/doesnotexist")
|
||||
assert response.status_code == 404
|
||||
Reference in New Issue
Block a user