diff --git a/README.md b/README.md
index ff01fba..e480122 100644
--- a/README.md
+++ b/README.md
@@ -268,6 +268,62 @@ pip install dist/*.whl
pip install --upgrade dist/*.whl
```
+# ๐ก Endpoints
+
+# ๐ Cache Admin Endpoints (Authentication)
+
+To use the cache admin endpoints (`/cache/purge`, `/cache/remove`), you must configure a secret token in your environment and send it in the request header.
+Setup
+
+Add a token in your .env file:
+
+```
+CACHE_PURGE_TOKEN=your-secret-token
+```
+
+๐งช How to test
+
+PowerShell
+
+```
+Invoke-RestMethod `
+ -Method DELETE `
+ -Uri "http://127.0.0.1:8000/cache/purge" `
+ -Headers @{ "X-Cache-Token" = "your-secret-token" }
+```
+
+๐งน Remove a single cache entry
+
+```
+Invoke-RestMethod `
+ -Method PATCH `
+ -Uri "http://127.0.0.1:8000/cache/remove?key=abc123" `
+ -Headers @{ "X-Cache-Token" = "your-secret-token" }
+```
+
+๐ฅ๏ธ UI Endpoints
+
+| Method | Path | Description |
+| ------ | --------------- | ------------------------------------ |
+| GET | `/` | Home page (URL shortener UI) |
+| GET | `/recent` | Shows recently shortened URLs |
+| GET | `/{short_code}` | Redirects to the original URL |
+| GET | `/cache/list` | ๐ง Debug cache view (local/dev only) |
+| DELETE | `/cache/purge` | ๐งน Remove all entries from cache |
+| PATCH | `/cache/remove` | ๐งน Remove a single cache entry |
+
+๐ API Endpoints (v1)
+
+| Method | Path | Description |
+| ------ | ------------------- | ------------------------------------ |
+| POST | `/api/v1/shorten` | Create a short URL |
+| GET | `/api/v1/version` | Get API version |
+| GET | `/api/v1/health` | Health check (DB + cache status) |
+| GET | `/api/{short_code}` | Redirect to original URL |
+| GET | `/cache/list` | ๐ง Debug cache view (local/dev only) |
+| DELETE | `/cache/purge` | ๐งน Remove all entries from cache |
+| PATCH | `/cache/remove` | ๐งน Remove a single cache entry |
+
## License
๐Docs
diff --git a/app/api/fast_api.py b/app/api/fast_api.py
index d16c37a..d8d5a99 100644
--- a/app/api/fast_api.py
+++ b/app/api/fast_api.py
@@ -1,40 +1,19 @@
-import os
-import re
import traceback
-from datetime import datetime, timezone
-from typing import TYPE_CHECKING
-
-from fastapi import APIRouter, FastAPI, Request
-from fastapi.responses import HTMLResponse, JSONResponse
-from pydantic import BaseModel, Field
-
-if TYPE_CHECKING:
- from pymongo.errors import PyMongoError
-else:
- try:
- from pymongo.errors import PyMongoError
- except ImportError:
-
- class PyMongoError(Exception):
- pass
-
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
from app import __version__
-from app.utils import db
-from app.utils.cache import get_short_from_cache, set_cache_pair
-from app.utils.helper import generate_code, is_valid_url, sanitize_url
-
-SHORT_CODE_PATTERN = re.compile(r"^[A-Za-z0-9]{6}$")
-MAX_URL_LENGTH = 2048
+from app.routes import api_router, ui_router
app = FastAPI(
title="Tiny API",
version=__version__,
description="Tiny URL Shortener API built with FastAPI",
+ docs_url="/docs",
+ redoc_url="/redoc",
+ openapi_url="/openapi.json",
)
-api_v1 = APIRouter(prefix=os.getenv("API_VERSION", "/api/v1"), tags=["v1"])
-
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
@@ -45,167 +24,5 @@ async def global_exception_handler(request: Request, exc: Exception):
)
-class ShortenRequest(BaseModel):
- url: str = Field(..., examples=["https://abcdkbd.com"])
-
-
-class ShortenResponse(BaseModel):
- success: bool = True
- input_url: str
- short_code: str
- created_on: datetime
-
-
-class ErrorResponse(BaseModel):
- success: bool = False
- error: str
- input_url: str
- message: str
-
-
-class VersionResponse(BaseModel):
- version: str
-
-
-# -------------------------------------------------
-# Home
-# -------------------------------------------------
-@app.get("/", response_class=HTMLResponse, tags=["Home"])
-async def read_root(_: Request):
- return """
-
-
- ๐ tiny API ๐
-
-
-
-
-
-
- """
-
-
-@api_v1.post("/shorten", response_model=ShortenResponse, status_code=201)
-def shorten_url(payload: ShortenRequest):
- print(" SHORTEN ENDPOINT HIT ", payload.url)
- raw_url = payload.url.strip()
-
- if len(raw_url) > MAX_URL_LENGTH:
- return JSONResponse(
- status_code=413, content={"success": False, "input_url": payload.url}
- )
-
- original_url = sanitize_url(raw_url)
-
- if not is_valid_url(original_url):
- return JSONResponse(
- status_code=400,
- content={
- "success": False,
- "error": "INVALID_URL",
- "input_url": payload.url,
- "message": "Invalid URL",
- },
- )
-
- if db.collection is None:
- cached_short = get_short_from_cache(original_url)
- short_code = cached_short or generate_code()
- set_cache_pair(short_code, original_url)
- return {
- "success": True,
- "input_url": original_url,
- "short_code": short_code,
- "created_on": datetime.now(timezone.utc),
- }
-
- try:
- existing = db.collection.find_one({"original_url": original_url})
- except PyMongoError:
- existing = None
-
- if existing:
- return {
- "success": True,
- "input_url": original_url,
- "short_code": existing["short_code"],
- "created_on": existing["created_at"],
- }
-
- short_code = generate_code()
- try:
- db.collection.insert_one(
- {
- "short_code": short_code,
- "original_url": original_url,
- "created_at": datetime.now(timezone.utc),
- }
- )
- except PyMongoError:
- pass
-
- return {
- "success": True,
- "input_url": original_url,
- "short_code": short_code,
- "created_on": datetime.now(timezone.utc),
- }
-
-
-@app.get("/version")
-def api_version():
- return {"version": __version__}
-
-
-@api_v1.get("/help")
-def get_help():
- return {"message": "Welcome to Tiny API. Visit /docs for API documentation."}
-
-
-app.include_router(api_v1)
+app.include_router(api_router)
+app.include_router(ui_router)
diff --git a/app/main.py b/app/main.py
index 8f393ba..c960eeb 100644
--- a/app/main.py
+++ b/app/main.py
@@ -1,58 +1,90 @@
+# app/main.py
from contextlib import asynccontextmanager
from pathlib import Path
-from typing import Optional
import logging
+import traceback
+import asyncio
-from fastapi import FastAPI, Form, Request, status
-from fastapi.responses import HTMLResponse, PlainTextResponse, RedirectResponse, JSONResponse
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
from fastapi.staticfiles import StaticFiles
-from fastapi.templating import Jinja2Templates
from starlette.middleware.sessions import SessionMiddleware
-from app.api.fast_api import app as api_app
+from app.routes import ui_router
from app.utils import db
-from app.utils.cache import (
- get_from_cache,
- get_recent_from_cache,
- get_short_from_cache,
- rev_cache,
- set_cache_pair,
- url_cache,
-)
-from app.utils.config import DOMAIN, MAX_RECENT_URLS, SESSION_SECRET
-from app.utils.helper import (
- format_date,
- generate_code,
- is_valid_url,
- sanitize_url,
+from app.utils.cache import cleanup_expired
+
+# -----------------------------
+# Background cache cleanup task
+# -----------------------------
+from app.utils.config import (
+ CACHE_TTL,
+ SESSION_SECRET,
)
-from app.utils.qr import generate_qr_with_logo
+
+
+async def cache_health_check():
+ logger = logging.getLogger(__name__)
+ logger.info("๐งน Cache cleanup task started")
+
+ interval = max(1, CACHE_TTL // 3) # pure TTL-based
+
+ logger.info(f"๐ Cache cleanup interval set to {interval}s")
+
+ while True:
+ try:
+ cleanup_expired()
+ except Exception as e:
+ logger.error(f"Cache cleanup error: {e}")
+ await asyncio.sleep(interval)
# -----------------------------
-# Lifespan: env + DB connect ONCE
+# Lifespan: env + DB connect ONCE (DB-optional)
# -----------------------------
@asynccontextmanager
async def lifespan(app: FastAPI):
logger = logging.getLogger(__name__)
- logger.info("Application startup: Connecting to database...")
- db.connect_db()
- db.start_health_check()
+ logger.info("Application startup: Initializing services...")
+
+ # DB init (optional)
+ db_ok = db.connect_db()
+ if db_ok:
+ db.start_health_check()
+ logger.info("๐ข MongoDB enabled")
+ else:
+ logger.warning("๐ก MongoDB disabled (cache-only mode)")
+
+ # Cache TTL cleanup
+ cache_task = asyncio.create_task(cache_health_check())
+ logger.info("๐งน Cache TTL cleanup enabled")
+
logger.info("Application startup complete")
-
yield
-
+
logger.info("Application shutdown: Cleaning up...")
- await db.stop_health_check()
-
- # Close MongoDB client gracefully
+
+ # Stop cache task
+ cache_task.cancel()
+ try:
+ await cache_task
+ except asyncio.CancelledError:
+ logger.info("๐งน Cache cleanup task stopped")
+
+ # Stop DB health check
+ try:
+ await db.stop_health_check()
+ except Exception as e:
+ logger.error(f"Error stopping health check: {str(e)}")
+
+ # Close Mongo client if exists
try:
if db.client is not None:
db.client.close()
logger.info("MongoDB client closed")
except Exception as e:
logger.error(f"Error closing MongoDB client: {str(e)}")
-
+
logger.info("Application shutdown complete")
@@ -63,218 +95,21 @@ async def lifespan(app: FastAPI):
STATIC_DIR = BASE_DIR / "static"
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
-templates = Jinja2Templates(directory=str(BASE_DIR / "templates"))
-
-
-def build_short_url(short_code: str, request_host_url: str) -> str:
- base_url = DOMAIN.rstrip("/")
- return f"{base_url}/{short_code}"
-
-
-@app.get("/", response_class=HTMLResponse)
-async def index(request: Request):
- session = request.session
-
- new_short_url = session.pop("new_short_url", None)
- qr_enabled = session.pop("qr_enabled", False)
- qr_type = session.pop("qr_type", "short")
- original_url = session.pop("original_url", None)
- short_code = session.pop("short_code", None)
- info_message = session.pop("info_message", None)
- error = session.pop("error", None)
-
- qr_image = None
- qr_data = None
-
- if qr_enabled and new_short_url and short_code:
- qr_data = new_short_url if qr_type == "short" else original_url
- qr_filename = f"{short_code}.png"
- qr_dir = STATIC_DIR / "qr"
- qr_dir.mkdir(parents=True, exist_ok=True)
- generate_qr_with_logo(qr_data, str(qr_dir / qr_filename))
- qr_image = f"/static/qr/{qr_filename}"
-
- all_urls = db.get_recent_urls(MAX_RECENT_URLS) or get_recent_from_cache(
- MAX_RECENT_URLS
- )
-
- return templates.TemplateResponse(
- "index.html",
- {
- "request": request,
- "urls": all_urls,
- "new_short_url": new_short_url,
- "qr_image": qr_image,
- "qr_data": qr_data,
- "qr_enabled": qr_enabled,
- "original_url": original_url,
- "error": error,
- "info_message": info_message,
- "db_available": db.get_collection() is not None,
- },
- )
-@app.post("/shorten", response_class=RedirectResponse)
-async def create_short_url(
- request: Request,
- original_url: str = Form(""),
- generate_qr: Optional[str] = Form(None),
- qr_type: str = Form("short"),
-) -> RedirectResponse:
- logger = logging.getLogger(__name__)
-
- session = request.session
- qr_enabled = bool(generate_qr)
- original_url = sanitize_url(original_url)
-
- # Basic validation (FastAPI can also handle this via Pydantic)
- if not original_url or not is_valid_url(original_url):
- session["error"] = "Please enter a valid URL."
- return RedirectResponse("/", status_code=status.HTTP_303_SEE_OTHER)
-
- # 1. Try Cache First
- short_code: Optional[str] = get_short_from_cache(original_url)
-
- if not short_code:
- # 2. Try Database if connected
- if db.is_connected():
- existing = db.find_by_original_url(original_url)
- db_code = existing.get("short_code") if existing else None
- if isinstance(db_code, str):
- short_code = db_code
- set_cache_pair(short_code, original_url)
-
- # 3. Generate New if still None
- if not short_code:
- short_code = generate_code()
- set_cache_pair(short_code, original_url)
-
- # Only write to database if connected
- if db.is_connected():
- db.insert_url(short_code, original_url)
- else:
- logger.warning(f"Database not connected, URL {short_code} created in cache only")
- session["info_message"] = "URL created (database temporarily unavailable)"
-
- # --- TYPE GUARD FOR MYPY ---
- if not isinstance(short_code, str):
- session["error"] = "Internal server error: Code generation failed."
- return RedirectResponse("/", status_code=status.HTTP_303_SEE_OTHER)
-
- # Mypy now knows short_code is strictly 'str'
- new_short_url = build_short_url(short_code, DOMAIN)
-
- session.update(
- {
- "new_short_url": new_short_url,
- "qr_enabled": qr_enabled,
- "qr_type": qr_type,
- "original_url": original_url,
- "short_code": short_code,
- }
- )
-
- return RedirectResponse("/", status_code=status.HTTP_303_SEE_OTHER)
-
-
-@app.get("/recent", response_class=HTMLResponse)
-async def recent_urls(request: Request):
- recent_urls_list = db.get_recent_urls(
- MAX_RECENT_URLS
- ) or get_recent_from_cache(MAX_RECENT_URLS)
-
- normalized = []
- for item in recent_urls_list:
- normalized.append(
- {
- "short_code": item.get("short_code"),
- "original_url": item.get("original_url"),
- "created_at": item.get("created_at"),
- "visit_count": item.get("visit_count", 0),
- }
- )
-
- return templates.TemplateResponse(
- "recent.html",
- {
- "request": request,
- "urls": normalized,
- "format_date": format_date,
- },
+# -----------------------------
+# Global error handler
+# -----------------------------
+@app.exception_handler(Exception)
+async def global_exception_handler(request: Request, exc: Exception):
+ traceback.print_exc()
+ return JSONResponse(
+ status_code=500,
+ content={"success": False, "error": "INTERNAL_SERVER_ERROR"},
)
-@app.post("/delete/{short_code}")
-async def delete_url(request: Request, short_code: str):
- db.delete_by_short_code(short_code)
-
- cached = url_cache.pop(short_code, None)
- if cached:
- rev_cache.pop(cached.get("url"), None)
-
- return PlainTextResponse("", status_code=204)
-
-
-@app.get("/{short_code}")
-async def redirect_short(request: Request, short_code: str):
- logger = logging.getLogger(__name__)
- # Try cache first
- cached_url = get_from_cache(short_code)
- if cached_url:
- return RedirectResponse(cached_url)
-
- # Check if database is connected
- if not db.is_connected():
- logger.warning(f"Database not connected, cannot redirect {short_code}")
- return PlainTextResponse(
- "Service temporarily unavailable. Please try again later.",
- status_code=503,
- headers={"Retry-After": "30"}
- )
-
- # Try database
- doc = db.increment_visit(short_code)
- if doc:
- set_cache_pair(short_code, doc["original_url"])
- return RedirectResponse(doc["original_url"])
-
- return PlainTextResponse("Invalid or expired short URL", status_code=404)
-
-
-@app.get("/coming-soon", response_class=HTMLResponse)
-async def coming_soon(request: Request):
- return templates.TemplateResponse("coming-soon.html", {"request": request})
-
-
-@app.get("/health")
-async def health_check():
- """Health check endpoint showing database and cache status."""
- state = db.get_connection_state()
-
- response_data = {
- "database": state,
- "cache": {
- "enabled": True,
- "size": len(url_cache),
- }
- }
-
- status_code = 200 if state["connected"] else 503
- return JSONResponse(content=response_data, status_code=status_code)
-
-
-app.mount("/api", api_app)
-
-
-@app.get("/_debug/cache")
-async def debug_cache():
- return {
- "url_cache": url_cache,
- "rev_cache": rev_cache,
- "recent_from_cache": get_recent_from_cache(MAX_RECENT_URLS),
- "size": {
- "url_cache": len(url_cache),
- "rev_cache": len(rev_cache),
- },
- }
+# -----------------------------
+# Routers (UI + API)
+# -----------------------------
+app.include_router(ui_router) # UI routes at "/"
diff --git a/app/routes.py b/app/routes.py
new file mode 100644
index 0000000..88493cc
--- /dev/null
+++ b/app/routes.py
@@ -0,0 +1,377 @@
+import os
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Optional
+from app.utils.cache import list_cache_clean, clear_cache
+from fastapi import (
+ APIRouter,
+ Form,
+ Request,
+ status,
+ HTTPException,
+ BackgroundTasks,
+ Header,
+ Query,
+)
+from fastapi.responses import (
+ HTMLResponse,
+ PlainTextResponse,
+ RedirectResponse,
+ JSONResponse,
+)
+from fastapi.templating import Jinja2Templates
+from pydantic import BaseModel, Field
+
+from app import __version__
+from app.utils import db
+from app.utils.cache import (
+ get_from_cache,
+ get_recent_from_cache,
+ get_short_from_cache,
+ set_cache_pair,
+ increment_visit_cache,
+ url_cache,
+ remove_cache_key,
+ rev_cache,
+)
+from app.utils.config import DOMAIN, MAX_RECENT_URLS, CACHE_PURGE_TOKEN
+from app.utils.helper import generate_code, is_valid_url, sanitize_url, format_date
+from app.utils.qr import generate_qr_with_logo
+
+BASE_DIR = Path(__file__).resolve().parent
+templates = Jinja2Templates(directory=str(BASE_DIR / "templates"))
+
+# Routers
+ui_router = APIRouter()
+api_router = APIRouter()
+api_v1 = APIRouter(prefix=os.getenv("API_VERSION", "/api/v1"), tags=["v1"])
+
+
+# ---------------- UI ROUTES ----------------
+
+
+@ui_router.get("/", response_class=HTMLResponse)
+async def index(request: Request):
+ session = request.session
+
+ new_short_url = session.pop("new_short_url", None)
+ qr_enabled = session.pop("qr_enabled", False)
+ qr_type = session.pop("qr_type", "short")
+ original_url = session.pop("original_url", None)
+ short_code = session.pop("short_code", None)
+ info_message = session.pop("info_message", None)
+ error = session.pop("error", None)
+
+ qr_image = None
+ qr_data = None
+
+ if qr_enabled and new_short_url and short_code:
+ qr_data = new_short_url if qr_type == "short" else original_url
+ qr_filename = f"{short_code}.png"
+ qr_dir = BASE_DIR / "static" / "qr"
+ qr_dir.mkdir(parents=True, exist_ok=True)
+ generate_qr_with_logo(qr_data, str(qr_dir / qr_filename))
+ qr_image = f"/static/qr/{qr_filename}"
+
+ recent_urls = db.get_recent_urls(MAX_RECENT_URLS) or get_recent_from_cache(
+ MAX_RECENT_URLS
+ )
+
+ return templates.TemplateResponse(
+ "index.html",
+ {
+ "request": request,
+ "urls": recent_urls,
+ "new_short_url": new_short_url,
+ "qr_image": qr_image,
+ "qr_data": qr_data,
+ "qr_enabled": qr_enabled,
+ "original_url": original_url,
+ "error": error,
+ "info_message": info_message,
+ "db_available": db.get_collection() is not None,
+ },
+ )
+
+
+@ui_router.post("/shorten", response_class=RedirectResponse)
+async def create_short_url(
+ request: Request,
+ original_url: str = Form(""),
+ generate_qr: Optional[str] = Form(None),
+ qr_type: str = Form("short"),
+):
+ session = request.session
+ original_url = sanitize_url(original_url)
+
+ if not original_url or not is_valid_url(original_url):
+ session["error"] = "Please enter a valid URL."
+ return RedirectResponse("/", status_code=status.HTTP_303_SEE_OTHER)
+
+ short_code: Optional[str] = get_short_from_cache(original_url)
+
+ if not short_code and db.is_connected():
+ existing = db.find_by_original_url(original_url)
+ db_code = (existing.get("short_code") if existing else None) or (
+ existing.get("code") if existing else None
+ )
+ if isinstance(db_code, str):
+ short_code = db_code
+ set_cache_pair(short_code, original_url)
+
+ if not short_code:
+ short_code = generate_code()
+ set_cache_pair(short_code, original_url)
+ if db.is_connected():
+ db.insert_url(short_code, original_url)
+
+ session.update(
+ {
+ "new_short_url": f"{DOMAIN.rstrip('/')}/{short_code}",
+ "short_code": short_code,
+ "qr_enabled": bool(generate_qr),
+ "qr_type": qr_type,
+ "original_url": original_url,
+ }
+ )
+
+ return RedirectResponse("/", status_code=status.HTTP_303_SEE_OTHER)
+
+
+@ui_router.get("/recent", response_class=HTMLResponse)
+async def recent_urls(request: Request):
+ recent_urls_list = db.get_recent_urls(MAX_RECENT_URLS) or get_recent_from_cache(
+ MAX_RECENT_URLS
+ )
+
+ return templates.TemplateResponse(
+ "recent.html",
+ {
+ "request": request,
+ "urls": recent_urls_list,
+ "format_date": format_date,
+ "db_available": db.get_collection() is not None,
+ "get_visit_count_from_cache": increment_visit_cache,
+ },
+ )
+
+
+@ui_router.get("/cache/list")
+def cache_list_ui():
+ return list_cache_clean()
+
+
+@ui_router.delete("/cache/purge", response_class=PlainTextResponse)
+def cache_purge_ui(x_cache_token: str = Header(..., alias="X-Cache-Token")):
+ """
+ Force delete everything from cache (secured by header)
+ """
+ if x_cache_token != CACHE_PURGE_TOKEN:
+ raise HTTPException(status_code=401, detail="Unauthorized")
+
+ if not url_cache and not rev_cache:
+ return "No URLs in cache"
+
+ clear_cache()
+ return "cleared ALL"
+
+
+@ui_router.patch("/cache/remove")
+def cache_remove_one_ui(
+ key: str = Query(..., description="short_code OR original_url"),
+ x_cache_token: str = Header(..., alias="X-Cache-Token"),
+):
+ # ๐ Header security
+ if x_cache_token != CACHE_PURGE_TOKEN:
+ raise HTTPException(status_code=401, detail="Unauthorized")
+
+ removed = remove_cache_key(key)
+
+ if not removed:
+ raise HTTPException(
+ status_code=404,
+ detail="Key not found in cache.",
+ )
+
+ return {
+ "status": "deleted",
+ }
+
+
+@ui_router.get("/{short_code}")
+def redirect_short_ui(short_code: str, background_tasks: BackgroundTasks):
+ cached_url = get_from_cache(short_code)
+ if cached_url:
+ if db.is_connected():
+ background_tasks.add_task(db.increment_visit, short_code)
+ else:
+ increment_visit_cache(short_code)
+ return RedirectResponse(cached_url)
+
+ if db.is_connected():
+ doc = db.increment_visit(short_code)
+ if doc and doc.get("original_url"):
+ set_cache_pair(short_code, doc["original_url"])
+ return RedirectResponse(doc["original_url"])
+
+ recent_cache = get_recent_from_cache(MAX_RECENT_URLS)
+ for item in recent_cache or []:
+ code = item.get("short_code") or item.get("code")
+ if code == short_code:
+ original_url = item.get("original_url")
+ if original_url:
+ set_cache_pair(short_code, original_url)
+ return RedirectResponse(original_url)
+
+ return PlainTextResponse("Invalid short URL", status_code=404)
+
+
+@ui_router.delete("/recent/{short_code}")
+def delete_recent_api(short_code: str):
+ recent = get_recent_from_cache(MAX_RECENT_URLS) or []
+ removed_from_cache = False
+
+ # Try removing from cache (memory only)
+ for i, item in enumerate(recent):
+ code = item.get("short_code") or item.get("code")
+ if code == short_code:
+ recent.pop(i)
+ removed_from_cache = True
+ break
+
+ db_available = db.is_connected()
+ db_deleted = False
+
+ # If DB available โ rely ONLY on DB
+ if db_available:
+ db_deleted = db.delete_by_short_code(short_code)
+
+ if not db_deleted:
+ raise HTTPException(
+ status_code=404, detail=f"short_code '{short_code}' not found"
+ )
+
+ return {
+ "status": "deleted",
+ "short_code": short_code,
+ "db_deleted": True,
+ "db_available": True,
+ }
+
+ # If DB NOT available โ rely on cache only
+ if not removed_from_cache:
+ raise HTTPException(
+ status_code=404, detail=f"short_code '{short_code}' not found"
+ )
+
+ return {
+ "status": "deleted",
+ "short_code": short_code,
+ "db_deleted": False,
+ "db_available": False,
+ }
+
+
+# ---------------- API ROUTES ----------------
+
+
+@api_router.get("/", response_class=HTMLResponse, tags=["Home"])
+async def read_root(_: Request):
+ return """
+
+
+ ๐ tiny API ๐
+
+
+
+
+
+
+ """
+
+
+@api_router.get("/version")
+def api_version():
+ return {"version": __version__}
+
+
+class ShortenRequest(BaseModel):
+ url: str = Field(..., examples=["https://abcdkbd.com"])
+
+
+@api_v1.post("/shorten")
+def shorten_api(payload: ShortenRequest):
+ original_url = sanitize_url(payload.url)
+ if not is_valid_url(original_url):
+ return JSONResponse(status_code=400, content={"error": "INVALID_URL"})
+
+ short_code = get_short_from_cache(original_url)
+ if not short_code:
+ short_code = generate_code()
+ set_cache_pair(short_code, original_url)
+ if db.is_connected():
+ db.insert_url(short_code, original_url)
+
+ return {
+ "success": True,
+ "input_url": original_url,
+ "short_code": short_code,
+ "created_on": datetime.now(timezone.utc),
+ }
+
+
+@api_router.get("/health")
+def health():
+ return {
+ "db": db.get_connection_state(),
+ "cache_size": len(url_cache),
+ }
+
+
+api_router.include_router(api_v1)
diff --git a/app/templates/index.html b/app/templates/index.html
index 7a5ecd4..986b414 100644
--- a/app/templates/index.html
+++ b/app/templates/index.html
@@ -28,7 +28,9 @@ tiny URL
@@ -68,6 +70,7 @@ ๐ tiny URL