from __future__ import annotations
from typing import Any
from django.conf import settings
from django.core.exceptions import ValidationError
from django.core.validators import EmailValidator, URLValidator
from django.db import models
from django.utils import timezone
from django.utils.crypto import get_random_string
[docs]
class GenKeyMixin(models.Model):
"""Mixin that automatically generates a random key on save if not provided."""
key: models.CharField[str, str]
[docs]
def save(self, *args: Any, **kwargs: Any) -> None:
if not self.key:
self.key = get_random_string(length=32)
super().save(*args, **kwargs)
[docs]
class Auth(GenKeyMixin):
"""Stores authorization grants during the IndieAuth flow."""
created = models.DateTimeField(auto_now_add=True)
modified = models.DateTimeField(auto_now=True)
"""
Model for storing IndieAuth authorization codes.
Used during the IndieAuth flow to temporarily store authorization details
before exchanging the auth code for an access token.
"""
key = models.CharField(max_length=32)
owner = models.ForeignKey(settings.AUTH_USER_MODEL, related_name="indieweb_auth", on_delete=models.CASCADE)
state = models.CharField(max_length=32)
client_id = models.CharField(max_length=512)
redirect_uri = models.CharField(max_length=1024)
scope = models.CharField(max_length=256, null=True, blank=True) # noqa
me = models.CharField(max_length=512)
code_challenge = models.CharField(max_length=128, null=True, blank=True) # noqa: DJ001
code_challenge_method = models.CharField(max_length=8, null=True, blank=True) # noqa: DJ001
class Meta:
unique_together = ("me", "client_id", "scope", "owner")
def __str__(self) -> str:
return f"{self.client_id} {self.me} {self.scope} {self.owner.username}"
[docs]
class Token(GenKeyMixin):
"""Stores access tokens for authenticated API access."""
created = models.DateTimeField(auto_now_add=True)
modified = models.DateTimeField(auto_now=True)
"""
Model for storing IndieAuth/Micropub access tokens.
Represents long-lived access tokens that clients can use to authenticate
requests to the Micropub endpoint and other IndieWeb services.
"""
key = models.CharField(max_length=32, db_index=True)
owner = models.ForeignKey(
settings.AUTH_USER_MODEL,
related_name="indieweb_token",
on_delete=models.CASCADE,
)
client_id = models.CharField(max_length=512)
me = models.CharField(max_length=512)
scope = models.CharField(max_length=256, null=True, blank=True) # noqa
expires_at = models.DateTimeField(null=True, blank=True, db_index=True)
class Meta:
unique_together = ("me", "client_id", "scope", "owner")
def __str__(self) -> str:
return f"{self.client_id} {self.me} {self.scope} {self.owner.username}"
[docs]
def is_expired(self) -> bool:
"""Return True if the token has an expires_at in the past.
Tokens with ``expires_at`` set to ``None`` are treated as
non-expiring for backwards compatibility with rows created
before expiration tracking was added.
"""
if self.expires_at is None:
return False
return self.expires_at <= timezone.now()
[docs]
class Webmention(models.Model):
"""
Model for storing webmentions.
Webmentions are a W3C recommendation for notifying when one site
mentions another, enabling cross-site conversations.
"""
STATUS_CHOICES = [
("pending", "Pending"),
("verified", "Verified"),
("failed", "Failed"),
("spam", "Spam"),
]
MENTION_TYPE_CHOICES = [
("mention", "Mention"),
("like", "Like"),
("reply", "Reply"),
("repost", "Repost"),
]
# Core webmention fields
source_url = models.URLField(max_length=500, db_index=True)
target_url = models.URLField(max_length=500, db_index=True)
# Status tracking
status = models.CharField(max_length=20, choices=STATUS_CHOICES, default="pending")
# Parsed content from microformats2
author_name = models.CharField(max_length=200, blank=True)
author_url = models.URLField(blank=True)
author_photo = models.URLField(blank=True)
content = models.TextField(blank=True)
content_html = models.TextField(blank=True)
published = models.DateTimeField(null=True, blank=True)
# Webmention type
mention_type = models.CharField(max_length=20, choices=MENTION_TYPE_CHOICES, default="mention")
# Tracking
created = models.DateTimeField(auto_now_add=True)
modified = models.DateTimeField(auto_now=True)
verified_at = models.DateTimeField(null=True, blank=True)
# Optional spam check result
spam_check_result = models.JSONField(null=True, blank=True)
class Meta:
unique_together = ["source_url", "target_url"]
indexes = [
models.Index(fields=["target_url", "status"]),
models.Index(fields=["created"]),
]
def __str__(self) -> str:
return f"{self.mention_type}: {self.source_url} -> {self.target_url}"
[docs]
class Profile(models.Model):
"""User profile with h-card data stored as JSON."""
user = models.OneToOneField(settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="indieweb_profile")
h_card = models.JSONField(default=dict, blank=True)
# Common fields for quick access/querying
name = models.CharField(max_length=200, blank=True)
photo_url = models.URLField(blank=True)
url = models.URLField(blank=True)
created = models.DateTimeField(auto_now_add=True)
modified = models.DateTimeField(auto_now=True)
class Meta:
db_table = "indieweb_profile"
verbose_name = "User Profile"
verbose_name_plural = "User Profiles"
def __str__(self) -> str:
return f"Profile for {self.user.username}"
[docs]
def save(self, *args: Any, **kwargs: Any) -> None:
"""Save profile and sync quick-access fields with h_card data."""
# Sync fields before saving
self._sync_fields_from_h_card()
# Validate
self.full_clean()
super().save(*args, **kwargs)
[docs]
def clean(self) -> None:
"""Validate h_card data before saving."""
super().clean()
if self.h_card:
self._validate_h_card_urls()
self._validate_h_card_emails()
def _validate_h_card_urls(self) -> None:
"""Validate all URLs in h_card data."""
url_validator = URLValidator()
for field in ("url", "photo"):
if field in self.h_card:
for url in self.h_card[field]:
self._validate_single_url(url_validator, url, f"h_card.{field}")
if "org" in self.h_card:
for org in self.h_card["org"]:
if isinstance(org, dict) and "url" in org:
self._validate_single_url(url_validator, org["url"], "h_card.org.url")
@staticmethod
def _validate_single_url(validator: URLValidator, url: str | dict[str, Any], context: str) -> None:
"""Validate a single URL value (string or dict with 'value' key)."""
if isinstance(url, dict) and "value" in url:
url_to_validate = url["value"]
elif isinstance(url, str):
url_to_validate = url
else:
return
try:
validator(url_to_validate)
except ValidationError as e:
raise ValidationError(f"Invalid URL in {context}: {url_to_validate}") from e
def _validate_h_card_emails(self) -> None:
"""Validate all emails in h_card data."""
email_validator = EmailValidator()
if "email" in self.h_card:
for email in self.h_card["email"]:
try:
email_validator(email)
except ValidationError as e:
raise ValidationError(f"Invalid email in h_card: {email}") from e
def _sync_fields_from_h_card(self) -> None:
"""Sync quick-access fields with h_card data."""
if not self.h_card:
return
# Sync name
if "name" in self.h_card and self.h_card["name"]:
self.name = self.h_card["name"][0]
# Sync URL
if "url" in self.h_card and self.h_card["url"]:
self.url = self.h_card["url"][0]
# Sync photo URL
if "photo" in self.h_card and self.h_card["photo"]:
photo = self.h_card["photo"][0]
if isinstance(photo, dict) and "value" in photo:
self.photo_url = photo["value"]
elif isinstance(photo, str):
self.photo_url = photo