Implement short link redirect service

.env.example

@@ -1,4 +1,4 @@
-DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/akarpov
+DATABASE_URL=postgresql://postgres:postgres@127.0.0.1:5432/akarpov
 CELERY_BROKER_URL=redis://localhost:6379/0
 REDIS_URL=redis://localhost:6379/1
 REDIS_CACHE=rediscache://localhost:6379/1

compose/production/django/entrypoint

@@ -14,7 +14,7 @@ if [ -z "${POSTGRES_USER}" ]; then
     base_postgres_image_default_user='postgres'
     export POSTGRES_USER="${base_postgres_image_default_user}"
 fi
-export DATABASE_URL="postgres://${POSTGRES_USER}:${POSTGRES_PASSWORD}@${POSTGRES_HOST}:${POSTGRES_PORT}/${POSTGRES_DB}"
+export DATABASE_URL="postgresql://${POSTGRES_USER}:${POSTGRES_PASSWORD}@${POSTGRES_HOST}:${POSTGRES_PORT}/${POSTGRES_DB}"
 
 python << END
 import sys

pyproject.toml

@@ -94,6 +94,9 @@ django-location-field = "^2.7.0"
 pydantic = "^2.0.2"
 channels-redis = "^4.1.0"
 django-ipware = "^5.0.0"
+fastapi = {extras = ["all"], version = "^0.101.0"}
+sqlalchemy = "^2.0.19"
+pydantic-settings = "^2.0.2"
 
 
 [build-system]

redirect/__init__.py

@@ -0,0 +1 @@
+"""Service to redirect users from short links."""

redirect/__main__.py

@@ -0,0 +1,19 @@
+import os
+import uvicorn
+
+from redirect.settings import settings
+
+
+def main():
+    """Main entrypoint of the app."""
+    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings.local")
+    uvicorn.run(
+        "redirect.app:app",
+        host=settings.redirect_host,
+        port=settings.redirect_port,
+        reload=settings.redirect_reload
+    )
+
+
+if __name__ == '__main__':
+    main()

redirect/app.py

@@ -0,0 +1,39 @@
+from typing import Annotated
+
+import django
+from fastapi import FastAPI, Depends, HTTPException, Header
+from fastapi.responses import RedirectResponse
+from sqlalchemy.orm import Session
+from starlette.requests import Request
+
+from redirect.db.curd import get_link_by_slug
+from redirect.db.dependency import get_db
+
+
+django.setup()
+app = FastAPI()
+
+
+from akarpov.tools.shortener.tasks import save_view_meta
+
+
+@app.get("/{slug}")
+def redirect(slug: str, request: Request, db: Session = Depends(get_db), user_agent: Annotated[str | None, Header()] = None) -> RedirectResponse:
+    """Main route that redirects to a page based on the slug."""
+    if '+' in slug:
+        return RedirectResponse(url=f'/tools/shortener/p/{slug.replace("+", "")}')
+
+    link = get_link_by_slug(db, slug)
+    if link is None:
+        raise HTTPException(status_code=404, detail="Unknown Short Link")
+
+    save_view_meta.apply_async(
+        kwargs={
+            "pk": link[0],
+            "ip": request.client.host,
+            "user_agent": user_agent,
+            "user_id": 0,
+        },
+    )
+
+    return RedirectResponse(url=link[1])

redirect/db/__init__.py

@@ -0,0 +1,12 @@
+from sqlalchemy import create_engine, make_url
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+
+from redirect.settings import settings
+
+SQLALCHEMY_DATABASE_URL = make_url(settings.database_url)
+
+engine = create_engine(SQLALCHEMY_DATABASE_URL)
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+Base = declarative_base()

redirect/db/curd.py

@@ -0,0 +1,37 @@
+from typing import Optional
+
+from sqlalchemy.orm import Session
+
+from redirect.db import models
+from redirect.util import slug_to_link_id
+
+
+def get_link(db: Session, link_id: int):
+    """Gets short link metdata by its id.
+
+    :param db Open connection to a database
+    :param link_id id of the link
+
+    :return link metadata"""
+    return db.query(models.Link).filter(models.Link.id == link_id).first()
+
+
+_get_link_by_slug_cache = {}
+
+
+def get_link_by_slug(db: Session, slug: str) -> Optional[tuple[int, str]]:
+    """Converts slug to id and gets related link target.
+
+    :param db Open connection to a database
+    :param slug of the link
+
+    :return target link id & url"""
+    if slug in _get_link_by_slug_cache:
+        return _get_link_by_slug_cache[slug]
+
+    link = get_link(db, slug_to_link_id(slug))
+    if link is None or not link.enabled:
+        return None
+
+    _get_link_by_slug_cache[slug] = (link.id, link.source)
+    return link.id, link.source

redirect/db/dependency.py

@@ -0,0 +1,10 @@
+from redirect.db import SessionLocal
+
+
+def get_db():
+    """Route dependency that provides a database session."""
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

redirect/db/models.py

@@ -0,0 +1,17 @@
+from sqlalchemy import Boolean, Column, Integer, String, DateTime
+
+from redirect.db import Base
+
+
+class Link(Base):
+    """Model of a short link that defines slug and target of a redirect."""
+    __tablename__ = "short_link"
+
+    id = Column(Integer, primary_key=True, index=True)
+    created = Column(DateTime)
+    modified = Column(DateTime)
+    source = Column(String)
+    slug = Column(String, index=True)
+    enabled = Column(Boolean)
+    viewed = Column(Integer)
+    creator_id = Column(Integer, index=True)

redirect/settings.py

@@ -0,0 +1,22 @@
+from pydantic_settings import BaseSettings
+
+
+class AppSettings(BaseSettings):
+    """General configuration of the app."""
+
+    redirect_slug_cutoff: int = 3
+
+    redirect_host: str = "localhost"
+    redirect_port: int = 8000
+    redirect_reload: bool = False
+
+    database_url: str = "postgresql://postgres:postgres@127.0.0.1:5432/akarpov"
+
+    class Config:
+        env_prefix = ''
+        env_file = '.env'
+        env_file_encoding = 'utf-8'
+        extra = 'allow'
+
+
+settings = AppSettings()

redirect/util.py

@@ -0,0 +1,18 @@
+import string
+
+from redirect.settings import settings
+
+URL_CHARACTERS = list(string.ascii_letters + string.digits + ";,:@&-_.!~*'()#")
+SLUG_CUTOFF = settings.redirect_slug_cutoff
+
+
+def slug_to_link_id(slug: str) -> int:
+    """Converts given slug to an id of a link."""
+    link_id = 0
+    try:
+        for i, ch in enumerate(slug[:SLUG_CUTOFF - 1:-1]):
+            value = URL_CHARACTERS.index(ch)
+            link_id += value * len(URL_CHARACTERS) ** i
+    except ValueError:
+        pass
+    return link_id