Use Django-ninja for API
This commit is contained in:
parent
88c958ef12
commit
869a931bda
13 changed files with 355 additions and 432 deletions
78
feedvault/api.py
Normal file
78
feedvault/api.py
Normal file
|
|
@ -0,0 +1,78 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from django.http import HttpRequest # noqa: TCH002
|
||||
from ninja import ModelSchema, NinjaAPI
|
||||
from ninja.pagination import paginate
|
||||
|
||||
from feedvault.models import Domain, Entry, Feed
|
||||
|
||||
api_v1 = NinjaAPI(
|
||||
title="FeedVault API",
|
||||
version="0.1.0",
|
||||
description="FeedVault API",
|
||||
urls_namespace="api_v1",
|
||||
)
|
||||
|
||||
|
||||
class FeedOut(ModelSchema):
|
||||
class Meta:
|
||||
model = Feed
|
||||
fields: str = "__all__"
|
||||
|
||||
|
||||
class EntriesOut(ModelSchema):
|
||||
class Meta:
|
||||
model = Entry
|
||||
fields: str = "__all__"
|
||||
|
||||
|
||||
class DomainsOut(ModelSchema):
|
||||
class Meta:
|
||||
model = Domain
|
||||
fields: str = "__all__"
|
||||
|
||||
|
||||
@api_v1.get("/feeds/", response=list[FeedOut])
|
||||
@paginate
|
||||
def list_feeds(request: HttpRequest) -> None:
|
||||
"""Get a list of feeds."""
|
||||
return Feed.objects.all() # type: ignore # noqa: PGH003
|
||||
|
||||
|
||||
@api_v1.get("/feeds/{feed_id}/", response=FeedOut)
|
||||
def get_feed(request: HttpRequest, feed_id: int) -> Feed:
|
||||
"""Get a feed by ID."""
|
||||
return Feed.objects.get(id=feed_id)
|
||||
|
||||
|
||||
@api_v1.get("/feeds/{feed_id}/entries/", response=list[EntriesOut])
|
||||
@paginate
|
||||
def list_entries(request: HttpRequest, feed_id: int) -> list[Entry]:
|
||||
"""Get a list of entries for a feed."""
|
||||
return Entry.objects.filter(feed_id=feed_id) # type: ignore # noqa: PGH003
|
||||
|
||||
|
||||
@api_v1.get("/entries/", response=list[EntriesOut])
|
||||
@paginate
|
||||
def list_all_entries(request: HttpRequest) -> list[Entry]:
|
||||
"""Get a list of entries."""
|
||||
return Entry.objects.all() # type: ignore # noqa: PGH003
|
||||
|
||||
|
||||
@api_v1.get("/entries/{entry_id}/", response=EntriesOut)
|
||||
def get_entry(request: HttpRequest, entry_id: int) -> Entry:
|
||||
"""Get an entry by ID."""
|
||||
return Entry.objects.get(id=entry_id)
|
||||
|
||||
|
||||
@api_v1.get("/domains/", response=list[DomainsOut])
|
||||
@paginate
|
||||
def list_domains(request: HttpRequest) -> list[Domain]:
|
||||
"""Get a list of domains."""
|
||||
return Domain.objects.all() # type: ignore # noqa: PGH003
|
||||
|
||||
|
||||
@api_v1.get("/domains/{domain_id}/", response=DomainsOut)
|
||||
def get_domain(request: HttpRequest, domain_id: int) -> Domain:
|
||||
"""Get a domain by ID."""
|
||||
return Domain.objects.get(id=domain_id)
|
||||
|
|
@ -0,0 +1,23 @@
|
|||
# Generated by Django 5.0.3 on 2024-03-15 16:42
|
||||
|
||||
from django.db import migrations
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('feedvault', '0002_alter_feed_status'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.RenameField(
|
||||
model_name='entry',
|
||||
old_name='_id',
|
||||
new_name='entry_id',
|
||||
),
|
||||
migrations.RenameField(
|
||||
model_name='feed',
|
||||
old_name='_id',
|
||||
new_name='feed_id',
|
||||
),
|
||||
]
|
||||
|
|
@ -177,7 +177,7 @@ class Feed(models.Model):
|
|||
)
|
||||
|
||||
icon = models.TextField(blank=True)
|
||||
_id = models.TextField(blank=True)
|
||||
feed_id = models.TextField(blank=True)
|
||||
image = JSONField(null=True, blank=True)
|
||||
info = models.TextField(blank=True)
|
||||
info_detail = JSONField(null=True, blank=True)
|
||||
|
|
@ -249,7 +249,7 @@ class Entry(models.Model):
|
|||
enclosures = JSONField(null=True, blank=True)
|
||||
expired = models.TextField(blank=True)
|
||||
expired_parsed = models.DateTimeField(null=True, blank=True)
|
||||
_id = models.TextField(blank=True)
|
||||
entry_id = models.TextField(blank=True)
|
||||
license = models.TextField(blank=True)
|
||||
link = models.TextField(blank=True)
|
||||
links = JSONField(null=True, blank=True)
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
from dotenv import find_dotenv, load_dotenv
|
||||
|
|
@ -37,6 +38,10 @@ SITE_ID = 1
|
|||
PASSWORD_HASHERS: list[str] = ["django.contrib.auth.hashers.Argon2PasswordHasher"]
|
||||
ROOT_URLCONF = "feedvault.urls"
|
||||
WSGI_APPLICATION = "feedvault.wsgi.application"
|
||||
NINJA_PAGINATION_PER_PAGE = 1000
|
||||
|
||||
# Is True when running tests, used for not spamming Discord when new users are created
|
||||
TESTING: bool = len(sys.argv) > 1 and sys.argv[1] == "test"
|
||||
|
||||
INSTALLED_APPS: list[str] = [
|
||||
"feedvault.apps.FeedVaultConfig",
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ import os
|
|||
from typing import TYPE_CHECKING
|
||||
|
||||
from discord_webhook import DiscordWebhook
|
||||
from django.conf import settings
|
||||
from django.contrib.auth.models import User
|
||||
from django.db.models.signals import post_save
|
||||
from django.dispatch import receiver
|
||||
|
|
@ -16,7 +17,7 @@ logger: logging.Logger = logging.getLogger(__name__)
|
|||
|
||||
|
||||
@receiver(post_save, sender=User)
|
||||
def notify_when_new_user(sender: User, instance: User, *, created: bool, **kwargs) -> None: # noqa: ANN003, ARG001
|
||||
def notify_when_new_user(sender: User, instance: User, *, created: bool, **kwargs) -> None: # noqa: ANN003
|
||||
"""Send a Discord notification when a new user is created.
|
||||
|
||||
Args:
|
||||
|
|
@ -33,5 +34,6 @@ def notify_when_new_user(sender: User, instance: User, *, created: bool, **kwarg
|
|||
|
||||
msg: str = f"New user registered on FeedVault 👀: {instance.username}"
|
||||
webhook = DiscordWebhook(url=webhook_url, content=msg)
|
||||
response: Response = webhook.execute()
|
||||
logger.info("Discord notification sent: (%s) %s", response.status_code, response.text)
|
||||
if not settings.TESTING:
|
||||
response: Response = webhook.execute()
|
||||
logger.info("Discord notification sent: (%s) %s", response.status_code, response.text)
|
||||
|
|
|
|||
|
|
@ -3,10 +3,11 @@ from __future__ import annotations
|
|||
from typing import TYPE_CHECKING
|
||||
|
||||
from django.contrib.auth.models import User
|
||||
from django.test import TestCase
|
||||
from django.http.response import HttpResponse
|
||||
from django.test import Client, TestCase
|
||||
from django.urls import reverse
|
||||
|
||||
from feedvault.models import Domain, Feed
|
||||
from feedvault.models import Domain, Entry, Feed
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from django.http import HttpResponse
|
||||
|
|
@ -98,47 +99,62 @@ class TestDomains(TestCase):
|
|||
class TestAPI(TestCase):
|
||||
def test_api_page(self) -> None:
|
||||
"""Test if the API page is accessible."""
|
||||
response: HttpResponse = self.client.get(reverse("api"))
|
||||
response: HttpResponse = self.client.get(reverse("api_v1:openapi-view"))
|
||||
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
||||
|
||||
|
||||
class TestAPIFeeds(TestCase):
|
||||
def test_api_feeds_page(self) -> None:
|
||||
"""Test if the API feeds page is accessible."""
|
||||
response: HttpResponse = self.client.get(reverse("api_feeds"))
|
||||
response: HttpResponse = self.client.get(reverse("api_v1:list_feeds"))
|
||||
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
||||
|
||||
|
||||
class TestAPIFeed(TestCase):
|
||||
class FeedVaultAPITests(TestCase):
|
||||
def setUp(self) -> None:
|
||||
"""Create a test feed."""
|
||||
self.domain: Domain = Domain.objects.create(
|
||||
name="feedvault",
|
||||
url="feedvault.se",
|
||||
)
|
||||
# Set up data for the whole TestCase
|
||||
self.client = Client()
|
||||
|
||||
self.user: User = User.objects.create_user(
|
||||
username="testuser",
|
||||
email="hello@feedvault.se",
|
||||
password="testpassword", # noqa: S106
|
||||
)
|
||||
# Creating a domain instance
|
||||
self.domain: Domain = Domain.objects.create(name="Example Domain")
|
||||
|
||||
self.feed: Feed = Feed.objects.create(
|
||||
user=self.user,
|
||||
bozo=False,
|
||||
feed_url="https://feedvault.se/feed.xml",
|
||||
domain=self.domain,
|
||||
)
|
||||
# Creating a feed instance
|
||||
self.feed: Feed = Feed.objects.create(title="Example Feed", domain=self.domain, bozo=False)
|
||||
|
||||
def test_api_feed_page(self) -> None:
|
||||
"""Test if the API feed page is accessible."""
|
||||
response: HttpResponse = self.client.get(reverse("api_feeds_id", kwargs={"feed_id": 1}))
|
||||
# Creating entry instances
|
||||
self.entry1: Entry = Entry.objects.create(title="Example Entry 1", feed=self.feed)
|
||||
self.entry2: Entry = Entry.objects.create(title="Example Entry 2", feed=self.feed)
|
||||
|
||||
def test_list_feeds(self) -> None:
|
||||
response: HttpResponse = self.client.get("/api/v1/feeds/")
|
||||
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
||||
assert "Example Feed" in response.content.decode()
|
||||
|
||||
def test_api_feed_page_not_found(self) -> None:
|
||||
"""Test if the API feed page is accessible."""
|
||||
response: HttpResponse = self.client.get(reverse("api_feeds_id", kwargs={"feed_id": 2}))
|
||||
assert response.status_code == 404, f"Expected 404, got {response.status_code}"
|
||||
def test_get_feed(self) -> None:
|
||||
response: HttpResponse = self.client.get(f"/api/v1/feeds/{self.feed.pk}/")
|
||||
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
||||
assert "Example Feed" in response.content.decode()
|
||||
|
||||
def test_list_entries(self) -> None:
|
||||
response: HttpResponse = self.client.get(f"/api/v1/feeds/{self.feed.pk}/entries/")
|
||||
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
||||
assert "Example Entry 1" in response.content.decode()
|
||||
assert "Example Entry 2" in response.content.decode()
|
||||
|
||||
def test_get_entry(self) -> None:
|
||||
response: HttpResponse = self.client.get(f"/api/v1/entries/{self.entry1.pk}/")
|
||||
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
||||
assert "Example Entry 1" in response.content.decode()
|
||||
|
||||
def test_list_domains(self) -> None:
|
||||
response: HttpResponse = self.client.get("/api/v1/domains/")
|
||||
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
||||
assert "Example Domain" in response.content.decode()
|
||||
|
||||
def test_get_domain(self) -> None:
|
||||
response: HttpResponse = self.client.get(f"/api/v1/domains/{self.domain.pk}/")
|
||||
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
|
||||
assert "Example Domain" in response.content.decode()
|
||||
|
||||
|
||||
class TestAccount(TestCase):
|
||||
|
|
|
|||
|
|
@ -6,12 +6,14 @@ from django.urls import URLPattern, path
|
|||
from django.views.decorators.cache import cache_page
|
||||
|
||||
from feedvault import views
|
||||
from feedvault.api import api_v1
|
||||
from feedvault.models import Domain, Feed
|
||||
from feedvault.sitemaps import StaticViewSitemap
|
||||
from feedvault.views import APIView, CustomLoginView, CustomLogoutView, ProfileView, RegisterView
|
||||
from feedvault.views import CustomLoginView, CustomLogoutView, ProfileView, RegisterView
|
||||
|
||||
app_name: str = "feedvault"
|
||||
|
||||
|
||||
sitemaps = {
|
||||
"static": StaticViewSitemap,
|
||||
"feeds": GenericSitemap({"queryset": Feed.objects.all(), "date_field": "created_at"}),
|
||||
|
|
@ -34,12 +36,7 @@ urlpatterns: list[URLPattern] = [
|
|||
),
|
||||
path(route="domains/", view=views.DomainsView.as_view(), name="domains"),
|
||||
path(route="domain/<int:domain_id>/", view=views.DomainView.as_view(), name="domain"),
|
||||
path(route="api/", view=APIView.as_view(), name="api"),
|
||||
path(route="api/feeds/", view=views.APIFeedsView.as_view(), name="api_feeds"),
|
||||
path(route="api/feeds/<int:feed_id>/", view=views.APIFeedView.as_view(), name="api_feeds_id"),
|
||||
path(route="api/feeds/<int:feed_id>/entries/", view=views.APIFeedEntriesView.as_view(), name="api_feed_entries"),
|
||||
path(route="api/entries/", view=views.APIEntriesView.as_view(), name="api_entries"),
|
||||
path(route="api/entries/<int:entry_id>/", view=views.APIEntryView.as_view(), name="api_entries_id"),
|
||||
path("api/v1/", api_v1.urls), # type: ignore # noqa: PGH003
|
||||
path(route="accounts/login/", view=CustomLoginView.as_view(), name="login"),
|
||||
path(route="accounts/register/", view=RegisterView.as_view(), name="register"),
|
||||
path(route="accounts/logout/", view=CustomLogoutView.as_view(), name="logout"),
|
||||
|
|
|
|||
|
|
@ -7,9 +7,8 @@ from django.contrib.auth import login
|
|||
from django.contrib.auth.forms import AuthenticationForm, UserCreationForm
|
||||
from django.contrib.auth.views import LoginView, LogoutView, PasswordChangeView
|
||||
from django.contrib.messages.views import SuccessMessageMixin
|
||||
from django.core.paginator import EmptyPage, Page, PageNotAnInteger, Paginator
|
||||
from django.forms.models import model_to_dict
|
||||
from django.http import HttpRequest, HttpResponse, JsonResponse
|
||||
from django.db.models.manager import BaseManager
|
||||
from django.http import HttpRequest, HttpResponse
|
||||
from django.shortcuts import get_object_or_404, render
|
||||
from django.template import loader
|
||||
from django.urls import reverse_lazy
|
||||
|
|
@ -22,6 +21,8 @@ from feedvault.models import Domain, Entry, Feed
|
|||
|
||||
if TYPE_CHECKING:
|
||||
from django.contrib.auth.models import User
|
||||
from django.core.files.uploadedfile import UploadedFile
|
||||
from django.db.models.manager import BaseManager
|
||||
|
||||
|
||||
class IndexView(View):
|
||||
|
|
@ -49,8 +50,8 @@ class FeedView(View):
|
|||
if not feed_id:
|
||||
return HttpResponse(content="No id", status=400)
|
||||
|
||||
feed = get_object_or_404(Feed, id=feed_id)
|
||||
entries = Entry.objects.filter(feed=feed).order_by("-created_parsed")[:100]
|
||||
feed: Feed = get_object_or_404(Feed, id=feed_id)
|
||||
entries: BaseManager[Entry] = Entry.objects.filter(feed=feed).order_by("-created_parsed")[:100]
|
||||
|
||||
context = {
|
||||
"feed": feed,
|
||||
|
|
@ -91,7 +92,7 @@ class AddView(View):
|
|||
def get(self, request: HttpRequest) -> HttpResponse:
|
||||
"""Load the index page."""
|
||||
template = loader.get_template(template_name="index.html")
|
||||
context = {
|
||||
context: dict[str, str] = {
|
||||
"description": "FeedVault allows users to archive and search their favorite web feeds.",
|
||||
"keywords": "feed, rss, atom, archive, rss list",
|
||||
"author": "TheLovinator",
|
||||
|
|
@ -134,7 +135,7 @@ class UploadView(View):
|
|||
def get(self, request: HttpRequest) -> HttpResponse:
|
||||
"""Load the index page."""
|
||||
template = loader.get_template(template_name="index.html")
|
||||
context = {
|
||||
context: dict[str, str] = {
|
||||
"description": "FeedVault allows users to archive and search their favorite web feeds.",
|
||||
"keywords": "feed, rss, atom, archive, rss list",
|
||||
"author": "TheLovinator",
|
||||
|
|
@ -150,7 +151,7 @@ class UploadView(View):
|
|||
if not request.user.is_active:
|
||||
return HttpResponse(content="User is not active", status=403)
|
||||
|
||||
file = request.FILES.get("file", None)
|
||||
file: UploadedFile | None = request.FILES.get("file", None)
|
||||
if not file:
|
||||
return HttpResponse(content="No file", status=400)
|
||||
|
||||
|
|
@ -193,7 +194,7 @@ class RegisterView(CreateView):
|
|||
# Add context data to the view
|
||||
def get_context_data(self, **kwargs) -> dict: # noqa: ANN003
|
||||
"""Get the context data."""
|
||||
context = super().get_context_data(**kwargs)
|
||||
context: dict[str, Any] = super().get_context_data(**kwargs)
|
||||
context["description"] = "Register a new account"
|
||||
context["keywords"] = "register, account, feed, rss, atom, archive, rss list"
|
||||
context["author"] = "TheLovinator"
|
||||
|
|
@ -234,7 +235,7 @@ class ProfileView(View):
|
|||
"""Load the profile page."""
|
||||
template = loader.get_template(template_name="accounts/profile.html")
|
||||
|
||||
user_feeds = Feed.objects.filter(user=request.user).order_by("-created_at")[:100]
|
||||
user_feeds: BaseManager[Feed] = Feed.objects.filter(user=request.user).order_by("-created_at")[:100]
|
||||
|
||||
context: dict[str, str | Any] = {
|
||||
"description": f"Profile page for {request.user.get_username()}",
|
||||
|
|
@ -247,22 +248,6 @@ class ProfileView(View):
|
|||
return HttpResponse(content=template.render(context=context, request=request))
|
||||
|
||||
|
||||
class APIView(View):
|
||||
"""API documentation page."""
|
||||
|
||||
def get(self, request: HttpRequest) -> HttpResponse:
|
||||
"""Load the API page."""
|
||||
template = loader.get_template(template_name="api.html")
|
||||
context = {
|
||||
"description": "FeedVault allows users to archive and search their favorite web feeds.",
|
||||
"keywords": "feed, rss, atom, archive, rss list",
|
||||
"author": "TheLovinator",
|
||||
"canonical": "https://feedvault.se/api/",
|
||||
"title": "API Documentation",
|
||||
}
|
||||
return HttpResponse(content=template.render(context=context, request=request))
|
||||
|
||||
|
||||
class RobotsView(View):
|
||||
"""Robots.txt view."""
|
||||
|
||||
|
|
@ -274,186 +259,12 @@ class RobotsView(View):
|
|||
)
|
||||
|
||||
|
||||
class APIFeedsView(View):
|
||||
"""API Feeds view."""
|
||||
|
||||
def get(self, request: HttpRequest) -> HttpResponse:
|
||||
"""Get all feeds with pagination."""
|
||||
# Retrieve all feeds
|
||||
feeds_list = Feed.objects.all()
|
||||
|
||||
# Pagination settings
|
||||
page: int = int(request.GET.get("page", 1)) # Get the page number from the query parameters, default to 1
|
||||
per_page: int = int(request.GET.get("per_page", 1000)) # Number of feeds per page, default to 1000 (max 1000)
|
||||
|
||||
# Add a ceiling to the per_page value
|
||||
max_per_page = 1000
|
||||
if per_page > max_per_page:
|
||||
per_page = max_per_page
|
||||
|
||||
# Create Paginator instance
|
||||
paginator = Paginator(feeds_list, per_page)
|
||||
|
||||
try:
|
||||
feeds: Page = paginator.page(page)
|
||||
except PageNotAnInteger:
|
||||
# If page is not an integer, deliver first page.
|
||||
feeds = paginator.page(1)
|
||||
except EmptyPage:
|
||||
# If page is out of range (e.g., 9999), deliver last page of results.
|
||||
feeds = paginator.page(paginator.num_pages)
|
||||
|
||||
# Convert feeds to dictionary
|
||||
feeds_dict = [model_to_dict(feed) for feed in feeds]
|
||||
|
||||
# Return the paginated entries as JsonResponse
|
||||
response = JsonResponse(feeds_dict, safe=False)
|
||||
|
||||
# Add pagination headers
|
||||
response["X-Page"] = feeds.number
|
||||
response["X-Page-Count"] = paginator.num_pages
|
||||
response["X-Per-Page"] = per_page
|
||||
response["X-Total-Count"] = paginator.count
|
||||
response["X-First-Page"] = 1
|
||||
response["X-Last-Page"] = paginator.num_pages
|
||||
|
||||
# Next and previous page links
|
||||
if feeds.has_next():
|
||||
response["X-Next-Page"] = feeds.next_page_number()
|
||||
if feeds.has_previous():
|
||||
response["X-Prev-Page"] = feeds.previous_page_number()
|
||||
|
||||
return response
|
||||
|
||||
|
||||
class APIFeedView(View):
|
||||
"""API Feed view."""
|
||||
|
||||
def get(self, request: HttpRequest, feed_id: int) -> HttpResponse: # noqa: ARG002
|
||||
"""Get a single feed."""
|
||||
feed = get_object_or_404(Feed, id=feed_id)
|
||||
return JsonResponse(model_to_dict(feed), safe=False)
|
||||
|
||||
|
||||
class APIEntriesView(View):
|
||||
"""API Entries view."""
|
||||
|
||||
def get(self: APIEntriesView, request: HttpRequest) -> HttpResponse:
|
||||
"""Get all entries with pagination."""
|
||||
# Retrieve all entries
|
||||
entries_list = Entry.objects.all()
|
||||
|
||||
# Pagination settings
|
||||
page: int = int(request.GET.get("page", 1)) # Get the page number from the query parameters, default to 1
|
||||
per_page: int = int(request.GET.get("per_page", 1000))
|
||||
|
||||
# Add a ceiling to the per_page value
|
||||
max_per_page = 1000
|
||||
if per_page > max_per_page:
|
||||
per_page = max_per_page
|
||||
|
||||
# Create Paginator instance
|
||||
paginator = Paginator(entries_list, per_page)
|
||||
|
||||
try:
|
||||
entries: Page = paginator.page(page)
|
||||
except PageNotAnInteger:
|
||||
# If page is not an integer, deliver first page.
|
||||
entries = paginator.page(1)
|
||||
except EmptyPage:
|
||||
# If page is out of range (e.g. 9999), deliver last page of results.
|
||||
entries = paginator.page(paginator.num_pages)
|
||||
|
||||
# Convert entries to dictionary
|
||||
entries_dict = [model_to_dict(entry) for entry in entries]
|
||||
|
||||
# Return the paginated entries as JsonResponse
|
||||
response = JsonResponse(entries_dict, safe=False)
|
||||
|
||||
# Add pagination headers
|
||||
response["X-Page"] = entries.number
|
||||
response["X-Page-Count"] = paginator.num_pages
|
||||
response["X-Per-Page"] = per_page
|
||||
response["X-Total-Count"] = paginator.count
|
||||
response["X-First-Page"] = 1
|
||||
response["X-Last-Page"] = paginator.num_pages
|
||||
|
||||
# Next and previous page links
|
||||
if entries.has_next():
|
||||
response["X-Next-Page"] = entries.next_page_number()
|
||||
if entries.has_previous():
|
||||
response["X-Prev-Page"] = entries.previous_page_number()
|
||||
|
||||
return response
|
||||
|
||||
|
||||
class APIEntryView(View):
|
||||
"""API Entry view."""
|
||||
|
||||
def get(self: APIEntryView, request: HttpRequest, entry_id: int) -> HttpResponse: # noqa: ARG002
|
||||
"""Get a single entry."""
|
||||
entry = get_object_or_404(Entry, id=entry_id)
|
||||
return JsonResponse(model_to_dict(entry), safe=False)
|
||||
|
||||
|
||||
class APIFeedEntriesView(View):
|
||||
"""API Feed Entries view."""
|
||||
|
||||
def get(self: APIFeedEntriesView, request: HttpRequest, feed_id: int) -> HttpResponse:
|
||||
"""Get all entries for a single feed with pagination."""
|
||||
# Retrieve all entries for a single feed
|
||||
entries_list = Entry.objects.filter(feed_id=feed_id)
|
||||
|
||||
# Pagination settings
|
||||
page: int = int(request.GET.get("page", 1)) # Get the page number from the query parameters, default to 1
|
||||
per_page: int = int(request.GET.get("per_page", 1000))
|
||||
|
||||
# Add a ceiling to the per_page value
|
||||
max_per_page = 1000
|
||||
if per_page > max_per_page:
|
||||
per_page = max_per_page
|
||||
|
||||
# Create Paginator instance
|
||||
paginator = Paginator(entries_list, per_page)
|
||||
|
||||
try:
|
||||
entries: Page = paginator.page(page)
|
||||
except PageNotAnInteger:
|
||||
# If page is not an integer, deliver first page.
|
||||
entries = paginator.page(1)
|
||||
except EmptyPage:
|
||||
# If page is out of range (e.g. 9999), deliver last page of results.
|
||||
entries = paginator.page(paginator.num_pages)
|
||||
|
||||
# Convert entries to dictionary
|
||||
entries_dict = [model_to_dict(entry) for entry in entries]
|
||||
|
||||
# Return the paginated entries as JsonResponse
|
||||
response = JsonResponse(entries_dict, safe=False)
|
||||
|
||||
# Add pagination headers
|
||||
response["X-Page"] = entries.number
|
||||
response["X-Page-Count"] = paginator.num_pages
|
||||
response["X-Per-Page"] = per_page
|
||||
response["X-Total-Count"] = paginator.count
|
||||
response["X-First-Page"] = 1
|
||||
response["X-Last-Page"] = paginator.num_pages
|
||||
|
||||
# Next and previous page links
|
||||
if entries.has_next():
|
||||
response["X-Next-Page"] = entries.next_page_number()
|
||||
if entries.has_previous():
|
||||
response["X-Prev-Page"] = entries.previous_page_number()
|
||||
|
||||
return response
|
||||
|
||||
|
||||
class DomainsView(View):
|
||||
"""All domains."""
|
||||
|
||||
def get(self: DomainsView, request: HttpRequest) -> HttpResponse:
|
||||
"""Load the domains page."""
|
||||
domains = Domain.objects.all()
|
||||
domains: BaseManager[Domain] = Domain.objects.all()
|
||||
template = loader.get_template(template_name="domains.html")
|
||||
context = {
|
||||
"domains": domains,
|
||||
|
|
@ -471,8 +282,8 @@ class DomainView(View):
|
|||
|
||||
def get(self: DomainView, request: HttpRequest, domain_id: int) -> HttpResponse:
|
||||
"""Load the domain page."""
|
||||
domain = get_object_or_404(Domain, id=domain_id)
|
||||
feeds = Feed.objects.filter(domain=domain).order_by("-created_at")[:100]
|
||||
domain: Domain = get_object_or_404(Domain, id=domain_id)
|
||||
feeds: BaseManager[Feed] = Feed.objects.filter(domain=domain).order_by("-created_at")[:100]
|
||||
|
||||
context = {
|
||||
"domain": domain,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue