storage.py

# storage.py
"""JSON-backed storage layer for the minimal blog.

Each post has keys: id, author, title, content, likes (int).
"""
from __future__ import annotations

from typing import Dict, List, Optional
import json
import os
import tempfile
import threading

_STORAGE_FILE = os.path.join(os.path.dirname(__file__), "posts.json")
_LOCK = threading.Lock()


def _atomic_write(path: str, data: str) -> None:
    """Atomically write data to avoid partial writes on crash."""
    dir_ = os.path.dirname(path) or "."
    fd, tmp_path = tempfile.mkstemp(prefix=".__tmp__", dir=dir_)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as tmp:
            tmp.write(data)
        os.replace(tmp_path, path)
    finally:
        try:
            if os.path.exists(tmp_path):
                os.remove(tmp_path)
        except Exception:
            pass


def _normalize(posts: List[Dict]) -> List[Dict]:
    """Ensure each post has 'likes' (backfill to 0 if missing)."""
    changed = False
    for p in posts:
        if "likes" not in p:
            p["likes"] = 0
            changed = True
    if changed:
        save_posts(posts)
    return posts


def load_posts() -> List[Dict]:
    """Return all posts (empty list if file missing)."""
    if not os.path.exists(_STORAGE_FILE):
        return []
    with _LOCK, open(_STORAGE_FILE, "r", encoding="utf-8") as f:
        posts: List[Dict] = json.load(f)
    return _normalize(posts)


def save_posts(posts: List[Dict]) -> None:
    """Persist posts to disk (atomic)."""
    with _LOCK:
        payload = json.dumps(posts, ensure_ascii=False, indent=2)
        _atomic_write(_STORAGE_FILE, payload)


def _next_id(posts: List[Dict]) -> int:
    """Return the next incremental integer ID."""
    return max((int(p.get("id", 0)) for p in posts), default=0) + 1


def get_post(post_id: int) -> Optional[Dict]:
    """Return a single post by ID, or None."""
    return next((p for p in load_posts() if int(p.get("id")) == post_id), None)


def add_post(author: str, title: str, content: str) -> Dict:
    """Create and persist a new post."""
    posts = load_posts()
    new_post = {
        "id": _next_id(posts),
        "author": author,
        "title": title,
        "content": content,
        "likes": 0,
    }
    posts.append(new_post)
    save_posts(posts)
    return new_post


def update_post(
    post_id: int,
    *,
    author: Optional[str] = None,
    title: Optional[str] = None,
    content: Optional[str] = None,
) -> Optional[Dict]:
    """Update fields of a post; return updated post or None."""
    posts = load_posts()
    for p in posts:
        if int(p.get("id")) == post_id:
            if author is not None:
                p["author"] = author
            if title is not None:
                p["title"] = title
            if content is not None:
                p["content"] = content
            save_posts(posts)
            return p
    return None


def like_post(post_id: int) -> Optional[Dict]:
    """Increment the 'likes' counter and return the updated post, or None."""
    posts = load_posts()
    for p in posts:
        if int(p.get("id")) == post_id:
            p["likes"] = int(p.get("likes", 0)) + 1
            save_posts(posts)
            return p
    return None


def delete_post(post_id: int) -> bool:
    """Delete a post by ID; return True if a post was removed."""
    posts = load_posts()
    new_posts = [p for p in posts if int(p.get("id")) != post_id]
    if len(new_posts) == len(posts):
        return False
    save_posts(new_posts)
    return True