Flask REST API Server w gunicorn — #1. 개발과정

20 min readMar 10, 2024

REST API Server 개발과 도커라이징, 쿠버네티스 배포까지

Flask Framework를 사용해서 REST API Server를 개발하고 gunicorn으로 구동시킨 다음, 도커라이징과 쿠버네티스 배포까지 진행해보는 토이프로젝트를 해봤습니다. API의 기능은 축구선수의 골 개수를 저장하고 조회하는 기능으로 준비했습니다.

REST API Server 구현이 많이 어설프지만, 개발과 배포 사이클의 전반적인 과정을 경험하고 배우기 위한 목적이 있는 토이프로젝트인 점을 고려해서 읽어주시면 감사하겠습니다.

개발 환경

m1 macbook air 13.4
vscode 1.86.0
docker desktop v20.10.22

언어 및 버전

Language
Python 3.9.18
Python Package
Flask==2.3.3
mysql-connector-python==8.3.0
gunicorn==21.2.0
MySQL 8.2.0

1. 데이터베이스 선택

CRUD 연산을 위해 가장먼저 데이터베이스를 선택하는것부터 고민을 시작했습니다.
파이썬 라이브러리로 제공되는 SQLite를 사용하면 간단하게 구현이 가능했지만, gunicorn을 사용한 Multi Worker 환경에서는, 각 Worker Process 마다 메모리 영역에서 SQLite 데이터베이스를 사용 하기 때문에 CRUD 연산이 제각각 이루어 졌습니다. 따라서, Application과 독립적으로 구동되는 MySQL 서버를 Container로 구동 시키기로 했습니다.

서버를 정한 다음에는 데이터베이스와 테이블을 미리 준비하는 작업이 필요했습니다.
우선, MySQL 컨테이너 구동 시 환경변수(ENV) 주입을 통해서 데이터베이스 생성과 Application에서 사용할 데이터베이스 접속용도 사용자와 비밀번호를 생성하도록 했습니다. 그리고 테이블을 생성하고 초기 값을 넣어주기 위해 .sql 스크립트 파일을 작성해서 MySQL Docker Image의 Entrypoint에 옮겨두는 작업을 했습니다. 본 작업을 위해 Dockerfile을 작성했습니다.

*Table 이름은 .sql 스크립트에 먼저 선언되고, Application Code 에서 동일한 이름을 사용하도록 했습니다.

요약하면, Docker 환경변수로 데이터베이스와 사용자,비밀번호를 생성하고 .sql 스크립트로 테이블을 생성과 값을 입력하여 데이터베이스 초기 세팅을 완료하게 됩니다.

scheme.sql

DROP TABLE IF EXISTS player;

CREATE TABLE player (
  id INT AUTO_INCREMENT PRIMARY KEY,
  created DATETIME DEFAULT CURRENT_TIMESTAMP,
  name VARCHAR(50) UNIQUE,
  team VARCHAR(100),
  goal INT
);

INSERT INTO player (name,team,goal)
VALUES ('sonny','spurs',10);

INSERT INTO player (name,team,goal)
VALUES ('kane','spurs',5);

INSERT INTO player (name,team,goal)
VALUES ('hyukjun','spurs',1);

mysql Dockerfile

# 베이스이미지
FROM mysql:8.2.0

# 데이터베이스 셋업
ADD scheme.sql /docker-entrypoint-initdb.d

# MySQL 포트 명시
EXPOSE 3306

2. Flask Framework를 사용한 API Server 개발

REST API Server를 구현하기 위해 Python Framework 중 현시점 가장 친숙한 Flask를 선택했고, Flask에서 사용할 수 있는 다양한 API 설계 관련 라이브러리는 되도록 사용하지 않았습니다. REST API Server를 제대로 설계해본 경험이 없으므로, 기본적인 코드를 사용해서 API에 대한 이해를 높히고자 했습니다. 그리고 현시점 에는 ORM사용보다는 SQL 문을 그대로 코드에서 사용하는 것이 배우는게 더 많을것 같았습니다.

2.1 Context 주입

첫번째로 고민한 것은 Database에 대한 Context를 Application에 어떻게 주입하는가 입니다. 저는 Application 코드 내에 직접 Context 작성하는것은 보안적으로 좋지 않다고 생각하여 Application이 구동되는 OS(로컬 and 컨테이너 이미지)에 환경변수로 선언해두고 가져오는 방법을 선택했습니다. (도커라이징을 고려해봐도 이 방법이 좋은 선택인것 같습니다.)

# 데이터베이스 연결정보 세팅
MYSQL_DATABASE_USER = os.environ.get('MYSQL_DATABASE_USER')
MYSQL_DATABASE_PASSWORD = os.environ.get('MYSQL_DATABASE_PASSWORD')
MYSQL_DATABASE_DB = os.environ.get('MYSQL_DATABASE_DB')
MYSQL_DATABASE_HOST = os.environ.get('MYSQL_DATABASE_HOST')

# 데이터베이스의 테이블 이름
TABLE_NAME="player"

2.2 MySQL Connector

Application에서 사용할 MySQL Connector를 선택하기 위해 다양한 Connector를 살펴보았지만 기능적인 면은 차이를 못느꼈습니다. 따라서, 공식 문서에 자세한 가이드가 나와있는 mysql.connector 패키지를 사용하였습니다. Connector는 앞서 주입한 Database Context를 사용해서 Database에대한 Connection을 생성합니다.

그리고 코드를 작성하다 보니 Connector 사용이 빈번해지면서 중복된 코드가 많아 짐에 따라 Connector를 호출하는 부분을 Function으로 분리했습니다.

글을 쓰는 시점에서도 계속 고민되는 부분은, Connection을 사용한 CRUD 연산이 끝난 뒤에는 매번 Connection을 닫아주면서 불필요한 메모리 사용을 줄일수 있다고 생각하지만, 프로덕션 환경에서는 매번 Connection을 닫기 보다는 Connection Pool을 사용해서 매번 Connection을 맺기 위한 리소스 소모를 방지하는 방향이 더 좋지 않을까 생각하고 있습니다. 이부분은 아직 정확한 개념을 알고있지 않기에 지금은 커넥션을 잘 닫아주는 쪽으로 구현했습니다.

# 데이터베이스 커넥션 획득을 위한 함수
def get_db_connection():
    connection = mysql.connector.connect(
        host=MYSQL_DATABASE_HOST,
        database=MYSQL_DATABASE_DB,
        user=MYSQL_DATABASE_USER,
        password=MYSQL_DATABASE_PASSWORD
    )
    cursor = connection.cursor()
    return connection, cursor

2.3 CRUD 구현

이제 가장 기본적인 GET, POST, PUT, DELETE 4가지 Method를 구현하기 위해 각각 별도의 함수를 작성했습니다.

처음에 가장 궁금했던 것은 @app.route() 모양의 데코레이터(Decorator) 함수였습니다. 여러가지 문서를 살펴본 바 아래처럼 이해하고 설계를 이어 갔습니다.

반복적으로 실행해야 하는 함수를 데코레이터 함수로 만들어 놓고, 그아래 위치한 함수는 데코레이터 함수의 인자값으로 들어갑니다. 데코레이터 함수 내에서는 넘겨진 함수를 정해진 위치에서 실행하게 됩니다. 즉, 특정 함수를 어떤 외부 함수의 내부에서 실행될 수 있도록 하는 기능으로 이해했습니다. (외부함수와 내부함수의 관계)

Method 별로 함수를 작성하였고, 각각 try-except 조건을 사용한 에러 처리와 Method에 맞는 적절한 SQL문, 그리고 필요한 Return 값을 반환하는 코드를 구현했습니다.

GET Method는 전체 Player를 조회해서 JSON 타입으로 데이터를 반환하는 함수를 작성했습니다.

POST Method는 1명의 선수를 등록하는 작업이 가능 합니다. 이를 위해 POST는 Body에서 새로운 Data를 받습니다.

PUT Method는 URL Path 에서 업데이트할 선수의 ID를 받도록 하고, 해당 ID를 가진 선수가 존재하는지, 존재한다면 Body에 실려온 Data를 토대로 선수 상태를 업데이트 하도록 설계했습니다.

Delete Method도 URL Path에서 삭제할 선수의 ID를 받도록 했으며, 삭제 전에 해당 ID의 선수 정보가 데이터베이스에 존재하는지 검사하는 조건을 넣었습니다. 조건에 만족한다면, 선수 정보는 데이터베이스에서 삭제됩니다.

# GET
@app.route("/api/player", methods=['GET'])
def get_player():
    try:
        connection, cursor = get_db_connection()
    except Exception as e:
        print(e)
        return "Failed to get DB Connection."
    try:
        select_query = f"""
            SELECT
                JSON_OBJECT (
                    'id', id,
                    'created', created,
                    'name', name,
                    'team', team,
                    'goal', goal
                )
            FROM {TABLE_NAME}
        """
        cursor.execute(select_query)
        data = cursor.fetchall()
        connection.close()
        return jsonify(data), 200
    except Exception as e:
         print(e)
         return "Fail to get player's info.", 400

# POST
@app.route("/api/player", methods=['POST'])
def create_player():
        try:
            connection, cursor = get_db_connection()
        except Exception as e:
            print(e)
            return "Get DB Connection Failed."
        try:
            body = request.get_data()
            data = json.loads(body)
            player_name = data['name']
            player_team = data['team']
            player_goal = data['goal']

            select_query = f"""
                SELECT name FROM {TABLE_NAME} 
                WHERE name = %s
                """
            # Check target exist, 변수는 리스트형으로 (혹은 튜플,딕셔너리)
            cursor.execute(select_query, [player_name])            
            result = cursor.fetchone()
            if result is not None:
                connection.close()
                return "Player already exist.", 400
            else:
                insert_query = f"""
                    INSERT INTO {TABLE_NAME}
                    (name, team, goal)
                    VALUES (%s, %s, %s)
                """
                cursor.execute(insert_query, [player_name, player_team, player_goal])
                connection.commit()
                connection.close()
                return "Completed to create player.", 201
        except Exception as e:
            print(e)
            return "bad request", 400

# PUT
@app.route("/api/player/<int:id>", methods=['PUT'])
def update_player(id):
    try:
        connection, cursor = get_db_connection()
    except Exception as e:
        print(e)
        return "Get DB Connection Failed."
    try:
        # Get Data
        body = request.get_data()
        data = json.loads(body)
        player_name = data['name']
        player_team = data['team']
        player_goal = data['goal']

        # Check target exist
        select_query = f"""
            SELECT id FROM {TABLE_NAME} 
            WHERE id=%s
        """
        cursor.execute(select_query, [id]
        )
        result = cursor.fetchone()

        if result is None:
            connection.close()
            return "Player don't exist.", 400
        else:
            update_query = f"""
                UPDATE player
                SET name=%(name)s, team=%(team)s, goal=%(goal)s
                WHERE id=%(id)s
            """
            cursor.execute(update_query, 
                           {
                            'name': player_name,
                            'team': player_team,
                            'goal': player_goal,
                            'id': id
                            }
                           )
            connection.commit()
            connection.close()
            return "Updated Player.", 200
    except Exception as e:
        print(e)
        return "bad request", 400

# DELETE
@app.route("/api/player/<int:id>", methods=['DELETE'])
def delete_player(id):
    try:
        connection, cursor = get_db_connection()
    except Exception as e:
        print(e)
        return "Get DB Connection Failed."
    try:

        # Check target exist
        select_query = f"""
            SELECT id FROM {TABLE_NAME}
            WHERE id=%s
        """
        cursor.execute(select_query, [id])
        result = cursor.fetchone()

        if result is None:
            connection.close()
            return "Player don't exist.", 400
        else:
            delete_query = f"""
                DELETE FROM {TABLE_NAME}
                WHERE id=%s
            """
            cursor.execute(delete_query, [id])
            connection.commit()
            connection.close()
            return "Deleted Player.", 200
    except Exception as e:
        print(e)
        return "bad request", 400

마지막으로 Health Check 함수를 작성해서 향후 쿠버네티스와 같은 오케스트레이션 툴에 배포될 경우 Probe가 바라보는 Endpoint 로 사용되도록 했으며, 마지막으로, 로컬 실행과 gunicorn을 사용한 실행에 차이를 두기 위해 Application 실행 부분에 분기처리를 추가하면서 코드 설계를 마무리 했습니다.

# Health Check 
@app.route('/healthz', methods=['GET'])
def healthz():
    return {"health": "ok"}, 200 # 200 Ok

# 실행 방법에 따라 구동 형태가 달라지도록 분기처리
if __name__ == "__main__":
    app.run(host='0.0.0.0',port=8000,debug=True)
else:
    gunicorn_logger = logging.getLogger('gunicorn.error')
    app.logger.handlers = gunicorn_logger.handlers
    app.logger.setLevel(gunicorn_logger.level)

2.4 WSGI — Gunicorn

Flask는 내장된 WSGI로 자체적인 실행이 되지만, 안정성과 성능을 고려하면 별도의 WSGI 서버를 두는것이 적합하다고 판단했습니다. (Flask를 단독으로 실행해보면, 항상 프로덕션에서는 단독으로 사용하지 말라는 경고문이 발생합니다.)

그리고 토이프로젝트의 목적이 프로덕션 환경을 간접 체험하기 위한 것이기 때문에, WSGI는 필수로 사용하고자 했습니다.

*참고) 저는 WSGI 개념을 아래 처럼 이해하고 있습니다.

WSGI는 Python으로 작성된 Web Application을 위한 Middleware 로서 Web Server와의 통신을 위한 인터페이스 역할을 수행한다.

우선, 사용할 WSGI로 gunicorn을 선택했으며, 패키지를 설치하고 gunicorn_config.py 파일에 필요한 설정을 작성했습니다.

주요 설정으로는 로그 출력 설정, Worker Process와 Thread의 개수, Worker 당 처리할 수 있는 요청의 최대값, 그리고 Timeout 시간을 지정해 줬습니다. 이외에도 다양한 설정 옵션이 존재하며, API서버가 필요한 스펙에 따라 각 값을 적절히 조절해주면 좋을것 같습니다.

import multiprocessing

# 로그 스트림 출력
accesslog = '-'
errorlog = '_'

# Worker and Thread
# 운영 환경에 권장되는 설정은 cpu 코어 개수 * 2 + 1 개가 권장되는것 같습니다.
# workers = multiprocessing.cpu_count() * 2 + 1 
workers = 4
threads = 4

# Worker 당 최대 요청 처리 개수 지정
max_requests = 5000

# Socket 타임아웃(초) 지정
timeout = 120
bind = '0.0.0.0:8000'

gunicorn의 설정은 아래 블로그와 문서를 통해 좋은 정보를 얻을 수 있었습니다.

gunicorn 설정의 A to Z

Settings — Gunicorn 21.2.0 documentation

3. 로컬환경에서 서버 실행과 동작 검증

이제 개발한 서버를 구동해보고 예상대로 동작하는지 확인해보겠습니다.

먼저 MySQL 데이터베이스를 Docker Container로 빌드하고 실행시킵니다.

# Docker build
docker build -t mysql-flask-api:1.0 .

# Docker run
docker run -d -p 3306:3306 \
--env MYSQL_DATABASE=myplayer \
--env MYSQL_ROOT_PASSWORD=rootpasswd \
--env MYSQL_USER=myadmin \
--env MYSQL_PASSWORD=mypass \
mysql-flask-api:1.0

그 다음 환경변수를 설정하고, Flask API 서버를 실행 합니다. 4개의 Worker Process가 실행된것을 확인할 수 있습니다.

# 가상환경 준비
source .venv/bin/activate
python -m pip install -r requirements.txt

# 환경 변수 주입
export MYSQL_DATABASE_HOST=localhost
export MYSQL_DATABASE_DB=myplayer
export MYSQL_DATABASE_USER=myadmin
export MYSQL_DATABASE_PASSWORD=mypass

# gunicorn으로 실행
gunicorn -c gunicorn_config.py main:app
...
[2024-03-10 19:24:38 +0900] [58425] [INFO] Starting gunicorn 21.2.0
[2024-03-10 19:24:38 +0900] [58425] [INFO] Listening at: http://0.0.0.0:8000 (58425)
[2024-03-10 19:24:38 +0900] [58425] [INFO] Using worker: gthread
[2024-03-10 19:24:38 +0900] [58426] [INFO] Booting worker with pid: 58426
[2024-03-10 19:24:38 +0900] [58427] [INFO] Booting worker with pid: 58427
[2024-03-10 19:24:38 +0900] [58428] [INFO] Booting worker with pid: 58428
[2024-03-10 19:24:39 +0900] [58429] [INFO] Booting worker with pid: 58429
...