pyr0ball
add5475d50
- app/services/directus.py: Directus CMS client using docker run curlimages/curl on website_cf-internal network; supports static admin token with fresh JWT login fallback; get/publish/update blog_posts - app/api/endpoints/blog.py: POST /api/v1/blog (publish), GET /slug, PATCH /id endpoints - app/api/routes.py: register blog router - app/core/config.py: add directus_url/token/email/password/network settings - mcp/server.js: add publish_blog_post and get_blog_post MCP tools Key gotcha: Directus filter[field][_eq] brackets must be percent-encoded when passed as a curl CLI URL arg — raw brackets cause curl to exit non-zero with empty stderr.
63 lines
1.9 KiB
Python
63 lines
1.9 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, HTTPException
|
|
from pydantic import BaseModel
|
|
|
|
from app.services.directus import get_blog_post, publish_blog_post, update_blog_post
|
|
|
|
router = APIRouter(prefix="/blog", tags=["blog"])
|
|
|
|
|
|
class PublishRequest(BaseModel):
|
|
title: str
|
|
body: str
|
|
slug: str | None = None
|
|
tags: list[str] | None = None
|
|
author: str | None = None
|
|
seo_description: str | None = None
|
|
published_at: str | None = None # ISO 8601; defaults to now
|
|
|
|
|
|
class UpdateRequest(BaseModel):
|
|
title: str | None = None
|
|
body: str | None = None
|
|
tags: list[str] | None = None
|
|
seo_description: str | None = None
|
|
published_at: str | None = None
|
|
|
|
|
|
@router.post("", summary="Publish a blog post to the CircuitForge website via Directus")
|
|
def publish(req: PublishRequest) -> dict:
|
|
try:
|
|
item = publish_blog_post(
|
|
title=req.title,
|
|
body=req.body,
|
|
slug=req.slug,
|
|
tags=req.tags,
|
|
author=req.author,
|
|
seo_description=req.seo_description,
|
|
published_at=req.published_at,
|
|
)
|
|
return {"ok": True, "post": item}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=502, detail=str(e))
|
|
|
|
|
|
@router.get("/{slug}", summary="Fetch a blog post by slug")
|
|
def get_post(slug: str) -> dict:
|
|
post = get_blog_post(slug)
|
|
if post is None:
|
|
raise HTTPException(status_code=404, detail=f"No blog post with slug '{slug}'")
|
|
return post
|
|
|
|
|
|
@router.patch("/{post_id}", summary="Update an existing blog post by Directus ID")
|
|
def update(post_id: int, req: UpdateRequest) -> dict:
|
|
fields = req.model_dump(exclude_none=True)
|
|
if not fields:
|
|
raise HTTPException(status_code=400, detail="No fields to update")
|
|
try:
|
|
item = update_blog_post(post_id, fields)
|
|
return {"ok": True, "post": item}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=502, detail=str(e))
|