f1b5aeac53
All checks were successful
Build and Deploy / build-and-deploy (push) Successful in 55s
Replace direct Post, MarketPlace, Calendar model queries and HTTP API calls with typed service calls. Events registers all 4 services via domain_services_fn with has() guards. Key changes: - app.py: use domain_services_fn, Post/Calendar/MarketPlace queries → services.blog/calendar/market, HTTP cart API → services.cart - calendars/markets services: Post → services.blog - post_associations: Post → services.blog, direct queries → services - markets routes: remove unused MarketPlace import - glue imports → shared imports throughout Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
122 lines
3.3 KiB
Python
122 lines
3.3 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
|
|
from shared.services.registry import services
|
|
|
|
|
|
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 services.blog.get_post_by_id(session, 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.content_type == "post",
|
|
CalendarEntryPost.content_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,
|
|
content_type="post",
|
|
content_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.content_type == "post",
|
|
CalendarEntryPost.content_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:
|
|
"""
|
|
Get all posts (as PostDTOs) associated with a calendar entry.
|
|
"""
|
|
result = await session.execute(
|
|
select(CalendarEntryPost.content_id).where(
|
|
CalendarEntryPost.entry_id == entry_id,
|
|
CalendarEntryPost.content_type == "post",
|
|
CalendarEntryPost.deleted_at.is_(None),
|
|
)
|
|
)
|
|
post_ids = list(result.scalars().all())
|
|
if not post_ids:
|
|
return []
|
|
posts = await services.blog.get_posts_by_ids(session, post_ids)
|
|
return sorted(posts, key=lambda p: (p.title or ""))
|
|
|
|
|
|
async def search_posts(
|
|
session: AsyncSession,
|
|
query: str,
|
|
page: int = 1,
|
|
per_page: int = 10
|
|
) -> tuple[list, int]:
|
|
"""
|
|
Search for posts by title with pagination.
|
|
If query is empty, returns all posts in published order.
|
|
Returns (post_dtos, total_count).
|
|
"""
|
|
return await services.blog.search_posts(session, query, page, per_page)
|