# retoor <retoor@molodetz.nl>
from .core import TTLCache, _in_clause, db, defaultdict, get_table
from .pagination import paginate
from devplacepy.content import enrich_items
from .users import get_users_by_uids
_comment_count_cache = TTLCache(ttl=15, max_size=10000)
def get_comment_counts_by_post_uids(post_uids):
if not post_uids or "comments" not in db.tables:
return {}
result = {}
misses = []
for uid in post_uids:
cached = _comment_count_cache.get(uid)
if cached is None:
misses.append(uid)
else:
result[uid] = cached
if misses:
placeholders, params = _in_clause(misses)
rows = db.query(
f"SELECT target_uid, COUNT(*) as c FROM comments WHERE target_type='post' AND target_uid IN ({placeholders}) AND deleted_at IS NULL GROUP BY target_uid",
**params,
)
fetched = {r["target_uid"]: r["c"] for r in rows}
for uid in misses:
count = fetched.get(uid, 0)
_comment_count_cache.set(uid, count)
result[uid] = count
return result
def get_post_counts_by_user_uids(user_uids):
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 c FROM posts WHERE user_uid IN ({placeholders}) AND deleted_at IS NULL GROUP BY user_uid",
**params,
)
return {r["user_uid"]: r["c"] for r in rows}
def get_vote_counts(target_uids):
if not target_uids or "votes" not in db.tables:
return {}, {}
placeholders, params = _in_clause(target_uids)
rows = db.query(
f"SELECT target_uid, value, COUNT(*) as c FROM votes WHERE target_uid IN ({placeholders}) AND deleted_at IS NULL GROUP BY target_uid, value",
**params,
)
ups = {}
downs = {}
for r in rows:
if r["value"] == 1:
ups[r["target_uid"]] = r["c"]
else:
downs[r["target_uid"]] = r["c"]
return ups, downs
def get_user_votes(user_uid, target_uids):
if not user_uid or not target_uids or "votes" not in db.tables:
return {}
placeholders, params = _in_clause(target_uids)
params["uid"] = user_uid
rows = db.query(
f"SELECT target_uid, value FROM votes WHERE user_uid = :uid AND target_uid IN ({placeholders}) AND deleted_at IS NULL",
**params,
)
return {r["target_uid"]: r["value"] for r in rows}
def get_reactions_by_targets(target_type, target_uids, user=None):
if not target_uids or "reactions" not in db.tables:
return {}
placeholders, params = _in_clause(target_uids)
params["tt"] = target_type
rows = db.query(
f"SELECT target_uid, emoji, COUNT(*) as c FROM reactions WHERE target_type=:tt AND target_uid IN ({placeholders}) AND deleted_at IS NULL GROUP BY target_uid, emoji",
**params,
)
counts = defaultdict(dict)
for row in rows:
counts[row["target_uid"]][row["emoji"]] = row["c"]
mine = defaultdict(list)
if user:
placeholders, params = _in_clause(target_uids, prefix="m")
params["tt"] = target_type
params["u"] = user["uid"]
for row in db.query(
f"SELECT target_uid, emoji FROM reactions WHERE user_uid=:u AND target_type=:tt AND target_uid IN ({placeholders}) AND deleted_at IS NULL",
**params,
):
mine[row["target_uid"]].append(row["emoji"])
result = {}
for uid in target_uids:
result[uid] = {
"counts": dict(counts.get(uid, {})),
"mine": list(mine.get(uid, [])),
}
return result
def get_user_bookmarks(user_uid, target_type, target_uids):
if not user_uid or not target_uids or "bookmarks" not in db.tables:
return set()
placeholders, params = _in_clause(target_uids)
params["u"] = user_uid
params["tt"] = target_type
rows = db.query(
f"SELECT target_uid FROM bookmarks WHERE user_uid=:u AND target_type=:tt AND target_uid IN ({placeholders}) AND deleted_at IS NULL",
**params,
)
return {row["target_uid"] for row in rows}
def get_polls_by_post_uids(post_uids, user=None):
if not post_uids or "polls" not in db.tables:
return {}
placeholders, params = _in_clause(post_uids)
polls = list(
db.query(
f"SELECT * FROM polls WHERE post_uid IN ({placeholders}) AND deleted_at IS NULL",
**params,
)
)
if not polls:
return {}
poll_uids = [poll["uid"] for poll in polls]
option_placeholders, option_params = _in_clause(poll_uids, prefix="o")
options = list(
db.query(
f"SELECT * FROM poll_options WHERE poll_uid IN ({option_placeholders}) AND deleted_at IS NULL ORDER BY position",
**option_params,
)
)
counts = defaultdict(dict)
totals = defaultdict(int)
if "poll_votes" in db.tables:
vote_placeholders, vote_params = _in_clause(poll_uids, prefix="v")
for row in db.query(
f"SELECT poll_uid, option_uid, COUNT(*) as c FROM poll_votes WHERE poll_uid IN ({vote_placeholders}) AND deleted_at IS NULL GROUP BY poll_uid, option_uid",
**vote_params,
):
counts[row["poll_uid"]][row["option_uid"]] = row["c"]
totals[row["poll_uid"]] += row["c"]
user_choice = {}
if user and "poll_votes" in db.tables:
placeholders, params = _in_clause(poll_uids, prefix="m")
params["u"] = user["uid"]
for row in db.query(
f"SELECT poll_uid, option_uid FROM poll_votes WHERE user_uid=:u AND poll_uid IN ({placeholders}) AND deleted_at IS NULL",
**params,
):
user_choice[row["poll_uid"]] = row["option_uid"]
options_by_poll = defaultdict(list)
for option in options:
options_by_poll[option["poll_uid"]].append(option)
result = {}
for poll in polls:
poll_uid = poll["uid"]
total = totals.get(poll_uid, 0)
rendered = []
for option in options_by_poll.get(poll_uid, []):
count = counts.get(poll_uid, {}).get(option["uid"], 0)
rendered.append(
{
"uid": option["uid"],
"label": option["label"],
"count": count,
"pct": round(count * 100 / total) if total else 0,
}
)
result[poll["post_uid"]] = {
"uid": poll_uid,
"question": poll["question"],
"options": rendered,
"total": total,
"my_choice": user_choice.get(poll_uid),
}
return result
def get_poll_for_post(post_uid, user=None):
return get_polls_by_post_uids([post_uid], user).get(post_uid)
def get_project_devlog(project_uid: str, before: str | None = None, viewer: dict | None = None) -> tuple[list, str | None]:
posts_table = get_table("posts")
posts, next_cursor = paginate(
posts_table,
before=before,
viewer_uid=viewer["uid"] if viewer else None,
project_uid=project_uid,
)
if not posts:
return [], None
authors = get_users_by_uids([p["user_uid"] for p in posts])
counts = get_comment_counts_by_post_uids([p["uid"] for p in posts])
result = enrich_items(
posts, "post", authors, {"comment_count": counts}, user=viewer
)
return result, next_cursor
Reference in New Issue View Git Blame Copy Permalink