Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 56 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
199 changes: 8 additions & 191 deletions app/api/fast_api.py
Original file line number Diff line number Diff line change
@@ -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):
Expand All @@ -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 """
<html>
<head>
<title>🌙 tiny API 🌙</title>
<style>
body {
margin: 0;
height: 100vh;
display: flex;
align-items: center;
justify-content: center;
background: linear-gradient(180deg, #0b1220, #050b14);
font-family: "Poppins", system-ui, Arial, sans-serif;
color: #f8fafc;
}
.card {
background: rgba(255, 255, 255, 0.06);
backdrop-filter: blur(12px);
border-radius: 16px;
padding: 50px 40px;
text-align: center;
box-shadow: 0 20px 60px rgba(0, 0, 0, 0.5);
max-width: 520px;
width: 90%;
}
h1 {
font-size: 2.8em;
margin-bottom: 12px;
background: linear-gradient(90deg, #5ab9ff, #4cb39f);
-webkit-background-clip: text;
-webkit-text-fill-color: transparent;
}
p {
font-size: 1.1em;
color: #cbd5e1;
margin-bottom: 30px;
}
a {
display: inline-block;
padding: 14px 26px;
border-radius: 12px;
background: linear-gradient(90deg, #4cb39f, #5ab9ff);
color: #fff;
text-decoration: none;
font-weight: 700;
}
</style>
</head>
<body>
<div class="card">
<h1>🚀 tiny API</h1>
<p>FastAPI backend for the Tiny URL shortener</p>
<a href="/docs">View API Documentation</a>
</div>
</body>
</html>
"""


@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)
Loading