aboutsummaryrefslogtreecommitdiff
path: root/webmentions_ssg/models.py
diff options
context:
space:
mode:
authorDennis Fink2026-08-09 14:15:29 +0200
committerDennis Fink2026-08-09 14:15:29 +0200
commite4a8194e6de25e8a48e2c61ce7f89a8159fd99d9 (patch)
tree234f63980aba8fe3c8d03e7bc8959598bce7e790 /webmentions_ssg/models.py
parentf53c1136184d2afabfb1e5974e527229aa71939a (diff)
downloadwebmentions-ssg-e4a8194e6de25e8a48e2c61ce7f89a8159fd99d9.tar.gz
webmentions-ssg-e4a8194e6de25e8a48e2c61ce7f89a8159fd99d9.zip
Implement received Webmention handling
Add the Flask application setup, database models and migrations, authentication, and configuration for development and testing. Implement asynchronous Webmention verification with Huey, including HTML and plain-text source validation, retries, status tracking, and size limits. Add status, login, and paginated received-Webmention views together with comprehensive tests for forms, views, and receiver tasks.
Diffstat (limited to '')
-rw-r--r--webmentions_ssg/models.py89
1 files changed, 89 insertions, 0 deletions
diff --git a/webmentions_ssg/models.py b/webmentions_ssg/models.py
new file mode 100644
index 0000000..8c378aa
--- /dev/null
+++ b/webmentions_ssg/models.py
@@ -0,0 +1,89 @@
+import uuid
+from datetime import datetime, timezone
+from typing import Optional
+
+from flask_login import UserMixin
+from sqlalchemy import DateTime, String, Text, UniqueConstraint, Uuid
+from sqlalchemy.orm import Mapped, mapped_column
+from werkzeug.security import check_password_hash, generate_password_hash
+
+from . import Base
+
+
+class User(UserMixin, Base):
+ __tablename__ = "users"
+
+ id: Mapped[int] = mapped_column(primary_key=True)
+ username: Mapped[str] = mapped_column(
+ String(64),
+ index=True,
+ unique=True,
+ )
+ password_hash: Mapped[Optional[str]] = mapped_column(String(256))
+
+ def __repr__(self):
+ return f"<User {self.username}>"
+
+ def set_password(self, password):
+ self.password_hash = generate_password_hash(password)
+
+ def check_password(self, password):
+ return check_password_hash(
+ self.password_hash,
+ password,
+ )
+
+
+class ReceivedWebmention(Base):
+ __tablename__ = "received_webmentions"
+
+ uuid: Mapped[uuid.UUID] = mapped_column(
+ Uuid(as_uuid=True),
+ primary_key=True,
+ )
+
+ source: Mapped[str] = mapped_column(
+ Text(),
+ nullable=False,
+ )
+ target: Mapped[str] = mapped_column(
+ Text(),
+ nullable=False,
+ )
+
+ status: Mapped[str] = mapped_column(
+ Text(),
+ nullable=False,
+ default="received",
+ )
+ failure_reason: Mapped[str | None] = mapped_column(
+ Text(),
+ nullable=True,
+ default=None,
+ )
+
+ updated_at: Mapped[datetime] = mapped_column(
+ DateTime(timezone=True),
+ default=lambda: datetime.now(timezone.utc),
+ onupdate=lambda: datetime.now(timezone.utc),
+ nullable=False,
+ )
+
+ __table_args__ = (
+ UniqueConstraint(
+ "source",
+ "target",
+ name="uq_webmention_source_target",
+ ),
+ )
+
+ @property
+ def verified(self) -> bool:
+ return self.status == "verified"
+
+ @property
+ def created_at(self) -> datetime:
+ return datetime.fromtimestamp(
+ self.uuid.time / 1000,
+ tz=timezone.utc,
+ )