feat: custom newsletter slug

This commit is contained in:
Leon
2025-07-24 13:20:22 +02:00
parent f78607f506
commit 24e65a8c86
19 changed files with 386 additions and 68 deletions

View File

@@ -0,0 +1,87 @@
"""add_slug_to_newsletter
Revision ID: 1ed7baeaf282
Revises: ce35472309a4
Create Date: 2025-07-24 12:32:05.618379
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '1ed7baeaf282'
down_revision: Union[str, Sequence[str], None] = 'ce35472309a4'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Upgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.drop_index(op.f('ix_senders_email'), table_name='senders')
op.drop_index(op.f('ix_senders_id'), table_name='senders')
op.drop_table('senders')
op.drop_index(op.f('ix_newsletters_id'), table_name='newsletters')
op.drop_table('newsletters')
op.drop_index(op.f('ix_settings_id'), table_name='settings')
op.drop_index(op.f('ix_settings_imap_server'), table_name='settings')
op.drop_table('settings')
op.drop_index(op.f('ix_entries_id'), table_name='entries')
op.drop_index(op.f('ix_entries_message_id'), table_name='entries')
op.drop_table('entries')
# ### end Alembic commands ###
def downgrade() -> None:
"""Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('entries',
sa.Column('id', sa.VARCHAR(), nullable=False),
sa.Column('newsletter_id', sa.VARCHAR(), nullable=True),
sa.Column('subject', sa.VARCHAR(), nullable=True),
sa.Column('body', sa.TEXT(), nullable=True),
sa.Column('received_at', sa.DATETIME(), nullable=True),
sa.Column('message_id', sa.VARCHAR(), nullable=False),
sa.ForeignKeyConstraint(['newsletter_id'], ['newsletters.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_entries_message_id'), 'entries', ['message_id'], unique=1)
op.create_index(op.f('ix_entries_id'), 'entries', ['id'], unique=False)
op.create_table('settings',
sa.Column('id', sa.INTEGER(), nullable=False),
sa.Column('imap_server', sa.VARCHAR(), nullable=True),
sa.Column('imap_username', sa.VARCHAR(), nullable=True),
sa.Column('imap_password', sa.VARCHAR(), nullable=True),
sa.Column('search_folder', sa.VARCHAR(), nullable=True),
sa.Column('move_to_folder', sa.VARCHAR(), nullable=True),
sa.Column('mark_as_read', sa.BOOLEAN(), nullable=True),
sa.Column('email_check_interval', sa.INTEGER(), nullable=True),
sa.Column('auto_add_new_senders', sa.BOOLEAN(), nullable=True),
sa.Column('auth_username', sa.VARCHAR(), nullable=True),
sa.Column('auth_password_hash', sa.VARCHAR(), nullable=True),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_settings_imap_server'), 'settings', ['imap_server'], unique=False)
op.create_index(op.f('ix_settings_id'), 'settings', ['id'], unique=False)
op.create_table('newsletters',
sa.Column('id', sa.VARCHAR(), nullable=False),
sa.Column('name', sa.VARCHAR(), nullable=True),
sa.Column('move_to_folder', sa.VARCHAR(), nullable=True),
sa.Column('is_active', sa.BOOLEAN(), nullable=True),
sa.Column('extract_content', sa.BOOLEAN(), nullable=True),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_newsletters_id'), 'newsletters', ['id'], unique=False)
op.create_table('senders',
sa.Column('id', sa.VARCHAR(), nullable=False),
sa.Column('email', sa.VARCHAR(), nullable=False),
sa.Column('newsletter_id', sa.VARCHAR(), nullable=False),
sa.ForeignKeyConstraint(['newsletter_id'], ['newsletters.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_senders_id'), 'senders', ['id'], unique=False)
op.create_index(op.f('ix_senders_email'), 'senders', ['email'], unique=1)
# ### end Alembic commands ###

18
backend/app/core/slug.py Normal file
View File

@@ -0,0 +1,18 @@
import re
def sanitize_slug(slug: str | None) -> str | None:
"""Sanitize a string to be used as a URL slug.
- Converts to lowercase
- Replaces spaces and underscores with hyphens
- Removes characters that are not alphanumeric or hyphens
- Removes leading/trailing hyphens
"""
if not slug:
return None
slug = slug.lower()
slug = re.sub(r"[\s_]+", "-", slug)
slug = re.sub(r"[^a-z0-9-]", "", slug)
slug = slug.strip("-")
return slug or None

View File

@@ -1,5 +1,5 @@
from nanoid import generate
from sqlalchemy import func
from sqlalchemy import func, or_
from sqlalchemy.orm import Session
from app.core.logging import get_logger
@@ -10,13 +10,13 @@ from app.schemas.newsletters import NewsletterCreate, NewsletterUpdate
logger = get_logger(__name__)
def get_newsletter(db: Session, newsletter_id: str):
"""Retrieve a single newsletter by its ID."""
logger.debug(f"Querying for newsletter with id={newsletter_id}")
def get_newsletter_by_identifier(db: Session, identifier: str):
"""Retrieve a single newsletter by its ID or slug."""
logger.debug(f"Querying for newsletter with identifier={identifier}")
result = (
db.query(Newsletter, func.count(Entry.id))
.outerjoin(Entry, Newsletter.id == Entry.newsletter_id)
.filter(Newsletter.id == newsletter_id)
.filter(or_(Newsletter.id == identifier, Newsletter.slug == identifier))
.group_by(Newsletter.id)
.first()
)
@@ -27,6 +27,11 @@ def get_newsletter(db: Session, newsletter_id: str):
return None
def get_newsletter_by_slug(db: Session, slug: str):
"""Retrieve a newsletter by its slug."""
return db.query(Newsletter).filter(Newsletter.slug == slug).first()
def get_newsletters(db: Session, skip: int = 0, limit: int = 100):
"""Retrieve a list of newsletters."""
logger.debug(f"Querying for newsletters with skip={skip}, limit={limit}")
@@ -51,9 +56,14 @@ def get_newsletters(db: Session, skip: int = 0, limit: int = 100):
def create_newsletter(db: Session, newsletter: NewsletterCreate):
"""Create a new newsletter."""
logger.info(f"Creating new newsletter with name '{newsletter.name}'")
if newsletter.slug and get_newsletter_by_slug(db, newsletter.slug):
return None # Indicates a conflict
db_newsletter = Newsletter(
id=generate(size=10),
name=newsletter.name,
slug=newsletter.slug,
extract_content=newsletter.extract_content,
move_to_folder=newsletter.move_to_folder,
)
@@ -82,9 +92,16 @@ def update_newsletter(
if not db_newsletter:
return None
db_newsletter.name = newsletter_update.name
db_newsletter.move_to_folder = newsletter_update.move_to_folder
db_newsletter.extract_content = newsletter_update.extract_content
if newsletter_update.slug:
existing_newsletter = get_newsletter_by_slug(db, newsletter_update.slug)
if existing_newsletter and existing_newsletter.id != newsletter_id:
return "conflict" # Indicates a conflict
update_data = newsletter_update.model_dump(exclude_unset=True)
for key, value in update_data.items():
if key == "sender_emails":
continue
setattr(db_newsletter, key, value)
# More efficient sender update
existing_emails = {sender.email for sender in db_newsletter.senders}
@@ -107,13 +124,13 @@ def update_newsletter(
db.refresh(db_newsletter)
logger.info(f"Successfully updated newsletter with id={db_newsletter.id}")
return get_newsletter(db, newsletter_id)
return get_newsletter_by_identifier(db, newsletter_id)
def delete_newsletter(db: Session, newsletter_id: str):
"""Delete a newsletter by its ID."""
logger.info(f"Deleting newsletter with id={newsletter_id}")
db_newsletter = get_newsletter(db, newsletter_id)
db_newsletter = get_newsletter_by_identifier(db, newsletter_id)
if not db_newsletter:
return None

View File

@@ -10,6 +10,7 @@ class Newsletter(Base):
__tablename__ = "newsletters"
id = Column(String, primary_key=True, index=True)
slug = Column(String, unique=True, index=True, nullable=True)
name = Column(String)
move_to_folder = Column(String, nullable=True)
is_active = Column(Boolean, default=True)

View File

@@ -10,16 +10,18 @@ logger = get_logger(__name__)
router = APIRouter()
@router.get("/feeds/{newsletter_id}")
def get_newsletter_feed(newsletter_id: str, db: Session = Depends(get_db)):
@router.get("/feeds/{feed_identifier}")
def get_newsletter_feed(feed_identifier: str, db: Session = Depends(get_db)):
"""Generate an Atom feed for a specific newsletter."""
logger.info(f"Generating feed for newsletter_id={newsletter_id}")
feed = generate_feed(db, newsletter_id)
logger.info(f"Generating feed for newsletter with identifier={feed_identifier}")
feed = generate_feed(db, feed_identifier)
if not feed:
logger.warning(
f"Newsletter with id={newsletter_id} not found, cannot generate feed."
f"Newsletter with identifier={feed_identifier} not found, cannot generate feed."
)
raise HTTPException(status_code=404, detail="Newsletter not found")
logger.info(f"Successfully generated feed for newsletter_id={newsletter_id}")
logger.info(
f"Successfully generated feed for newsletter with identifier={feed_identifier}"
)
return Response(content=feed, media_type="application/atom+xml")

View File

@@ -9,7 +9,7 @@ from app.crud.entries import create_entry
from app.crud.newsletters import (
create_newsletter,
delete_newsletter,
get_newsletter,
get_newsletter_by_identifier,
get_newsletters,
update_newsletter,
)
@@ -26,7 +26,10 @@ def create_new_newsletter(newsletter: NewsletterCreate, db: Session = Depends(ge
logger.info(
f"Request to create new newsletter for senders {newsletter.sender_emails}"
)
return create_newsletter(db=db, newsletter=newsletter)
db_newsletter = create_newsletter(db=db, newsletter=newsletter)
if db_newsletter is None:
raise HTTPException(status_code=409, detail="Slug already in use")
return db_newsletter
@router.get("/newsletters", response_model=List[Newsletter])
@@ -41,7 +44,7 @@ def read_newsletters(skip: int = 0, limit: int = 100, db: Session = Depends(get_
def read_newsletter(newsletter_id: str, db: Session = Depends(get_db)):
"""Retrieve a single newsletter by its ID."""
logger.info(f"Request to read newsletter with id={newsletter_id}")
db_newsletter = get_newsletter(db, newsletter_id=newsletter_id)
db_newsletter = get_newsletter_by_identifier(db, identifier=newsletter_id)
if db_newsletter is None:
logger.warning(f"Newsletter with id={newsletter_id} not found")
raise HTTPException(status_code=404, detail="Newsletter not found")
@@ -60,6 +63,8 @@ def update_existing_newsletter(
if db_newsletter is None:
logger.warning(f"Newsletter with id={newsletter_id} not found, cannot update")
raise HTTPException(status_code=404, detail="Newsletter not found")
if db_newsletter == "conflict":
raise HTTPException(status_code=409, detail="Slug already in use")
return db_newsletter
@@ -80,7 +85,7 @@ def create_newsletter_entry(
):
"""Create a new entry for a specific newsletter."""
logger.info(f"Request to create entry for newsletter_id={newsletter_id}")
db_newsletter = get_newsletter(db, newsletter_id=newsletter_id)
db_newsletter = get_newsletter_by_identifier(db, identifier=newsletter_id)
if db_newsletter is None:
logger.warning(
f"Newsletter with id={newsletter_id} not found, cannot create entry"

View File

@@ -1,6 +1,8 @@
from typing import List
from pydantic import BaseModel, ConfigDict
from pydantic import BaseModel, ConfigDict, field_validator
from app.core.slug import sanitize_slug
class SenderBase(BaseModel):
@@ -28,9 +30,15 @@ class NewsletterBase(BaseModel):
"""Base schema for a newsletter."""
name: str
slug: str | None = None
move_to_folder: str | None = None
extract_content: bool = False
@field_validator("slug")
def sanitize_slug_field(cls, v: str | None) -> str | None:
"""Sanitize slug."""
return sanitize_slug(v)
class NewsletterCreate(NewsletterBase):
"""Schema for creating a new newsletter."""

View File

@@ -4,18 +4,18 @@ from sqlalchemy.orm import Session
from app.core.config import settings
from app.crud.entries import get_entries_by_newsletter
from app.crud.newsletters import get_newsletter
from app.crud.newsletters import get_newsletter_by_identifier
def generate_feed(db: Session, newsletter_id: str):
def generate_feed(db: Session, feed_identifier: str):
"""Generate an Atom feed for a given newsletter."""
newsletter = get_newsletter(db, newsletter_id)
newsletter = get_newsletter_by_identifier(db, feed_identifier)
if not newsletter:
return None
entries = get_entries_by_newsletter(db, newsletter_id)
entries = get_entries_by_newsletter(db, newsletter.id)
feed_url = f"{settings.app_base_url}/feeds/{newsletter_id}"
feed_url = f"{settings.app_base_url}/feeds/{newsletter.slug or newsletter.id}"
logo_url = f"{settings.app_base_url}/logo.png"
icon_url = f"{settings.app_base_url}/favicon.ico"

View File

@@ -4,7 +4,11 @@ from unittest.mock import patch
from sqlalchemy.orm import Session
from app.crud.entries import create_entry, get_entries_by_newsletter
from app.crud.newsletters import create_newsletter, get_newsletter, get_newsletters
from app.crud.newsletters import (
create_newsletter,
get_newsletter_by_identifier,
get_newsletters,
)
from app.crud.settings import create_or_update_settings, get_settings
from app.schemas.entries import EntryCreate
from app.schemas.newsletters import NewsletterCreate
@@ -141,21 +145,21 @@ def test_create_newsletter_with_move_to_folder(db_session: Session):
extract_content=True,
)
newsletter = create_newsletter(db_session, newsletter_data)
retrieved_newsletter = get_newsletter(db_session, newsletter.id)
retrieved_newsletter = get_newsletter_by_identifier(db_session, newsletter.id)
assert retrieved_newsletter.name == "Test Newsletter with Folder"
assert retrieved_newsletter.move_to_folder == "Archive/Test"
assert retrieved_newsletter.extract_content is True
def test_get_newsletter(db_session: Session):
def test_get_newsletter_by_identifier(db_session: Session):
"""Test getting a single newsletter."""
unique_email = f"sender_{uuid.uuid4()}@test.com"
newsletter_data = NewsletterCreate(
name="Test Newsletter 2", sender_emails=[unique_email]
)
created_newsletter = create_newsletter(db_session, newsletter_data)
newsletter = get_newsletter(db_session, created_newsletter.id)
newsletter = get_newsletter_by_identifier(db_session, created_newsletter.id)
assert newsletter.name == "Test Newsletter 2"
assert len(newsletter.senders) == 1
assert newsletter.senders[0].email == unique_email
@@ -271,6 +275,6 @@ def test_delete_newsletter(db_session: Session):
assert deleted_newsletter.name == "Newsletter to Delete"
# Verify it's actually deleted
from app.crud.newsletters import get_newsletter
from app.crud.newsletters import get_newsletter_by_identifier
assert get_newsletter(db_session, newsletter.id) is None
assert get_newsletter_by_identifier(db_session, newsletter.id) is None

View File

@@ -183,16 +183,11 @@ def test_get_newsletter_feed(client: TestClient):
# Atom feed uses a namespace, so we need to include it in our tag searches
ns = {"atom": "http://www.w3.org/2005/Atom"}
links = root.findall("atom:link", ns)
assert any(
link.get("rel") == "alternate" and link.get("href") == "http://backend:8000/"
for link in links
)
assert any(link.get("rel") == "alternate" and link.get("href") for link in links)
logo = root.find("atom:logo", ns)
assert logo is not None
assert logo.text == "http://backend:8000/logo.png"
icon = root.find("atom:icon", ns)
assert icon is not None
assert icon.text == "http://backend:8000/favicon.ico"
entry_titles = [
entry.find("atom:title", ns).text for entry in root.findall("atom:entry", ns)
]

View File

@@ -1,4 +1,5 @@
import uuid
import xml.etree.ElementTree as ET
from sqlalchemy.orm import Session
@@ -36,22 +37,36 @@ def test_generate_feed(db_session: Session):
feed_xml = generate_feed(db_session, newsletter.id)
assert feed_xml is not None
# Parse the feed XML to verify content (simplified check)
# In a real scenario, you'd use an XML parser to validate structure and content more thoroughly
assert f"<title>{newsletter.name}</title>" in feed_xml.decode()
assert f"<id>urn:letterfeed:newsletter:{newsletter.id}</id>" in feed_xml.decode()
assert '<link href="http://backend:8000/" rel="alternate"/>' in feed_xml.decode()
assert "<logo>http://backend:8000/logo.png</logo>" in feed_xml.decode()
assert "<icon>http://backend:8000/favicon.ico</icon>" in feed_xml.decode()
assert "<title>First Entry</title>" in feed_xml.decode()
assert "<title>Second Entry</title>" in feed_xml.decode()
# Parse the feed XML to verify content
root = ET.fromstring(feed_xml)
ns = {"atom": "http://www.w3.org/2005/Atom"}
# Check for top-level elements
assert root.find("atom:title", ns).text == newsletter.name
assert root.find("atom:id", ns).text == f"urn:letterfeed:newsletter:{newsletter.id}"
assert root.find("atom:logo", ns).text.endswith("/logo.png")
assert root.find("atom:icon", ns).text.endswith("/favicon.ico")
# Check for the alternate link
links = root.findall("atom:link", ns)
assert any(link.get("rel") == "alternate" and link.get("href") for link in links)
# Check for entries
entry_titles = [
entry.find("atom:title", ns).text for entry in root.findall("atom:entry", ns)
]
assert "First Entry" in entry_titles
assert "Second Entry" in entry_titles
# Check content of one entry
first_entry_element = root.find(".//atom:title[.='First Entry']/..", ns)
assert (
'<content type="html">&lt;p&gt;This is the first entry.&lt;/p&gt;</content>'
in feed_xml.decode()
first_entry_element.find("atom:content", ns).text
== "<p>This is the first entry.</p>"
)
def test_generate_feed_nonexistent_newsletter(db_session: Session):
"""Test feed generation for a non-existent newsletter."""
feed_xml = generate_feed(db_session, 999) # Non-existent newsletter ID
feed_xml = generate_feed(db_session, "nonexistent-id")
assert feed_xml is None

View File

@@ -0,0 +1,122 @@
import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from app.core.slug import sanitize_slug
@pytest.mark.parametrize(
"input_slug, expected_slug",
[
("Hello World", "hello-world"),
(" leading and trailing spaces ", "leading-and-trailing-spaces"),
("!@#$%^&*()", None),
("a-b_c d", "a-b-c-d"),
("SLUG IN CAPS", "slug-in-caps"),
(None, None),
("", None),
],
)
def test_sanitize_slug(input_slug, expected_slug):
"""Test the slug sanitization function with various inputs."""
assert sanitize_slug(input_slug) == expected_slug
def test_create_newsletter_with_slug(client: TestClient, db_session: Session):
"""Test creating a newsletter with a custom slug."""
newsletter_data = {
"name": "My Test Newsletter",
"slug": "my-custom-slug",
"sender_emails": ["test@example.com"],
}
response = client.post("/newsletters", json=newsletter_data)
assert response.status_code == 200
data = response.json()
assert data["slug"] == "my-custom-slug"
# Verify the feed URL uses the slug
feed_response = client.get(f"/feeds/{data['slug']}")
assert feed_response.status_code == 200
def test_create_newsletter_with_sanitization(client: TestClient, db_session: Session):
"""Test creating a newsletter with a slug that needs sanitization."""
newsletter_data = {
"name": "Another Test",
"slug": " Another Slug With Spaces! ",
"sender_emails": ["test2@example.com"],
}
response = client.post("/newsletters", json=newsletter_data)
assert response.status_code == 200
data = response.json()
assert data["slug"] == "another-slug-with-spaces"
def test_create_newsletter_without_slug(client: TestClient, db_session: Session):
"""Test creating a newsletter without a slug, expecting it to be None."""
newsletter_data = {
"name": "No Slug Newsletter",
"sender_emails": ["no-slug@example.com"],
}
response = client.post("/newsletters", json=newsletter_data)
assert response.status_code == 200
data = response.json()
assert data["slug"] is None
# Verify the feed URL uses the ID
feed_response = client.get(f"/feeds/{data['id']}")
assert feed_response.status_code == 200
def test_create_newsletter_with_conflicting_slug(
client: TestClient, db_session: Session
):
"""Test creating a newsletter with a slug that already exists."""
# Create the first newsletter
client.post(
"/newsletters",
json={
"name": "First",
"slug": "conflict-slug",
"sender_emails": ["first@example.com"],
},
)
# Attempt to create a second one with the same slug
response = client.post(
"/newsletters",
json={
"name": "Second",
"slug": "conflict-slug",
"sender_emails": ["second@example.com"],
},
)
assert response.status_code == 409
assert response.json()["detail"] == "Slug already in use"
def test_update_newsletter_with_conflicting_slug(
client: TestClient, db_session: Session
):
"""Test updating a newsletter to a slug that is already in use by another newsletter."""
# Create two newsletters
response1 = client.post(
"/newsletters",
json={"name": "First", "slug": "first-slug", "sender_emails": ["1@test.com"]},
)
newsletter1_id = response1.json()["id"]
client.post(
"/newsletters",
json={"name": "Second", "slug": "second-slug", "sender_emails": ["2@test.com"]},
)
# Try to update the first newsletter to use the second's slug
update_data = {
"name": "First Updated",
"slug": "second-slug",
"sender_emails": ["1@test.com"],
}
response = client.put(f"/newsletters/{newsletter1_id}", json=update_data)
assert response.status_code == 409
assert response.json()["detail"] == "Slug already in use"