ad/molodetz/database/posts.py

# retoor <retoor@molodetz.nl>
from molodetz.database.core import db, generate_uid, in_clause, make_combined_slug, now_iso, resolve_by_slug
from molodetz.database.pagination import paginate


def insert_post(user_uid, title, body, topic, status="draft", is_placeholder=False, created_at=None):
    uid = generate_uid()
    stamp = created_at or now_iso()
    row = {
        "uid": uid,
        "slug": make_combined_slug(title, uid),
        "title": title,
        "body": body,
        "topic": topic,
        "status": status,
        "is_placeholder": bool(is_placeholder),
        "user_uid": user_uid,
        "published_at": stamp if status == "published" else None,
        "created_at": stamp,
        "updated_at": stamp,
        "deleted_at": None,
        "deleted_by": None,
    }
    db["posts"].insert(row)
    return row


def update_post_row(uid, **fields):
    fields["uid"] = uid
    db["posts"].update(fields, ["uid"])


def get_post(uid):
    if "posts" not in db.tables:
        return None
    return db["posts"].find_one(uid=uid, deleted_at=None)


def get_post_by_slug(slug):
    return resolve_by_slug("posts", slug)


def get_post_by_slug_tail(slug):
    tail = (slug or "").rsplit("-", 1)[-1]
    if "posts" not in db.tables or len(tail) != 12 or not all(c in "0123456789abcdef" for c in tail):
        return None
    for row in db["posts"].find(deleted_at=None):
        if row["uid"].replace("-", "").endswith(tail):
            return row
    return None


def published_posts(topic=None, before=None, limit=25, user_uid=None):
    filters = {"status": "published"}
    if topic:
        filters["topic"] = topic
    if user_uid:
        filters["user_uid"] = user_uid
    return paginate("posts", before=before, cursor_field="published_at", limit=limit, **filters)


def all_posts(limit=200):
    if "posts" not in db.tables:
        return []
    return list(db["posts"].find(deleted_at=None, order_by=["-created_at"], _limit=limit))


def count_published():
    if "posts" not in db.tables:
        return 0
    return db["posts"].count(deleted_at=None, status="published")


def post_counts_by_author(user_uids):
    user_uids = [uid for uid in set(user_uids) if uid]
    if not user_uids or "posts" not in db.tables:
        return {}
    placeholders, params = in_clause(user_uids)
    rows = db.query(
        f"SELECT user_uid, COUNT(*) AS n FROM posts WHERE user_uid IN ({placeholders}) "
        "AND deleted_at IS NULL AND status = 'published' GROUP BY user_uid",
        **params,
    )
    return {row["user_uid"]: int(row["n"]) for row in rows}