8f7a15186c
Some checks failed
Build and Deploy / build-and-deploy (push) Has been cancelled
Extract blog-specific code from the coop monolith into a standalone repository. Includes auth, blog, post, admin, menu_items, snippets blueprints, associated templates, Dockerfile (APP_MODULE=app:app), entrypoint, and Gitea CI workflow. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
144 lines
4.0 KiB
Python
144 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select
|
|
from sqlalchemy.sql import func
|
|
|
|
from models.calendars import CalendarEntry, CalendarEntryPost, Calendar
|
|
from models.ghost_content import Post
|
|
|
|
|
|
async def toggle_entry_association(
|
|
session: AsyncSession,
|
|
post_id: int,
|
|
entry_id: int
|
|
) -> tuple[bool, str | None]:
|
|
"""
|
|
Toggle association between a post and calendar entry.
|
|
Returns (is_now_associated, error_message).
|
|
"""
|
|
# Check if entry exists (don't filter by deleted_at - allow associating with any entry)
|
|
entry = await session.scalar(
|
|
select(CalendarEntry).where(CalendarEntry.id == entry_id)
|
|
)
|
|
if not entry:
|
|
return False, f"Calendar entry {entry_id} not found in database"
|
|
|
|
# Check if post exists
|
|
post = await session.scalar(
|
|
select(Post).where(Post.id == post_id)
|
|
)
|
|
if not post:
|
|
return False, "Post not found"
|
|
|
|
# Check if association already exists
|
|
existing = await session.scalar(
|
|
select(CalendarEntryPost).where(
|
|
CalendarEntryPost.entry_id == entry_id,
|
|
CalendarEntryPost.post_id == post_id,
|
|
CalendarEntryPost.deleted_at.is_(None)
|
|
)
|
|
)
|
|
|
|
if existing:
|
|
# Remove association (soft delete)
|
|
existing.deleted_at = func.now()
|
|
await session.flush()
|
|
return False, None
|
|
else:
|
|
# Create association
|
|
association = CalendarEntryPost(
|
|
entry_id=entry_id,
|
|
post_id=post_id
|
|
)
|
|
session.add(association)
|
|
await session.flush()
|
|
return True, None
|
|
|
|
|
|
async def get_post_entry_ids(
|
|
session: AsyncSession,
|
|
post_id: int
|
|
) -> set[int]:
|
|
"""
|
|
Get all entry IDs associated with this post.
|
|
Returns a set of entry IDs.
|
|
"""
|
|
result = await session.execute(
|
|
select(CalendarEntryPost.entry_id)
|
|
.where(
|
|
CalendarEntryPost.post_id == post_id,
|
|
CalendarEntryPost.deleted_at.is_(None)
|
|
)
|
|
)
|
|
return set(result.scalars().all())
|
|
|
|
|
|
async def get_associated_entries(
|
|
session: AsyncSession,
|
|
post_id: int,
|
|
page: int = 1,
|
|
per_page: int = 10
|
|
) -> dict:
|
|
"""
|
|
Get paginated associated entries for this post.
|
|
Returns dict with entries, total_count, and has_more.
|
|
"""
|
|
# Get all associated entry IDs
|
|
entry_ids_result = await session.execute(
|
|
select(CalendarEntryPost.entry_id)
|
|
.where(
|
|
CalendarEntryPost.post_id == post_id,
|
|
CalendarEntryPost.deleted_at.is_(None)
|
|
)
|
|
)
|
|
entry_ids = set(entry_ids_result.scalars().all())
|
|
|
|
if not entry_ids:
|
|
return {
|
|
"entries": [],
|
|
"total_count": 0,
|
|
"has_more": False,
|
|
"page": page,
|
|
}
|
|
|
|
# Get total count
|
|
from sqlalchemy import func
|
|
total_count = len(entry_ids)
|
|
|
|
# Get paginated entries ordered by start_at desc
|
|
# Only include confirmed entries
|
|
offset = (page - 1) * per_page
|
|
result = await session.execute(
|
|
select(CalendarEntry)
|
|
.where(
|
|
CalendarEntry.id.in_(entry_ids),
|
|
CalendarEntry.deleted_at.is_(None),
|
|
CalendarEntry.state == "confirmed" # Only confirmed entries in nav
|
|
)
|
|
.order_by(CalendarEntry.start_at.desc())
|
|
.limit(per_page)
|
|
.offset(offset)
|
|
)
|
|
entries = result.scalars().all()
|
|
|
|
# Recalculate total_count based on confirmed entries only
|
|
total_count = len(entries) + offset # Rough estimate
|
|
if len(entries) < per_page:
|
|
total_count = offset + len(entries)
|
|
|
|
# Load calendar relationship for each entry
|
|
for entry in entries:
|
|
await session.refresh(entry, ["calendar"])
|
|
if entry.calendar:
|
|
await session.refresh(entry.calendar, ["post"])
|
|
|
|
has_more = len(entries) == per_page # More accurate check
|
|
|
|
return {
|
|
"entries": entries,
|
|
"total_count": total_count,
|
|
"has_more": has_more,
|
|
"page": page,
|
|
}
|