REST API v1 using django-ninja (#1397)

archivebox/api/apps.py

@@ -0,0 +1,5 @@
+from django.apps import AppConfig
+
+
+class APIConfig(AppConfig):
+    name = 'api'

archivebox/api/archive.py

@@ -0,0 +1,184 @@
+# archivebox_api.py
+from typing import List, Optional
+from enum import Enum
+from pydantic import BaseModel
+from ninja import Router
+from main import (
+    add,
+    remove,
+    update,
+    list_all,
+    ONLY_NEW,
+)  # Assuming these functions are defined in main.py
+
+
+# Schemas
+
+class StatusChoices(str, Enum):
+    indexed = 'indexed'
+    archived = 'archived'
+    unarchived = 'unarchived'
+    present = 'present'
+    valid = 'valid'
+    invalid = 'invalid'
+    duplicate = 'duplicate'
+    orphaned = 'orphaned'
+    corrupted = 'corrupted'
+    unrecognized = 'unrecognized'
+
+
+class AddURLSchema(BaseModel):
+    urls: List[str]
+    tag: str = ""
+    depth: int = 0
+    update: bool = not ONLY_NEW  # Default to the opposite of ONLY_NEW
+    update_all: bool = False
+    index_only: bool = False
+    overwrite: bool = False
+    init: bool = False
+    extractors: str = ""
+    parser: str = "auto"
+
+
+class RemoveURLSchema(BaseModel):
+    yes: bool = False
+    delete: bool = False
+    before: Optional[float] = None
+    after: Optional[float] = None
+    filter_type: str = "exact"
+    filter_patterns: Optional[List[str]] = None
+
+
+class UpdateSchema(BaseModel):
+    resume: Optional[float] = None
+    only_new: Optional[bool] = None
+    index_only: Optional[bool] = False
+    overwrite: Optional[bool] = False
+    before: Optional[float] = None
+    after: Optional[float] = None
+    status: Optional[StatusChoices] = None
+    filter_type: Optional[str] = 'exact'
+    filter_patterns: Optional[List[str]] = None
+    extractors: Optional[str] = ""
+
+
+class ListAllSchema(BaseModel):
+    filter_patterns: Optional[List[str]] = None
+    filter_type: str = 'exact'
+    status: Optional[StatusChoices] = None
+    after: Optional[float] = None
+    before: Optional[float] = None
+    sort: Optional[str] = None
+    csv: Optional[str] = None
+    json: bool = False
+    html: bool = False
+    with_headers: bool = False
+
+
+# API Router
+router = Router()
+
+
+@router.post("/add", response={200: dict})
+def api_add(request, payload: AddURLSchema):
+    try:
+        result = add(
+            urls=payload.urls,
+            tag=payload.tag,
+            depth=payload.depth,
+            update=payload.update,
+            update_all=payload.update_all,
+            index_only=payload.index_only,
+            overwrite=payload.overwrite,
+            init=payload.init,
+            extractors=payload.extractors,
+            parser=payload.parser,
+        )
+        # Currently the add function returns a list of ALL items in the DB, ideally only return new items
+        return {
+            "status": "success",
+            "message": "URLs added successfully.",
+            "result": str(result),
+        }
+    except Exception as e:
+        # Handle exceptions raised by the add function or during processing
+        return {"status": "error", "message": str(e)}
+
+
+@router.post("/remove", response={200: dict})
+def api_remove(request, payload: RemoveURLSchema):
+    try:
+        result = remove(
+            yes=payload.yes,
+            delete=payload.delete,
+            before=payload.before,
+            after=payload.after,
+            filter_type=payload.filter_type,
+            filter_patterns=payload.filter_patterns,
+        )
+        return {
+            "status": "success",
+            "message": "URLs removed successfully.",
+            "result": result,
+        }
+    except Exception as e:
+        # Handle exceptions raised by the remove function or during processing
+        return {"status": "error", "message": str(e)}
+
+
+@router.post("/update", response={200: dict})
+def api_update(request, payload: UpdateSchema):
+    try:
+        result = update(
+            resume=payload.resume,
+            only_new=payload.only_new,
+            index_only=payload.index_only,
+            overwrite=payload.overwrite,
+            before=payload.before,
+            after=payload.after,
+            status=payload.status,
+            filter_type=payload.filter_type,
+            filter_patterns=payload.filter_patterns,
+            extractors=payload.extractors,
+        )
+        return {
+            "status": "success",
+            "message": "Archive updated successfully.",
+            "result": result,
+        }
+    except Exception as e:
+        # Handle exceptions raised by the update function or during processing
+        return {"status": "error", "message": str(e)}
+
+
+@router.post("/list_all", response={200: dict})
+def api_list_all(request, payload: ListAllSchema):
+    try:
+        result = list_all(
+            filter_patterns=payload.filter_patterns,
+            filter_type=payload.filter_type,
+            status=payload.status,
+            after=payload.after,
+            before=payload.before,
+            sort=payload.sort,
+            csv=payload.csv,
+            json=payload.json,
+            html=payload.html,
+            with_headers=payload.with_headers,
+        )
+        # TODO: This is kind of bad, make the format a choice field
+        if payload.json:
+            return {"status": "success", "format": "json", "data": result}
+        elif payload.html:
+            return {"status": "success", "format": "html", "data": result}
+        elif payload.csv:
+            return {"status": "success", "format": "csv", "data": result}
+        else:
+            return {
+                "status": "success",
+                "message": "List generated successfully.",
+                "data": result,
+            }
+    except Exception as e:
+        # Handle exceptions raised by the list_all function or during processing
+        return {"status": "error", "message": str(e)}

archivebox/api/auth.py

@@ -0,0 +1,48 @@
+from django.contrib.auth import authenticate
+from ninja import Form, Router, Schema
+from ninja.security import HttpBearer
+
+from api.models import Token
+
+router = Router()
+
+
+class GlobalAuth(HttpBearer):
+    def authenticate(self, request, token):
+        try:
+            return Token.objects.get(token=token).user
+        except Token.DoesNotExist:
+            pass
+
+
+class AuthSchema(Schema):
+    email: str
+    password: str
+
+
+@router.post("/authenticate", auth=None)  # overriding global auth
+def get_token(request, auth_data: AuthSchema):
+    user = authenticate(username=auth_data.email, password=auth_data.password)
+    if user:
+        # Assuming a user can have multiple tokens and you want to create a new one every time
+        new_token = Token.objects.create(user=user)
+        return {"token": new_token.token, "expires": new_token.expiry_as_iso8601}
+    else:
+        return {"error": "Invalid credentials"}
+
+
+class TokenValidationSchema(Schema):
+    token: str
+
+
+@router.post("/validate_token", auth=None) # No authentication required for this endpoint
+def validate_token(request, token_data: TokenValidationSchema):
+    try:
+        # Attempt to authenticate using the provided token
+        user = GlobalAuth().authenticate(request, token_data.token)
+        if user:
+            return {"status": "valid"}
+        else:
+            return {"status": "invalid"}
+    except Token.DoesNotExist:
+        return {"status": "invalid"}

archivebox/api/migrations/0001_initial.py

@@ -0,0 +1,28 @@
+# Generated by Django 3.1.14 on 2024-04-09 18:52
+
+import api.models
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='Token',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('token', models.CharField(default=auth.models.hex_uuid, max_length=32, unique=True)),
+                ('created', models.DateTimeField(auto_now_add=True)),
+                ('expiry', models.DateTimeField(blank=True, null=True)),
+                ('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='tokens', to=settings.AUTH_USER_MODEL)),
+            ],
+        ),
+    ]

archivebox/api/models.py

@@ -0,0 +1,30 @@
+import uuid
+from datetime import timedelta
+
+from django.conf import settings
+from django.db import models
+from django.utils import timezone
+from django.utils.translation import gettext_lazy as _
+
+def hex_uuid():
+    return uuid.uuid4().hex
+
+
+class Token(models.Model):
+    user = models.ForeignKey(
+        settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="tokens"
+    )
+    token = models.CharField(max_length=32, default=hex_uuid, unique=True)
+    created = models.DateTimeField(auto_now_add=True)
+    expiry = models.DateTimeField(null=True, blank=True)
+
+    @property
+    def expiry_as_iso8601(self):
+        """Returns the expiry date of the token in ISO 8601 format or a date 100 years in the future if none."""
+        expiry_date = (
+            self.expiry if self.expiry else timezone.now() + timedelta(days=365 * 100)
+        )
+        return expiry_date.isoformat()
+
+    def __str__(self):
+        return self.token

archivebox/api/tests.py

@@ -0,0 +1,27 @@
+from django.test import TestCase
+from ninja.testing import TestClient
+from archivebox.api.archive import router as archive_router
+
+class ArchiveBoxAPITestCase(TestCase):
+    def setUp(self):
+        self.client = TestClient(archive_router)
+
+    def test_add_endpoint(self):
+        response = self.client.post("/add", json={"urls": ["http://example.com"], "tag": "test"})
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.json()["status"], "success")
+
+    def test_remove_endpoint(self):
+        response = self.client.post("/remove", json={"filter_patterns": ["http://example.com"]})
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.json()["status"], "success")
+
+    def test_update_endpoint(self):
+        response = self.client.post("/update", json={})
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.json()["status"], "success")
+
+    def test_list_all_endpoint(self):
+        response = self.client.post("/list_all", json={})
+        self.assertEqual(response.status_code, 200)
+        self.assertTrue("success" in response.json()["status"])

archivebox/core/settings.py

@@ -61,6 +61,7 @@
     'django.contrib.admin',
 
     'core',
+    'api',
 
     'django_extensions',
 ]

archivebox/core/urls.py

@@ -8,6 +8,13 @@
 
 from core.views import HomepageView, SnapshotView, PublicIndexView, AddView, HealthCheckView
 
+from ninja import NinjaAPI
+from api.auth import GlobalAuth
+
+api = NinjaAPI(auth=GlobalAuth())
+api.add_router("/auth/", "api.auth.router")
+api.add_router("/archive/", "api.archive.router")
+
 # GLOBAL_CONTEXT doesn't work as-is, disabled for now: https://github.com/ArchiveBox/ArchiveBox/discussions/1306
 # from config import VERSION, VERSIONS_AVAILABLE, CAN_UPGRADE
 # GLOBAL_CONTEXT = {'VERSION': VERSION, 'VERSIONS_AVAILABLE': VERSIONS_AVAILABLE, 'CAN_UPGRADE': CAN_UPGRADE}
@@ -36,6 +43,8 @@
     path('accounts/', include('django.contrib.auth.urls')),
     path('admin/', admin.site.urls),
 
+    path("api/", api.urls),
+
     # do not add extra_context like this as not all admin views (e.g. ModelAdmin.autocomplete_view accept extra kwargs)
     # path('admin/', admin.site.urls, {'extra_context': GLOBAL_CONTEXT}),