feat: add new encoder and restructure project

app/__init__.py

@@ -0,0 +1,2 @@
+# app/__init__.py
+"""App package initializer. Required for Python imports."""

app/config.py

@@ -0,0 +1,16 @@
+from pydantic_settings import BaseSettings
+from typing import Literal
+
+class Settings(BaseSettings):
+    database_url: str
+    environment: Literal["dev", "stage", "prod"]
+    hashids_salt: str
+    encoder_alphabet: str
+
+    class Config:
+        env_file = ".env"
+        env_file_encoding = "utf-8"
+
+
+# Singleton instance of settings
+settings = Settings()

app/db/__init__.py

@@ -0,0 +1 @@
+from .database import database, Base, engine

app/db/database.py

@@ -0,0 +1,13 @@
+from app.config import settings
+from databases import Database
+from sqlalchemy import create_engine, MetaData
+from sqlalchemy.orm import declarative_base, sessionmaker
+
+# for Alembic migrations and sync operations
+engine = create_engine(settings.database_url)
+# to use with async requests
+database = Database(settings.database_url)
+
+metadata = MetaData()
+Base = declarative_base(metadata=metadata)
+

app/main.py

@@ -0,0 +1,20 @@
+from fastapi import FastAPI
+from app.config import settings
+from app.db import database, Base, engine
+from app.routes import shortener_router
+
+app = FastAPI()
+
+@app.on_event("startup")
+async def startup():
+    await database.connect()
+
+    # only create missing tables in dev environment
+    if settings.environment == "dev":
+        Base.metadata.create_all(bind=engine)
+
+@app.on_event("shutdown")
+async def shutdown():
+    await database.disconnect()
+
+app.include_router(shortener_router)

app/models/__init__.py

@@ -0,0 +1 @@
+from .url import Url

app/models/url.py

@@ -0,0 +1,9 @@
+from sqlalchemy import Column, Integer, String
+from app.db.database import Base
+
+class Url(Base):
+    __tablename__ = "urls"
+
+    id = Column(Integer, primary_key=True, index=True)
+    url = Column(String, nullable=False)
+    shortcode = Column(String, index=True, unique=True, nullable=False)

app/routes/__init__.py

@@ -0,0 +1 @@
+from .shortener import router as shortener_router

app/routes/shortener.py

@@ -0,0 +1,47 @@
+from fastapi import APIRouter, HTTPException
+from fastapi.responses import RedirectResponse
+from sqlalchemy.exc import IntegrityError
+
+from app.db import database
+from app.models import Url
+from app.schemas import UrlPayload
+from app.services import encoder
+
+router = APIRouter()
+
+@router.get("/{shortcode}")
+async def redirect_to_original_url(shortcode: str):
+    if not encoder.is_valid(shortcode):
+        raise HTTPException(status_code=400, detail="Shortcode is not valid")
+
+    query = Url.__table__.select().where(Url.shortcode == shortcode)
+    url_result = await database.fetch_one(query)
+
+    if url_result:
+        return RedirectResponse(url=url_result.url, status_code=302)
+    else:
+        raise HTTPException(status_code=404, detail="Shortcode not found")
+
+
+@router.post("/shorten")
+async def create_url_shortcode(url_payload: UrlPayload):
+    original_url = str(url_payload.url)
+
+    query = "SELECT nextval('urls_id_seq')"
+    next_id = await database.fetch_val(query)
+    shortcode = encoder.encode(next_id)
+
+    try:
+        query = Url.__table__.insert().values({
+            "id": next_id,
+            "url": original_url,
+            "shortcode": shortcode
+        })
+        await database.execute(query)
+
+        return {"shortcode": shortcode, "original_url": original_url}
+    except IntegrityError as e:
+        if 'shortcode' in str(e):
+            raise HTTPException(status_code=409, detail="Shortcode already in use")
+        else:
+            raise HTTPException(status_code=500, detail="Internal Server Error")

app/schemas/__init__.py

@@ -0,0 +1 @@
+from .url import UrlPayload

app/schemas/url.py

@@ -0,0 +1,4 @@
+from pydantic import BaseModel, HttpUrl
+
+class UrlPayload(BaseModel):
+    url: HttpUrl

app/services/__init__.py

@@ -0,0 +1 @@
+from .encoder import encoder

app/services/encoder.py

@@ -0,0 +1,27 @@
+from app.config import settings
+from hashids import Hashids
+
+class ShortCodeEncoder:
+    def __init__(self, salt=None, alphabet=None, min_length=6):
+        self.salt = salt or settings.hashids_salt
+        self.alphabet = alphabet or settings.encoder_alphabet
+
+        self.hashids = Hashids(salt=self.salt, min_length=min_length, alphabet=self.alphabet)
+        self.min_length = min_length
+
+    def encode(self, id: int) -> str:
+        return self.hashids.encode(id)
+
+    def decode(self, shortcode: str) -> int:
+        decoded = self.hashids.decode(shortcode)
+        return decoded[0] if decoded else None
+
+    def is_valid(self, shortcode: str) -> bool:
+        if len(shortcode) < self.min_length:
+            return False
+        if any(char not in self.alphabet for char in shortcode):
+            return False
+        return True
+
+# Singleton instance of encoder
+encoder = ShortCodeEncoder()