[python] Flask 프레임워크로 api 서버 구축(1)

업데이트: March 22, 2020

파이썬 Flask 프레임워크로 api 서버 구축(1)

참고링크

참고도서

깔끔한 파이썬 탄탄한 백엔드

0. 사전 준비

위 링크를 참고 하셔서 가상환경을 설치하시면 되겠습니다.

1. 가상환경 실행 및 flask 프레임워크 실행확인

저는 가상환경 이름을 py3_7_2라고 지었습니다.

$ pyenv activate py3_7_2
(py3_7_2)$ python
>>> from flask import flask
>>> app = Flask("test")
>>> app
<Flask 'test'> 

2. 헬스체크(health check) 엔드포인트 구현

‘ping’ 이라고 찍으면 ‘pong’라고 답이 오는 엔드포인트를 구현해보겠습니다. 먼저 원하는 디렉터리로 이동해 파이썬 파일을 생성해 보겠습니다.

$ cd ./practice/practice_flask
$ vim app.py

app.py 파일은 아래와 같이 생겼습니다.

from flask import Flask

# Flask 클래스를 객체화 시켜 app이라는 변수에 저장
app = Flask(__name__)

# ping 함수를 엔드포인트로 등록
@app.route("/ping", methods=['GET'])
def ping():
    return "pong"

그리고 api를 실행해 봅시다.

(py3_7_2)$ ~/practice/practice_flask$ FLASK_APP=app.py FLASK_DEBUG=1 flask run
 * Serving Flask app "app.py" (lazy loading)
 * Environment: production
   WARNING: This is a development server. Do not use it in a production deployment.
   Use a production WSGI server instead.
 * Debug mode: on
 * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 130-161-363

터미널을 하나 더 띄워서 api 테스트를 위해 httpie 설치하겠습니다.

$ sudo apt install httpie //우분투 
$ brew install httpie //맥

그리고 api테스트를 해봅시다.

$ http -v GET http://localhost:5000/ping
GET /ping HTTP/1.1
Accept: */*
Accept-Encoding: gzip, deflate
Connection: keep-alive
Host: localhost:5000
User-Agent: HTTPie/0.9.8

HTTP/1.0 200 OK
Content-Length: 4
Content-Type: text/html; charset=utf-8
Date: Sun, 22 Mar 2020 06:54:45 GMT
Server: Werkzeug/1.0.0 Python/3.7.2

pong

위 결과 처럼 api요청을 하면 pong 이라는 응답이 잘 오는 것을 볼 수 있습니다.

3. 미니 트위터 api 서버 개발

3.1 회원가입

필요한 정보

id
name
email
password
profile

위에서 작성했던 app.py함수를 수정합시다.

# jsonify: dictionary -> JSON 변환
# request: http 요청 가능
from flask import Flask, jsonify, request

app = Flask(__name__)
app.id_count = 1
app.users = {}
 
@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            = app.id_count + 1

    return jsonify(new_user)

3.2 회원가입 테스트

위에서 작성한 app.py 파일을 사용해서 서버를 띄워봅시다.

(py3_7_2)$ ~/practice/practice_flask$ FLASK_ENV=development FKAS_APP=app.py flask run
 * Environment: development
 * Debug mode: on
 * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 130-161-363

서버가 띄워졌으면 다른 터미널 창을 열어서 회원가입 요청을 보내봅시다.

$ http -v POST localhost:5000/sign-up name=홍길동 email=gildong@naver.com password=abcd1234
POST /sign-up HTTP/1.1
Accept: application/json, */*
Accept-Encoding: gzip, deflate
Connection: keep-alive
Content-Length: 84
Content-Type: application/json
Host: localhost:5000
User-Agent: HTTPie/0.9.8

{
    "email": "gildong@naver.com",
    "name": "홍길동",
    "password": "abcd1234"
}

HTTP/1.0 200 OK
Content-Length: 107
Content-Type: application/json
Date: Sun, 22 Mar 2020 07:59:28 GMT
Server: Werkzeug/1.0.0 Python/3.7.2

{
    "email": "gildong@naver.com",
    "id": 1,
    "name": "홍길동",
    "password": "abcd1234"
}

3.3 트윗 엔드포인트

트윗을 할 때 서버에 전송하는 json데이터는 아래와 같습니다.

{
  "id": 1, # 사용자 아이디
  "tweet":"This is my first tweet" # 트윗 내용
}

위 형식을 생각하고 app.py 파일을 수정합니다.

from flask import Flask, jsonify, request

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

@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			= 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
	user_id = int(payload['id'])
	app.tweets.append({
		'user_id' : user_id,
		'tweet' : tweet
	})
	return '', 200

3.4 트윗 테스트

서버를 먼저 띄우고

(py3_7_2)$ ~/practice/practice_flask$ FLASK_ENV=development FKAS_APP=app.py flask run
 * Environment: development
 * Debug mode: on
 * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 130-161-363

다른 터미널 창을 띄우고 트윗을 포스팅 해 봅시다.

$ http -v POST localhost:5000/tweet id:=1 tweet="This is my first tweet"
POST /tweet HTTP/1.1
Accept: application/json, */*
Accept-Encoding: gzip, deflate
Connection: keep-alive
Content-Length: 44
Content-Type: application/json
Host: localhost:5000
User-Agent: HTTPie/0.9.8

{
    "id": 1,
    "tweet": "This is my first tweet"
}

HTTP/1.0 200 OK
Content-Length: 0
Content-Type: text/html; charset=utf-8
Date: Sun, 22 Mar 2020 16:28:37 GMT
Server: Werkzeug/1.0.0 Python/3.7.2

3.5 팔로우&언팔로우 엔드포인트

팔로우 및 언팔로우 요청은 아래와 같습니다.

{
    "id" : 1,
    "follow" : 2
} 

{
    "id" : 1,
    "unfollow" : 2
} 

위 내용을 기억하면서 app.py 파일을 작성합시다.

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


### default json endocer는 set을 json으로 변환 불가능
### 따라서 커스텀인코더 작성을 통해 set을 list로 변환하여
### JSON 변환가능하게 만들어줘야함
class CustomJSONEncoder(JSONEncoder):
	def default(self, obj):
		if isinstane(obj, set):
			return list(obj)
		return JSONEncoder.default(self, obj)

app = Flask(__name__)
app.id_count = 1
app.users = {}
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			= 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
	user_id = int(payload['id'])
	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_follow = int(payload['unfollow'])

	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()).descard(user_id_to_follow)

	return jsonify(user)

3.6 팔로우테스트

서버를 띄우고

(py3_7_2)$ ~/practice/practice_flask$ FLASK_ENV=development FKAS_APP=app.py flask run
 * Environment: development
 * Debug mode: on
 * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 130-161-363

터미널 하나 더 띄워서 아래처럼 팔로우 테스트할 두명의 사용자를 회원가입 시킵시다.

$ http -v POST localhost:5000/sign-up name=홍길동 email=gildong@naver.com password=abcd1234
$ http -v POST localhost:5000/sign-up name=임꺽정 email=ggukjeong@naver.com password=abcd1233

위 처럼 입력하면 홍길동의 아이디는 1이고, 임꺽정의 아이디는 2가 된다. 따라서 홍길동이 임꺽정을 팔로우 하고 싶으면 아래와 같이 명령하면 팔로우 테스트 끝

$ http -v POST localhost:5000/follow id:=1 follow:=2
POST /follow HTTP/1.1
Accept: application/json, */*
Accept-Encoding: gzip, deflate
Connection: keep-alive
Content-Length: 22
Content-Type: application/json
Host: localhost:5000
User-Agent: HTTPie/0.9.8

{
    "follow": 2,
    "id": 1
}

HTTP/1.0 200 OK
Content-Length: 133
Content-Type: application/json
Date: Sun, 22 Mar 2020 17:14:06 GMT
Server: Werkzeug/1.0.0 Python/3.7.2

{
    "email": "gildong@naver.com",
    "follow": [
        2
    ],
    "id": 1,
    "name": "홍길동",
    "password": "abcd1234"
}

3.7 타임라인 엔트포인트

타임라인엔드포인트 응답형태는 다음과 같습니다.

{
	"user_id" : 1,
	"timeline" : [
		{
			"user_id" : 2,
			"tweet" : "Hello, world!"
		},
		{
			"user_id":1,
			"tweet" : "My first tweet!"
		}
	]
}

그리고 app.py를 수정합시다.

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


### default json endocer는 set을 json으로 변환 불가능
### 따라서 커스텀인코더 작성을 통해 set을 list로 변환하여
### JSON 변환가능하게 만들어줘야함
class CustomJSONEncoder(JSONEncoder):
	def default(self, obj):
		if isinstance(obj, set):
			return list(obj)
		return JSONEncoder.default(self, obj)

app = Flask(__name__)
app.id_count = 1
app.users = {}
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			= 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
	user_id = int(payload['id'])
	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_follow = int(payload['unfollow'])

	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()).descard(user_id_to_follow)

	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
	})

3.8 타임라인 엔트포인트 테스트

서버를 띄우고

(py3_7_2)$ ~/practice/practice_flask$ FLASK_ENV=development FKAS_APP=app.py flask run
 * Environment: development
 * Debug mode: on
 * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 130-161-363

터미널 하나 더 띄워서 아래처럼 팔로우 테스트할 두명의 사용자를 회원가입 시킵시다.

$ http -v POST localhost:5000/sign-up name=홍길동 email=gildong@naver.com password=abcd1234
$ http -v POST localhost:5000/sign-up name=임꺽정 email=ggukjeong@naver.com password=abcd1233

그리고 서로를 팔로우하게 만들고

$ http -v POST localhost:5000/follow id:=1 follow:=2
$ http -v POST localhost:5000/follow id:=2 follow:=1

각자 하고싶은 말을 트윗해봅시다.

$ http -v POST localhost:5000/tweet id:=1 tweet="This is my first tweet"
$ http -v POST localhost:5000/tweet id:=2 tweet="Life is valuable"

그리고 마지막으로 타임라인 테스트를 하면 완성

http -v GET localhost:5000/timeline/1
GET /timeline/1 HTTP/1.1
Accept: */*
Accept-Encoding: gzip, deflate
Connection: keep-alive
Host: localhost:5000
User-Agent: HTTPie/0.9.8

HTTP/1.0 200 OK
Content-Length: 183
Content-Type: application/json
Date: Sun, 22 Mar 2020 17:42:45 GMT
Server: Werkzeug/1.0.0 Python/3.7.2

{
    "timeline": [
        {
            "tweet": "This is my first tweet",
            "user_id": 1
        },
        {
            "tweet": "Life is valuable",
            "user_id": 2
        }
    ],
    "user_id": 1
}

Twitter Facebook Google+ LinkedIn

장철원(Cheolwon Jang)