feat: refactor db calls to url service

.vscode/extensions.json

@@ -2,10 +2,11 @@
     "recommendations": [
         "ms-python.python",
         "ms-python.debugpy",
+        "ms-python.pylint",
         "mtxr.sqltools",
         "mtxr.sqltools-driver-pg",
-        "genieai.chatgpt-vscode",
+        "humao.rest-client",
-        "humao.rest-client"
+        "genieai.chatgpt-vscode"
     ],
     "unwantedRecommendations": [
     ]

.vscode/http/url.rest

@@ -1,10 +0,0 @@
-### redirect_to_original_url
-GET http://localhost:8000/J7aDVn
-
-### create_url_shortcode
-POST http://localhost:8000/shorten
-Content-Type: application/json
-
-{
-    "url": "https://google.com/"
-}

.vscode/sql/minxadb-dev.session.sql

@@ -1 +1,2 @@
-SELECT * FROM url_mapping
+
+SELECT * FROM urls

app/config.py

@@ -11,6 +11,5 @@ class Settings(BaseSettings):
         env_file = ".env"
         env_file_encoding = "utf-8"
 
-
 # Singleton instance of settings
 settings = Settings()

app/db/database.py

@@ -1,7 +1,8 @@
-from app.config import settings
 from databases import Database
 from sqlalchemy import create_engine, MetaData
-from sqlalchemy.orm import declarative_base, sessionmaker
+from sqlalchemy.orm import declarative_base
+
+from app.config import settings
 
 # for Alembic migrations and sync operations
 engine = create_engine(settings.database_url)

app/exceptions.py

@@ -0,0 +1,8 @@
+class ShortcodeConflict(Exception):
+    pass
+
+class ShortcodeNotFound(Exception):
+    pass
+
+class InvalidShortcode(Exception):
+    pass

app/main.py

@@ -1,4 +1,5 @@
 from fastapi import FastAPI
+
 from app.config import settings
 from app.db import database, Base, engine
 from app.routes import shortener_router

app/models/url.py

@@ -1,4 +1,5 @@
 from sqlalchemy import Column, Integer, String
+
 from app.db.database import Base
 
 class Url(Base):

app/routes/shortener.py

@@ -1,47 +1,34 @@
 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.exceptions import ShortcodeNotFound, ShortcodeConflict
+from app.utils import encoder
 from app.schemas import UrlPayload
-from app.services import encoder
+from app.services import UrlService
 
 router = APIRouter()
+url_service = UrlService(db=database)
 
 @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")
+        raise HTTPException(status_code=400, detail=f"Shortcode '{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({
+        url = await url_service.get_url_by_shortcode(shortcode)
-            "id": next_id,
+        return RedirectResponse(url=url, status_code=302)
-            "url": original_url,
+    except ShortcodeNotFound as e:
-            "shortcode": shortcode
+        raise HTTPException(status_code=404, detail=str(e)) from e
-        })
-        await database.execute(query)
 
-        return {"shortcode": shortcode, "original_url": original_url}
+
-    except IntegrityError as e:
+@router.post("/")
-        if 'shortcode' in str(e):
+async def create_url_shortcode(url_payload: UrlPayload):
-            raise HTTPException(status_code=409, detail="Shortcode already in use")
+    url = str(url_payload.url)
-        else:
+
-            raise HTTPException(status_code=500, detail="Internal Server Error")
+    try:
+        return await url_service.generate_shortcode_and_save(url)
+    except ShortcodeConflict as e:
+        raise HTTPException(status_code=409, detail=str(e)) from e
+    except:
+        raise HTTPException(status_code=500, detail="Internal Server Error") from None

app/services/__init__.py

@@ -1 +1 @@
-from .encoder import encoder
+from .url_service import UrlService

app/services/url_service.py

@@ -0,0 +1,36 @@
+from sqlalchemy import select, insert
+from sqlalchemy.exc import IntegrityError
+
+from app.utils import encoder
+from app.models import Url
+from app.exceptions import ShortcodeConflict, ShortcodeNotFound
+
+class UrlService:
+    def __init__(self, db):
+        self.db = db
+
+    async def get_url_by_shortcode(self, shortcode: str) -> str:
+        query = select(Url.url).where(Url.shortcode == shortcode)
+        result = await self.db.fetch_val(query)
+
+        if result is None:
+            raise ShortcodeNotFound(f"Shortcode '{shortcode}' not found")
+
+        return result
+
+    async def generate_shortcode_and_save(self, url: str):
+        try:
+            # Get the next ID from the sequence
+            next_id = await self.db.fetch_val("SELECT nextval('urls_id_seq')")
+            shortcode = encoder.encode(next_id)
+
+            # Insert the new URL entry
+            insert_stmt = insert(Url).values(id=next_id, url=url, shortcode=shortcode)
+            await self.db.execute(insert_stmt)
+
+            return {"shortcode": shortcode, "url": url}
+
+        except IntegrityError as e:
+            if 'shortcode' in str(e):
+                raise ShortcodeConflict(f"Shortcode '{shortcode}' already in use") from None
+            raise

app/utils/__init__.py

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

app/utils/encoder.py

@@ -1,6 +1,7 @@
-from app.config import settings
 from hashids import Hashids
 
+from app.config import settings
+
 class ShortCodeEncoder:
     def __init__(self, salt=None, alphabet=None, min_length=6):
         self.salt = salt or settings.hashids_salt
@@ -9,8 +10,8 @@ class ShortCodeEncoder:
         self.hashids = Hashids(salt=self.salt, min_length=min_length, alphabet=self.alphabet)
         self.min_length = min_length
 
-    def encode(self, id: int) -> str:
+    def encode(self, url_id: int) -> str:
-        return self.hashids.encode(id)
+        return self.hashids.encode(url_id)
 
     def decode(self, shortcode: str) -> int:
         decoded = self.hashids.decode(shortcode)

requirements.txt

@@ -1,24 +1,31 @@
 alembic==1.16.4
 annotated-types==0.7.0
 anyio==4.9.0
+astroid==3.3.11
 asyncpg==0.30.0
 click==8.2.1
 databases==0.9.0
+dill==0.4.0
 fastapi==0.116.1
 greenlet==3.2.3
 h11==0.16.0
 hashids==1.3.1
 idna==3.10
+isort==6.0.1
 Mako==1.3.10
 MarkupSafe==3.0.2
+mccabe==0.7.0
+platformdirs==4.3.8
 psycopg2-binary==2.9.10
 pydantic==2.11.7
 pydantic-settings==2.10.1
 pydantic_core==2.33.2
+pylint==3.3.7
 python-dotenv==1.1.1
 sniffio==1.3.1
 SQLAlchemy==2.0.41
 starlette==0.47.2
+tomlkit==0.13.3
 typing-inspection==0.4.1
 typing_extensions==4.14.1
 uvicorn==0.35.0