url_shortening_service/site/links/api.py

"""Internal JSON API for minting and managing short links."""

from __future__ import annotations

import json

from django.http import Http404, HttpRequest, JsonResponse
from django.views.decorators.http import require_GET, require_http_methods, require_POST

from links.auth import require_bearer
from links.models import ShortLink
from links.services import (
    CodeCollisionError,
    ValidationError,
    create_link,
    parse_expires_at,
    validate_target_url,
)


def _json_body(request: HttpRequest) -> dict | None:
    if not request.body:
        return {}
    try:
        data = json.loads(request.body)
    except json.JSONDecodeError:
        return None
    if not isinstance(data, dict):
        return None
    return data


def _iso(dt) -> str | None:
    if dt is None:
        return None
    return dt.isoformat().replace("+00:00", "Z")


def serialize_link(link: ShortLink) -> dict:
    from django.conf import settings

    origin = (settings.PUBLIC_SHORT_URL or "").rstrip("/")
    return {
        "code": link.code,
        "short_url": f"{origin}/{link.code}",
        "target_url": link.target_url,
        "title": link.title,
        "is_active": link.is_active,
        "click_count": link.click_count,
        "created_at": _iso(link.created_at),
    }


@require_bearer
@require_http_methods(["GET", "POST"])
def links_collection(request: HttpRequest):
    if request.method == "POST":
        return _create(request)
    return _list(request)


def _create(request: HttpRequest) -> JsonResponse:
    data = _json_body(request)
    if data is None:
        return JsonResponse({"detail": "invalid json"}, status=400)

    raw_url = data.get("target_url")
    try:
        target_url = validate_target_url(raw_url if isinstance(raw_url, str) else "")
        expires_at = parse_expires_at(data.get("expires_at"))
    except ValidationError as exc:
        return JsonResponse({"detail": str(exc)}, status=400)

    title = data.get("title") or ""
    if not isinstance(title, str):
        return JsonResponse({"detail": "invalid title"}, status=400)
    title = title[:200]

    external_ref = data.get("external_ref") or ""
    if not isinstance(external_ref, str):
        return JsonResponse({"detail": "invalid external_ref"}, status=400)
    external_ref = external_ref[:64]

    try:
        link, created = create_link(
            target_url=target_url,
            title=title,
            external_ref=external_ref,
            expires_at=expires_at,
            token_name=request.token_name,
        )
    except CodeCollisionError:
        return JsonResponse({"detail": "could not allocate a unique code"}, status=500)

    return JsonResponse(serialize_link(link), status=201 if created else 200)


def _list(request: HttpRequest) -> JsonResponse:
    qs = ShortLink.objects.all()

    external_ref = request.GET.get("external_ref")
    if external_ref is not None:
        qs = qs.filter(external_ref=external_ref)

    is_active = request.GET.get("is_active")
    if is_active is not None:
        lowered = is_active.lower()
        if lowered in {"true", "1"}:
            qs = qs.filter(is_active=True)
        elif lowered in {"false", "0"}:
            qs = qs.filter(is_active=False)
        else:
            return JsonResponse({"detail": "invalid is_active"}, status=400)

    try:
        limit = int(request.GET.get("limit", 20))
        offset = int(request.GET.get("offset", 0))
    except (TypeError, ValueError):
        return JsonResponse({"detail": "invalid pagination"}, status=400)

    limit = min(max(limit, 0), 100)
    offset = max(offset, 0)

    total = qs.count()
    rows = list(qs[offset : offset + limit])
    return JsonResponse(
        {
            "count": total,
            "limit": limit,
            "offset": offset,
            "results": [serialize_link(link) for link in rows],
        }
    )


@require_bearer
@require_GET
def link_detail(request: HttpRequest, code: str):
    try:
        link = ShortLink.objects.get(code=code)
    except ShortLink.DoesNotExist as exc:
        raise Http404() from exc
    return JsonResponse(serialize_link(link))


@require_bearer
@require_POST
def link_disable(request: HttpRequest, code: str):
    try:
        link = ShortLink.objects.get(code=code)
    except ShortLink.DoesNotExist as exc:
        raise Http404() from exc
    if link.is_active:
        link.is_active = False
        link.save(update_fields=["is_active", "updated_at"])
    return JsonResponse(serialize_link(link))