feat: initialize events app with calendars, slots, tickets, and internal API
Some checks failed
Build and Deploy / build-and-deploy (push) Has been cancelled
Some checks failed
Build and Deploy / build-and-deploy (push) Has been cancelled
Extract events/calendar functionality into standalone microservice: - app.py and events_api.py from apps/events/ - Calendar blueprints (calendars, calendar, calendar_entries, calendar_entry, day, slots, slot, ticket_types, ticket_type) - Templates for all calendar/event views including admin - Dockerfile (APP_MODULE=app:app, IMAGE=events) - entrypoint.sh (no Alembic - migrations managed by blog app) - Gitea CI workflow for build and deploy Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
137
bp/calendar_entry/services/post_associations.py
Normal file
137
bp/calendar_entry/services/post_associations.py
Normal file
@@ -0,0 +1,137 @@
|
||||
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
|
||||
from models.ghost_content import Post
|
||||
|
||||
|
||||
async def add_post_to_entry(
|
||||
session: AsyncSession,
|
||||
entry_id: int,
|
||||
post_id: int
|
||||
) -> tuple[bool, str | None]:
|
||||
"""
|
||||
Associate a post with a calendar entry.
|
||||
Returns (success, error_message).
|
||||
"""
|
||||
# Check if entry exists
|
||||
entry = await session.scalar(
|
||||
select(CalendarEntry).where(
|
||||
CalendarEntry.id == entry_id,
|
||||
CalendarEntry.deleted_at.is_(None)
|
||||
)
|
||||
)
|
||||
if not entry:
|
||||
return False, "Calendar entry not found"
|
||||
|
||||
# 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:
|
||||
return False, "Post is already associated with this entry"
|
||||
|
||||
# Create association
|
||||
association = CalendarEntryPost(
|
||||
entry_id=entry_id,
|
||||
post_id=post_id
|
||||
)
|
||||
session.add(association)
|
||||
await session.flush()
|
||||
|
||||
return True, None
|
||||
|
||||
|
||||
async def remove_post_from_entry(
|
||||
session: AsyncSession,
|
||||
entry_id: int,
|
||||
post_id: int
|
||||
) -> tuple[bool, str | None]:
|
||||
"""
|
||||
Remove a post association from a calendar entry (soft delete).
|
||||
Returns (success, error_message).
|
||||
"""
|
||||
# Find the association
|
||||
association = await session.scalar(
|
||||
select(CalendarEntryPost).where(
|
||||
CalendarEntryPost.entry_id == entry_id,
|
||||
CalendarEntryPost.post_id == post_id,
|
||||
CalendarEntryPost.deleted_at.is_(None)
|
||||
)
|
||||
)
|
||||
|
||||
if not association:
|
||||
return False, "Association not found"
|
||||
|
||||
# Soft delete
|
||||
association.deleted_at = func.now()
|
||||
await session.flush()
|
||||
|
||||
return True, None
|
||||
|
||||
|
||||
async def get_entry_posts(
|
||||
session: AsyncSession,
|
||||
entry_id: int
|
||||
) -> list[Post]:
|
||||
"""
|
||||
Get all posts associated with a calendar entry.
|
||||
"""
|
||||
result = await session.execute(
|
||||
select(Post)
|
||||
.join(CalendarEntryPost)
|
||||
.where(
|
||||
CalendarEntryPost.entry_id == entry_id,
|
||||
CalendarEntryPost.deleted_at.is_(None)
|
||||
)
|
||||
.order_by(Post.title)
|
||||
)
|
||||
return list(result.scalars().all())
|
||||
|
||||
|
||||
async def search_posts(
|
||||
session: AsyncSession,
|
||||
query: str,
|
||||
page: int = 1,
|
||||
per_page: int = 10
|
||||
) -> tuple[list[Post], int]:
|
||||
"""
|
||||
Search for posts by title with pagination.
|
||||
If query is empty, returns all posts in published order.
|
||||
Returns (posts, total_count).
|
||||
"""
|
||||
# Build base query
|
||||
if query:
|
||||
# Search by title
|
||||
count_stmt = select(func.count(Post.id)).where(Post.title.ilike(f"%{query}%"))
|
||||
posts_stmt = select(Post).where(Post.title.ilike(f"%{query}%")).order_by(Post.title)
|
||||
else:
|
||||
# All posts in published order (newest first)
|
||||
count_stmt = select(func.count(Post.id))
|
||||
posts_stmt = select(Post).order_by(Post.published_at.desc().nullslast())
|
||||
|
||||
# Count total
|
||||
count_result = await session.execute(count_stmt)
|
||||
total = count_result.scalar() or 0
|
||||
|
||||
# Get paginated results
|
||||
offset = (page - 1) * per_page
|
||||
result = await session.execute(
|
||||
posts_stmt.limit(per_page).offset(offset)
|
||||
)
|
||||
return list(result.scalars().all()), total
|
||||
Reference in New Issue
Block a user