Miniter API 개발하기

그동안 공부한 API, HTTP의 개념, 구조를 활용해서 축소된 규모의 트위터인 미니터 API를 직접 개발해보자!

미니터의 기능

구현할 미니터의 핵심 기능들은 다음과 같다.

회원가입
로그인
트윗
다른 회원 팔로우
다른 회원 언팔로우
타임라인

회원가입

사용자에게 이름, 이메일, 비밀번호 등의 기본적인 회원정보를 HTTP 요청을 통해 받은 후 시스템상에 저장하는 구조
회원 가입에 필요한 정보로는

id
name
email
password
profile
이 있다고 가정하자.

from flask import Flask, jsonify, request

app = Flask(__name__)
#새로 가입한 사용자를 저장한 dictionary
app.users = {}
app.id_count = 1

# Flask의 route 데코레이터(decorator)를 사용하여 엔드포인트 등록
@app.route("/sign-up", method=['POST'])
def sign_up():
	#HTTP를 통해 전송된 회원 정보를 읽음
    #request는 엔드포인트에 전송된 HTTP 요청 정보를 저장하고 있음.
    new_user = request.json
    new_user['id'] = app.id_count
    app.users[app.id_count] = new_user
    app.id_count = app.id_count + 1
	
    #회원가입한 사용자의 정보를 JSON 형태로 전송
    return jsonify(new_user)

jsonify 모듈 : request를 통해 사용자가 HTTP를 통해 전송한 JSON를 읽고, api내부의 dictionary 객체를 JSON으로 변환하여 HTTP 응답으로 보낼 수 있음

# flask 웹 어플리케이션은 development 모드로 run
export FLASK_ENV=development # flask가 실행되는 개발 스테이지, development로 정해놓으면 debug mode 실행
export FLASK_app=app.py
flask run

# 이후 ide내의 터미널에서 http 요청 보내기
http -v POST localhost:5000/sign-up name="홍진우" email="[email protected]" password=1234

실행화면

300자 제한 트윗 글 올리기

메인 기능 중 하나인 300자 제한 트윗 (tweet) 글 올리기 엔드포인트 (endpoint)를 구현

사용자는 300자를 초과하지 않는 글을 올릴 수 있음.
300자를 초과하면 endpoint는 응답 (response)로 400 Bad Request
사용자가 300자 이내의 글을 전송하면 endpoint는 사용자의 글을 저장하고 있으며, 사용자의 타임라인 endpoint를 통하여 읽을 수 있도록 해야함.

전송할 json 데이터

{
  "id": 1, 
  "tweet": "My First Tweet"
}

app.tweets = [] #사용자의 트윗을 저장할 디렉터리

@app.route("/tweet", methods=["POST"])  # 2
def tweet():
	#HTTP 요청으로 전송된 JSON 데이터에서 "tweet" 필드를 읽음
    payload = request.json
    user_id = int(payload["id"])
    tweet = payload["tweet"]

    if user_id not in app.users:
        return "사용자가 존재하지 않습니다", 400

    if len(tweet) > 300:
        return "300자를 초과했습니다", 400

    app.tweets.append({"user_id": user_id, "tweet": tweet})

    return "", 200

실행 결과

팔로우와 언팔로우 엔드포인트

팔로우 또는 언팔로우하고 싶은 사용자의 아이디를 HTTP 요청 (request)로 보내면 API에서 해당 request를 처리하는 방식으로 구현

팔로우 엔드포인트에 전송할 JSON 데이터
{
"id": 1, -> 해당 사용자의 아이디
"follow": 2 -> 팔로우하고자 하는 사용자의 아이디
}

import os
from flask import Flask, jsonify, request
from flask.json import JSONEncoder


# dict의 key에 대한 value가 set일 때, list로 변경, 그 이외에는 원래의 동작대로
class CustomJSONEncoder(JSONEncoder):  # 1
    def default(self, obj):
        if isinstance(obj, set):
            return list(obj)
        return JSONEncoder.default(self, obj)


app = Flask(__name__)
app.users = {}
app.id_count = 1
app.tweets = []
app.json_encoder = CustomJSONEncoder  # 2


@app.route("/follow", methods=["POST"])  # 3
def follow():
    payload = request.json
    user_id = int(payload["id"])
    user_id_to_follow = int(payload["follow"])

    if user_id not in app.users or user_id_to_follow not in app.users:
        return "사용자가 존재하지 않습니다", 400

    user = app.users[user_id]
    user.setdefault("follow", set()).add(user_id_to_follow)

    return jsonify(user)


@app.route("/unfollow", methods=["POST"])  # 4
def unfollow():
    payload = request.json
    user_id = int(payload["id"])
    user_id_to_unfollow = int(payload["unfollow"])

    if user_id not in app.users or user_id_to_unfollow not in app.users:
        return "사용자가 존재하지 않습니다", 400

    user = app.users[user_id]

    if not user.get("follow") or len(user.get("follow")) == 0:
        return "팔로우하고 있는 사용자가 없습니다", 400

    user.setdefault("follow", set()).discard(user_id_to_unfollow)

CustomJSONEncoder를 구현

user.setdefault("follow", set()).discard(user_id_to_unfollow)
라는 코드에서 읽어 들인 사용자의 정보를 담고 있는 딕셔너리가 이미 "follow"라는 필드를 가지고 있다면, 즉 이미 사용자가 다른 사용자를 팔로우 한 적이 있다면, 사용자의 follow 키와 연결되어 있는 set에 팔로우 하고자 하는 사용자 아이디를 추가함.

처음 다른 사용자를 팔로우 한다는 것이라면, 사용자의 정보를 담고 있는 딕셔너리에 follow라는 키를 empty set과 연결하여 추가

dict의 key에 대한 value가 set의 instance일 경우에 대해서 처리하기 위해
key에 대한 value가 set instance가 아닐 경우에는 원래 JSONEncoder의 default 메소드 (method)를 이용

class CustomJSONEncoder(JSONEncoder):  # 1
    def default(self, obj):
        if isinstance(obj, set):
            return list(obj)
        return JSONEncoder.default(self, obj)

사용자의 아이디를 저장하는 자료구조로 사용하는 set을 파이썬의 json모듈이 JSON으로 변경하지 못하기 때문에, 커스텀 JSON 인코더를 사용

실행결과

타임라인 엔드포인트

타임라인 endpoint는 해당 사용자의 트윗들 그리고 팔로우하는 사용자들의 트윗들을 리턴해주는 endpoint.

HTTP 메소드 (method)는 GET으로 구현 (데이터의 수정없이 받아오기만하기 때문에)
HTTP 응답 (response)의 바디 (body)는 JSON으로

@app.route("/timeline/<int:user_id>", methods=["GET"])  # 1
def timeline(user_id):
    if user_id not in app.users:
        return "사용자가 존재하지 않습니다.", 400

    follow_list = app.users[user_id].get("follow", set())
    follow_list.add(user_id)
    timeline = [tweet for tweet in app.tweets if tweet["user_id"] in follow_list]

    return jsonify({"user_id": user_id, "timeline": timeline})

타임라인 엔드포인트를 호출하기 전에 사용자, 사용자별 트윗 생성, 사용자 팔로우를 한 후 실행 화면

http -v POST localhost:5000/sign-up name="김보섭" && \
http -v POST localhost:5000/sign-up name="클로바" && \
http -v POST localhost:5000/tweet id=1 tweet="Hello World" && \
http -v POST localhost:5000/tweet id=2 tweet="언젠가는 훌륭한 백엔드 개발자가 될 거야" && \
http -v POST localhost:5000/follow id=1 follow=2 && \
http -v GET localhost:5000/timeline/1

전체 소스코드

import os
from flask import Flask, jsonify, request
from flask.json import JSONEncoder


# dict의 key에 대한 value가 set일 때, list로 변경, 그 이외에는 원래의 동작대로
class CustomJSONEncoder(JSONEncoder):
    def default(self, obj):
        if isinstance(obj, set):
            return list(obj)
        return JSONEncoder.default(self, obj)


app = Flask(__name__)
app.users = {}
app.id_count = 1
app.tweets = []
app.json_encoder = CustomJSONEncoder


@app.route("/ping", methods=["GET"])
def ping():
    return "pong"


@app.route("/sign-up", methods=["POST"])
def sign_up():
    new_user = request.json
    new_user["id"] = app.id_count
    app.users[app.id_count] = new_user
    app.id_count += 1
    return jsonify(new_user)


@app.route("/tweet", methods=["POST"])
def tweet():
    payload = request.json
    user_id = int(payload["id"])
    tweet = payload["tweet"]

    if user_id not in app.users:
        return "사용자가 존재하지 않습니다", 400

    if len(tweet) > 300:
        return "300자를 초과했습니다", 400

    app.tweets.append({"user_id": user_id, "tweet": tweet})

    return "", 200


@app.route("/follow", methods=["POST"])
def follow():
    payload = request.json
    user_id = int(payload["id"])
    user_id_to_follow = int(payload["follow"])

    if user_id not in app.users or user_id_to_follow not in app.users:
        return "사용자가 존재하지 않습니다", 400

    user = app.users[user_id]
    user.setdefault("follow", set()).add(user_id_to_follow)

    return jsonify(user)


@app.route("/unfollow", methods=["POST"])
def unfollow():
    payload = request.json
    user_id = int(payload["id"])
    user_id_to_unfollow = int(payload["unfollow"])

    if user_id not in app.users or user_id_to_unfollow not in app.users:
        return "사용자가 존재하지 않습니다", 400

    user = app.users[user_id]

    if not user.get("follow") or len(user.get("follow")) == 0:
        return "팔로우하고 있는 사용자가 없습니다", 400

    user.setdefault("follow", set()).discard(user_id_to_unfollow)

    return jsonify(user)


@app.route("/timeline/<int:user_id>", methods=["GET"])
def timeline(user_id):
    if user_id not in app.users:
        return "사용자가 존재하지 않습니다.", 400

    follow_list = app.users[user_id].get("follow", set())
    follow_list.add(user_id)
    timeline = [tweet for tweet in app.tweets if tweet["user_id"] in follow_list]

    return jsonify({"user_id": user_id, "timeline": timeline})


if __name__ == "__main__":
    os.system("./run_server.sh")

Author And Source

이 문제에 관하여(Miniter API 개발하기), 우리는 이곳에서 더 많은 자료를 발견하고 링크를 클릭하여 보았다 https://velog.io/@eclat12450/Miniter-API-개발하기

우수한 개발자 콘텐츠 발견에 전념 (Collection and Share based on the CC Protocol.)