[python] Flask 프레임워크로 api 서버 구축(2)
업데이트:
파이썬 Flask 프레임워크로 api 서버 구축(2)
참고링크
- 아나콘다로 가상환경 설치하기(윈도우)
- 파이썬 가상환경 설치하기(우분투)
- 파이썬 가상환경 설치하기(centOS6)
- mysql 설치하기
- 파이썬과 DB 연동하기
- Flask를 이용한 api 서버 구축(1)
- Flask를 이용한 api 서버 구축(2)
- Flask로 한글 받기
참고도서
- 깔끔한 파이썬 탄탄한 백엔드
1. DB 및 테이블 생성
mysql 접속
$ mysql -u root -p
DB생성
mysql> create database miniter;
DB선택
mysql> use miniter;
Database changed
DB선택 확인
mysql> select database();
+------------+
| database() |
+------------+
| miniter |
+------------+
1 row in set (0.00 sec)
테이블 생성
CREATE TABLE users(
id INT NOT NULL AUTO_INCREMENT,
name VARCHAR(255) NOT NULL,
email VARCHAR(255) NOT NULL,
hashed_password VARCHAR(255) NOT NULL,
profile VARCHAR(2000) NOT NULL,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP NULL DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (id),
UNIQUE KEY email (email)
);
CREATE TABLE users_follow_list(
user_id INT NOT NULL,
follow_user_id INT NOT NULL,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (user_id, follow_user_id),
CONSTRAINT users_follow_list_user_id_fkey FOREIGN KEY (user_id) REFERENCES users(id),
CONSTRAINT users_follow_list_follow_user_id_fkey FOREIGN KEY (follow_user_id) REFERENCES users(id)
);
CREATE TABLE tweets(
id INT NOT NULL AUTO_INCREMENT,
user_id INT NOT NULL,
tweet VARCHAR(300) NOT NULL,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (id),
CONSTRAINT tweets_user_id_fkey FOREIGN KEY (user_id) REFERENCES users(id)
);
테이블을 다 만들었으면 어디 잘 만들어졌는지 확인해봅시다
mysql> show tables;
+-------------------+
| Tables_in_miniter |
+-------------------+
| tweets |
| users |
| users_follow_list |
+-------------------+
3 rows in set (0.00 sec)
mysql> explain users;
+-----------------+---------------+------+-----+-------------------+-----------------------------+
| Field | Type | Null | Key | Default | Extra |
+-----------------+---------------+------+-----+-------------------+-----------------------------+
| id | int(11) | NO | PRI | NULL | auto_increment |
| name | varchar(255) | NO | | NULL | |
| email | varchar(255) | NO | UNI | NULL | |
| hashed_password | varchar(255) | NO | | NULL | |
| profile | varchar(2000) | NO | | NULL | |
| created_at | timestamp | NO | | CURRENT_TIMESTAMP | |
| updated_at | timestamp | YES | | NULL | on update CURRENT_TIMESTAMP |
+-----------------+---------------+------+-----+-------------------+-----------------------------+
7 rows in set (0.03 sec)
mysql> explain users_follow_list;
+----------------+-----------+------+-----+-------------------+-------+
| Field | Type | Null | Key | Default | Extra |
+----------------+-----------+------+-----+-------------------+-------+
| user_id | int(11) | NO | PRI | NULL | |
| follow_user_id | int(11) | NO | PRI | NULL | |
| created_at | timestamp | NO | | CURRENT_TIMESTAMP | |
+----------------+-----------+------+-----+-------------------+-------+
3 rows in set (0.01 sec)
mysql> explain tweets;
+------------+--------------+------+-----+-------------------+----------------+
| Field | Type | Null | Key | Default | Extra |
+------------+--------------+------+-----+-------------------+----------------+
| id | int(11) | NO | PRI | NULL | auto_increment |
| user_id | int(11) | NO | MUL | NULL | |
| tweet | varchar(300) | NO | | NULL | |
| created_at | timestamp | NO | | CURRENT_TIMESTAMP | |
+------------+--------------+------+-----+-------------------+----------------+
4 rows in set (0.00 sec)
작업이 끝났으면 이제 나갑시다.
mysql> exit
2. 파이썬 db연동
sqlalchemy설치
$ pyenv activate py3_7_2
$ pip install sqlalchemy
$ pip install mysql-connector-python
API서버와 DB서버 연동하기
API서버가 있는 곳으로 이동 후 db 설정 파일 생성
$ cd practice/practice_flask
$ vim config.py
db = {
'user' : 'root',
'password' : '자신이 설정한 비밀번호',
'host' : 'localhost',
'port' : 3306,
'database' : 'miniter'
}
DB_URL = f"mysql+mysqlconnector://{db['user']}:{db['password']}@{db['host']}:{db['port']}/{db['database']}?charset=utf8"
app2.py생성
app.py와는 다르게 DB생성 부분을 추가시켜줘야한다.
def create_app(test_config=None):
app = Flask(__name__)
app.json_encoder = CustomJSONEncoder
if test_config is None:
app.config.from_pyfile("config.py")
else:
app.config.update(test_config)
database = create_engine(app.config['DB_URL'], encoding='utf-8', max_overflow=0)
app.database = database
return app
회원가입 엔드포인트를 바꿔봅시다. 바꾸기 이전에는 아래와 같은 코드가
@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
DB를 연동시키면 아래와 같이 바뀝니다.
def create_app(test_config=None):
app = Flask(__name__)
app.json_encoder = CustomJSONEncoder
if test_config is None:
app.config.from_pyfile("config.py")
else:
app.config.update(test_config)
database = create_engine(app.config['DB_URL'], encoding='utf-8', max_overflow=0)
app.database = database
return app
@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.database.execute(text("""
INSERT INTO users(
name,
email,
profile,
hashed_password
) VALUES (
:name,
:email,
:profile,
:password
)
"""), new_user).lastrowid
row = current_app.database.execute(text("""
SELECT
id,
name,
email,
profile
FROM users
WHERE id = :user_id
"""), {
'user_id' : new_user_id
}).fetchone()
created_user = {
'id' : row['id'],
'name' : row['name'],
'email' : row['email'],
'profile' : row['profile']
} if row else None
return jsonify(created_user)
회원가입 테스트
먼저 서버를 띄우고
$ FLASK_ENV=development FKAS_APP=app2.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 "http://localhost:5000/sign-up" \
> name=홍길동 \
> email=gildong@naver.com \
> password=test1234 \
> profile="I am Gildong, nice to meet you"
POST /sign-up HTTP/1.1
Accept: application/json, */*
Accept-Encoding: gzip, deflate
Connection: keep-alive
Content-Length: 129
Content-Type: application/json
Host: localhost:5000
User-Agent: HTTPie/0.9.8
{
"email": "gildong@naver.com",
"name": "홍길동",
"password": "test1234",
"profile": "I am Gildong, nice to meet you"
}
HTTP/1.0 200 OK
Content-Length: 155
Content-Type: application/json
Date: Mon, 23 Mar 2020 06:14:51 GMT
Server: Werkzeug/1.0.0 Python/3.7.2
{
"email": "gildong@naver.com",
"id": 1,
"name": "홍길동",
"password": "test1234",
"profile": "I am Gildong, nice to meet you"
}
그래서 이것 저것 다 고치면 아래와 같은 app3.py 완성
from flask import Flask, request, jsonify, current_app
from flask.json import JSONEncoder
from sqlalchemy import create_engine, text
## Default JSON encoder는 set를 JSON으로 변환할 수 없다.
## 그럼으로 커스텀 엔코더를 작성해서 set을 list로 변환하여
## JSON으로 변환 가능하게 해주어야 한다.
class CustomJSONEncoder(JSONEncoder):
def default(self, obj):
if isinstance(obj, set):
return list(obj)
return JSONEncoder.default(self, obj)
def get_user(user_id):
user = current_app.database.execute(text("""
SELECT
id,
name,
email,
profile
FROM users
WHERE id = :user_id
"""), {
'user_id' : user_id
}).fetchone()
return {
'id' : user['id'],
'name' : user['name'],
'email' : user['email'],
'profile' : user['profile']
} if user else None
def insert_user(user):
return current_app.database.execute(text("""
INSERT INTO users (
name,
email,
profile,
hashed_password
) VALUES (
:name,
:email,
:profile,
:password
)
"""), user).lastrowid
def insert_tweet(user_tweet):
return current_app.database.execute(text("""
INSERT INTO tweets (
user_id,
tweet
) VALUES (
:id,
:tweet
)
"""), user_tweet).rowcount
def insert_follow(user_follow):
return current_app.database.execute(text("""
INSERT INTO users_follow_list (
user_id,
follow_user_id
) VALUES (
:id,
:follow
)
"""), user_follow).rowcount
def insert_unfollow(user_unfollow):
return current_app.database.execute(text("""
DELETE FROM users_follow_list
WHERE user_id = :id
AND follow_user_id = :unfollow
"""), user_unfollow).rowcount
def get_timeline(user_id):
timeline = current_app.database.execute(text("""
SELECT
t.user_id,
t.tweet
FROM tweets t
LEFT JOIN users_follow_list ufl ON ufl.user_id = :user_id
WHERE t.user_id = :user_id
OR t.user_id = ufl.follow_user_id
"""), {
'user_id' : user_id
}).fetchall()
return [{
'user_id' : tweet['user_id'],
'tweet' : tweet['tweet']
} for tweet in timeline]
def create_app(test_config = None):
app = Flask(__name__)
app.json_encoder = CustomJSONEncoder
if test_config is None:
app.config.from_pyfile("config.py")
else:
app.config.update(test_config)
database = create_engine(app.config['DB_URL'], encoding = 'utf-8', max_overflow = 0)
app.database = database
@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 = insert_user(new_user)
new_user = get_user(new_user_id)
return jsonify(new_user)
@app.route('/tweet', methods=['POST'])
def tweet():
user_tweet = request.json
tweet = user_tweet['tweet']
if len(tweet) > 300:
return '300자를 초과했습니다', 400
insert_tweet(user_tweet)
return '', 200
@app.route('/follow', methods=['POST'])
def follow():
payload = request.json
insert_follow(payload)
return '', 200
@app.route('/unfollow', methods=['POST'])
def unfollow():
payload = request.json
insert_unfollow(payload)
return '', 200
@app.route('/timeline/<int:user_id>', methods=['GET'])
def timeline(user_id):
return jsonify({
'user_id' : user_id,
'timeline' : get_timeline(user_id)
})
return app