Rippling Web API：简化 Quora 发帖与赞踩接口

题目重点：从资源路由、用户×题目描述投票状态、反向改票差量、HTTP 错误码与并发原子性实现可运行训练 API。

题意与假设

这道练习题保留了面经中值得训练的核心约束，并将需要澄清的条件单独列出。现场拿到变体时，先和面试官确认这些条件。

训练端点为 POST /posts、GET /posts、GET /posts/{id}、POST /posts/{id}/votes；`X-User-Id` 是本机测试替身，不是安全认证。真实方案应使用经验证的 session/token 与权限校验。
每位用户对一篇帖当前只持有一个 up(+1) 或 down(−1) 状态；同向重复为 no-op，反向改变净分 ±2。题目没公布投票重复与撤销语义，本文沿用同站 Article Voting q8 的明确训练契约，并标明这不是 Quora API 轮原题逐字规则。
训练仅内存、单进程，重启丢数据；题目描述 ID 单调分配，无编辑、删除、评论、分页或幂等发帖键。题目没有要求这些功能。

解题路线

先定义资源与错误语义：POST 新建返回 201、GET 不存在题目描述返回 404、无测试身份返回 401、坏 JSON/投票方向返回 400；列帖按 ID 排，控制输入大小。
存 Post(id,author,title,body,score) 与 Vote(post_id,user_id)→±1；更新时计算 `delta = new−old`，同向重复 0，up→down 为 −2。投票记录和 score 要在同一锁/事务中更新，避免并发重复加分。
把 HTTP 层的解析、路由、状态码与存储层分开；训练用 ThreadingHTTPServer 与锁。生产改为数据库时，可在事务里锁该用户投票行或用唯一键+乐观重试，同步更新得分或由投票表重算，队列异步计分需声明短暂不一致。
网络安全追问从真实身份验证、跨用户写入、输入大小、速率限制、CSRF/CORS 和正文安全渲染出发；不要把任意客户端给的 X-User-Id 当认证。持久化时为 (post_id,user_id) 建唯一索引，题目描述列表再根据查询模式分页。

手动走一遍

用户 U 创建题目描述得 ID 1、分数 0。U up 后 1，再 up 仍 1，改 down 后 −1；用户 V up 后总分 0。若题目描述不存在，投票返回 404；无身份头创建帖返回 401。

Python 参考实现

下面的代码按上文假设实现，适合本地运行并追加自己的测试用例。

代码区可横向滚动；使用键盘时，先聚焦代码区再按方向键查看。

"""Training Quora-like API: posts and per-user up/down votes, in memory.

The source names operations but does not publish endpoint schema or auth rules.
X-User-Id is only a local stand-in for a verified identity, not production auth.
"""

from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
import re
from threading import Lock


class PostStore:
    def __init__(self):
        self.lock = Lock()
        self.next_id = 1
        self.posts = {}
        self.votes = {}  # (post_id, user_id) -> -1 or +1

    def create(self, user_id, title, body):
        if not isinstance(title, str) or not title.strip() or not isinstance(body, str):
            raise ValueError("nonempty title and string body required")
        with self.lock:
            post_id = self.next_id
            self.next_id += 1
            post = {"id": post_id, "author": user_id, "title": title.strip(),
                    "body": body, "score": 0}
            self.posts[post_id] = post
            return dict(post)

    def get(self, post_id):
        with self.lock:
            if post_id not in self.posts:
                raise KeyError(post_id)
            return dict(self.posts[post_id])

    def list_posts(self):
        with self.lock:
            return [dict(self.posts[key]) for key in sorted(self.posts)]

    def vote(self, post_id, user_id, direction):
        if direction not in ("up", "down"):
            raise ValueError("vote must be up or down")
        new = 1 if direction == "up" else -1
        with self.lock:
            if post_id not in self.posts:
                raise KeyError(post_id)
            key = (post_id, user_id)
            old = self.votes.get(key, 0)
            self.votes[key] = new
            self.posts[post_id]["score"] += new - old
            return dict(self.posts[post_id])


def make_handler(store):
    class Handler(BaseHTTPRequestHandler):
        def respond(self, status, result):
            payload = json.dumps(result).encode("utf-8")
            self.send_response(status)
            self.send_header("Content-Type", "application/json; charset=utf-8")
            self.send_header("Content-Length", str(len(payload)))
            self.end_headers()
            self.wfile.write(payload)

        def do_GET(self):
            if self.path == "/posts":
                return self.respond(200, store.list_posts())
            match = re.fullmatch(r"/posts/([1-9][0-9]*)", self.path)
            if not match:
                return self.respond(404, {"error": "unknown route"})
            try:
                return self.respond(200, store.get(int(match.group(1))))
            except KeyError:
                return self.respond(404, {"error": "post not found"})

        def do_POST(self):
            user_id = self.headers.get("X-User-Id", "").strip()
            if not user_id:
                return self.respond(401, {"error": "user required"})
            try:
                size = int(self.headers.get("Content-Length", "0"))
                if size < 0 or size > 65536:
                    raise ValueError("body too large")
                data = json.loads(self.rfile.read(size))
                if not isinstance(data, dict):
                    raise ValueError("JSON object required")
                if self.path == "/posts":
                    return self.respond(201, store.create(user_id, data.get("title"), data.get("body")))
                match = re.fullmatch(r"/posts/([1-9][0-9]*)/votes", self.path)
                if not match:
                    return self.respond(404, {"error": "unknown route"})
                return self.respond(200, store.vote(int(match.group(1)), user_id, data.get("vote")))
            except (ValueError, json.JSONDecodeError) as error:
                return self.respond(400, {"error": str(error)})
            except KeyError:
                return self.respond(404, {"error": "post not found"})

        def log_message(self, format, *args):
            pass

    return Handler


if __name__ == "__main__":
    ThreadingHTTPServer(("127.0.0.1", 8000), make_handler(PostStore())).serve_forever()

复杂度

训练内存中创建、取单帖与单次投票平均 O(1)；列出 P 帖并按 ID 排为 O(P log P)，题目描述与投票状态空间 O(P+V)。生产数据库需要索引和分页，上述复杂度不包含 HTTP/序列化、锁竞争或磁盘成本。

易错点与追问

容易失分的边界

每次 up 都累加，用户刷新页面可无限刷分。
up→down 只减一，忘了撤销旧 up。
先写 Vote 再另行写 score，崩溃或并发使两者失配。
把示例 X-User-Id 当安全认证。
45 分钟只画架构，没有能启动/调用的最小 API。

继续深挖

发帖请求超时重试如何避免重复题目描述？
用户撤销投票时 score 如何变？
数据库中如何用唯一约束和事务处理双设备并发？
热门题目描述投票写热点与列表排名如何扩展？
如何防止脚本刷票和 XSS？