[Flask, flask-rextx] - REST API 서버 구현, Swagger 문서화

  슬슬 현재 인턴생활도 막바지에 다다랐다. 이 카테고리는 마지막 프로젝트에서 다뤘던 Flask 프레임워크에 대해 기록하고 마무리하려고 한다. 인턴 중반 부까지는 기존에 익숙했던 스프링 프레임워크를 사용하지 않아서, "인기가 없는 프레임워크 배우는 게 상대적으로 손해 같다..."라고 솔직히 생각을 했었다. 근데 해보면 결국 다 비슷한 구조를 가지게 되고 어떤 형식으로 사용하나 정도의 차이였던 것 같다(물론 깊게 파게 된다면 또 다른 느낌일 수도 있지만 말이다). 그리고 기술이란 게 언제 어떻게 흐름이 변할지도 모르는데 하나만 우직하게 파는 것보다는 이것저것 건드려보는 게 it 직군에서는 더 좋은 게 아닐까 지금은 생각하고 있다.

  각설하고, flask에 대한 내용으로 넘어가자. 파이썬의 웹 프레임워크에서 Django 다음으로 잘 사용되는 프레임워크라는 인식이 있는 것 같다. 차이라면 flask는 마이크로 웹 프레임워크이기 때문에 필요에 따라 간결하게 유지하고 확장할 수 있다. 즉 처음부터 웬만한 기능이 갖추어진게 아닌 필요한 상황에 따라 확장 모듈을 통해 기능들을 보완할 수 있는 것이다.

디렉토리 구조

fig 1

  나머지야 부수적인 것들이고, 중요한 부분은 app 패키지이다. 특별히 정해진 형식은 없지만 MVC 형식을 따르려고 해 봤다. View 부분은 프론트에서 별도로 처리하니깐 템플릿이나 static 같은 부분은 없지만 말이다. apps의 각 패키지는 다음과 같은 역할을 한다.

• apis: 클라이언트의 요청을 받고 주요 비즈니스 로직을 처리한 뒤 반환하는 api에 대해 다룬다.
• config: DB, AWS, 애플리케이션 설정 값등에 대한 정보를 다룬다.
• model: 이 프로젝트에서는 SqlAlchemy를 사용하여 db 관련 작업을 수행하기 때문에, 모델별 엔티티 클래스 및 ORM 관련 함수 등을 다룬다.
• utils: 자주 사용될 것 같은 작업에 대한  함수에 대해 다룬다.

 

app(__init__.py)과 애플리케이션 팩토리

from flask import Flask

from flask_migrate import Migrate

from flask_sqlalchemy import SQLAlchemy

from flask_cors import CORS

from flask_restx import Api, Resource

from oauthlib.oauth2 import WebApplicationClient

from flask_jwt_extended import JWTManager

from flask_redis import FlaskRedis

from app.config.flask_config import LocalConfig, DevConfig

db = SQLAlchemy()

migrate = Migrate()

client = WebApplicationClient(DevConfig.GOOGLE_CLIENT_ID)

jwt = JWTManager()

redis = FlaskRedis()

authorizations = {'bearer_auth': {

        'type': 'apiKey',

        'in': 'header',

        'name': 'Authorization'

        }

    }

api = Api(

    version='0.1',

    title="프로젝트 이름",

    terms_url="/",

    authorizations=authorizations,

    security='bearer_auth'

)

def register_router(app: Flask):

    from app.apis import auth_api, container_api, medium_api, event_api, tag_api, script_api

 

    api.add_namespace(auth_api.ns)

    api.add_namespace(container_api.ns)

    api.add_namespace(medium_api.ns)

    api.add_namespace(event_api.ns)

    api.add_namespace(tag_api.ns)

    api.add_namespace(script_api.ns)

# TODO: 환경에 따른 config값 변경

def create_app():

    app = Flask(__name__)

    app.config.from_object(LocalConfig)

    # CORS

    CORS(app)

    # JWT

    jwt.init_app(app)

    # ORM

    db.init_app(app)

    migrate.init_app(app, db)

    from app.model import (

        container,

        container_auth,

        event,

        medium,

        oauth_service,

        tag,

        user

    )

    # Redis

    redis.init_app(app)

    # restx

    api.init_app(app)

    register_router(app)

    return app

  가장 상단에 있는 이 파일에서 애플리케이션 객체가 생성되고 실행된다. 즉 여기서 초기화와 설정 과정이 진행되는 것이다. 여러 flask 튜토리얼에서는 flask 앱을 실행할 때, 해당 파일 마지막에 이런 식의 코드를 넣어 애플리케이션을 실행시킨다.

if __name__ == '__main__':

    app.run()

   하지만 여기서는 create_app()이라는 함수를 통해 애플리케이션 객체를 생성하고 반환하고 있다. 이는 애플리케이션 팩토리 패턴을 사용한 것이다. 해당 패턴의 목표는 애플리케이션 객체를 생성하는 과정을 중앙 집중식으로 관리하여, 모듈화 및 확장성을 증가시키는 것이다. 여기서는 create_app()이 팩토리 함수가 되어 객체 생성을 단순화하고 있다. flask에 익숙치 않아도 create_app()을 통해 애플리케이션 설정을 로드하고, 각 모듈을 초기화하고 라우팅 설정을 진행하는 것을 파악할 수 있다.

  register_router()를 통해 요청에 대한 라우터들을 등록할 수 있다. 이때 apis 모듈의 각 파일에 정의된 ns(nam space)를 등록하고 있는 것을 볼 수 있다. 다시말해 여러 api 요청을 namespace라는 것을 기준으로 하여 분류할 수 있고 이것을 앱 초기화 과정에서 등록해 주는 것이다.

 

flask-rextx

  보통 일반적인 flask에서는 blueprint라는 것을 사용해 모듈화를 돕지만, 이 api 서버는 restful api 서버이기 때문에 본인은 이것에 더 도움이 되는 기능을 갖춘 확장 모듈인 flask-restx의 객체인 namespace를 사용했다. 이러한 점을 포함하여 flask-restx를 사용하면서 다음과 같은 점이 편리하다고 느꼈다.

• 직관적인 구조: 이후에 예시를 들겠지만, blueprint는 요청처리 메소드에 매핑할 url을 매개변수로 하는 route()라는 데코레이터를 붙여 라우팅을 설정하는데, namespace에서는 유사한 route() 데코레이터를 클래스 단위로 붙여주고 각 클래스의 메소드의 이름을 해당 url에 대한 http method로 하여 한 url에 대한 여러 종류의 http request를 처리하는 코드를 좀 더 구조 있게 작성할 수 있다.
• 데이터 검증 및 변환: 요청 파라미터 검증 및 응답 데이터를 직렬화하여 안정적이고 기능적인 API 구현 가능
• Swagger 문서 자동 생성: namespace에서 제공하는 기능을 통해 swagger 문서를 생성할 수 있다. 이것은 뒤에서 자세하게 다룰 예정이다.

 

namespace의 사용

  다음은 한가지 name space 파일의 일부다.

from flask import request, jsonify

from flask_jwt_extended import jwt_required, get_jwt_identity

from app.config.container_config import BUCKET_NAME

from app.utils import container_util, s3_util

from app.model.user import User

from app.model.container import Container

from flask_restx import Namespace, Resource, fields, reqparse

ns = Namespace(

    name='containers',

    description='컨테이너 관련 API'

    )

class _Schema():

    post_fields = ns.model('컨테이너 생성시 필요 데이터', {

        'description': fields.String(description='Container Description', example='Container for tag management'),

        'domain': fields.String(description='Domain of the Container', example='samsung.com')

    })

    basic_fields = ns.model('컨테이너 기본 정보', {

        'name': fields.String(description='Container Name', example='test-container-1'),

        'domain': fields.String(description='Domain of the Container', example='samsung.com')

    })

    detail_fields = ns.inherit('컨테이너 상세 정보', basic_fields, {

        'description': fields.String(description='Container Description', example='Container for tag management')

    })

    msg_fields = ns.model('상태 코드에 따른 설명', {

        'msg': fields.String(description='상태 코드에 대한 메세지', example='처리 내용')

    })

    container_list = fields.List(fields.Nested(basic_fields))

@ns.route('')

class ContainerListOrCreate(Resource):

    @jwt_required()

    @ns.response(200, '컨테이너 리스트 조회 성공', _Schema.container_list)

    def get(self):

        """현재 회원의 컨테이너 리스트를 가져옵니다."""

        user_code = get_jwt_identity()

        containers = User.get_containers(user_code)

        response = [

            {

                "domain": container.domain

            }

            for container in containers

        ]

        return response, 200

    @jwt_required()

    @ns.expect(_Schema.post_fields)

    @ns.response(201, '컨테이너 생성 성공', _Schema.msg_fields)

    @ns.response(400, '컨테이너 생성 실패', _Schema.msg_fields)

    def post(self):

        """새 컨테이너를 추가합니다."""

        user_code = get_jwt_identity()

        body = request.json

        domain = body['domain']

        desc = body['description']

        container = Container.save(domain=domain, description=desc, user_code=user_code)

        if not container:

            return {'msg':'이미 존재하는 도메인입니다.'}, 400

        return {'msg':'ok'}, 201  

우선 ns라는 변수에 Namespace 객체를 생성하여 초기화하고 있다. 따로 'path'라는 파라미터가 없으면 name이 해당 namespace의 prefix가 된다. 그러므로 ContainerListOrCreate라는 클래스는 '/container'라는 경로와 매핑되는 것이다. 그리고 해당 클래스는 flask-restx의 Resource라는 객체를 오버라이딩 하고 있다. 여기서 get, post 같은 http method를 이름으로 하는 메소드를 오버라이딩 할 수 있다. 요청을 보내면 해당 http method 이름에 해당하는 메소드가 해당 요청을 처리하는 것이다. 그리고 따로 직렬화를 해주지 않고 dict 형식으로 반환하여도 flask-restx에서 자동으로 직렬화를 해주기 때문에 dict 형식으로 반환해주고 있다.

 

flask-restx를 이용한 swagger 문서 생성

  swagger 문서를 자동 생성할 수 있다는 점이 restx를 선택한 큰 이유이기도 했다. 아마 spring boot에도 유사한 기능이 있는걸로 알고 있는데 그때는 다른 기능하기에도 벅차서 등한시했었다... 어쨌든 여기서는 어떤 식으로 swagger 문서를 만들 수 있는지 확인해 보자.

Api() (apis의 __init__.py에 위치)

api = Api(

    version='0.1',

    title="API 이름",

    description="api 설명",

    terms_url="/",

    contact="연락처",

    authorizations=authorizations,

    security='bearer_auth'

)

fig 2

  처음 애플리케이션을 초기화할 때 restx의 Api라는 객체를 초기화하면서 해당 API에 대한 소개를 넣을 수 있다. 이름, 설명, 버전, 테스트 시 사용할 보안 설정 등 여러 가지 파라미터가 존재한다.

 

Namespace()

ns = Namespace(

    name='containers',

    description='컨테이너 관련 API'

    )

fig 3

  namespace를 생성할 때 각 namespace에 대한 설명을 넣어줄 수 있다.

 

Resourceful Routing

  어떤 url에 대한 api인지는 위처럼 Namespace의 route() 메소드를 통해 결정할 수 있다. path variable이나 query string은 다음의 방법으로 설정한 뒤 Namespace의 doc() 메소드를 사용하여 각각에 대한 설명을 삽입할 수 있다..

• path variable: <타입명:변수명> 형태로 작성 후 doc()의 params 파라미터로 설명 삽입

@ns.route('/<string:container_domain>')

@ns.doc(params={'container_domain': '컨테이너의 도메인'})

fig 4

• qurey string: doc()의 params 파라미터로 설명 삽입

@ns.route('/platforms')

@ns.doc(params={'container_domain': {'description': '컨테이너 도메인', 'in': 'query', 'type': 'string'},})

fig 5

 

Resource 상속 클래스 메소드에 대한 설명

@ns.route('/containers/<string:container_domain>/mediums/<string:platform_name>')

@ns.doc(params={'container_domain': '컨테이너 도메인', 'platform_name': '플랫폼 이름'})

class MediumManage(Resource):

    @ns.response(200, "매체 정보 조회 성공", _Schema.detail_fields)

    def get(self, container_domain, platform_name):

        """특정 컨테이너의 한 플랫폼에 대한 매체의 상세정보를 가져옵니다"""

        medium = Medium.get_by_container_and_platform(container_domain, platform_name)

        response = {

            'platform_name': PlatformList.get_name(medium.platform_id),

            'base_code': medium.base_code,

            "tracking_list": medium.tracking_list,

            'is_using': medium.is_using

        }

        return response, 200

   

    @ns.expect(200, "새로운 매체 데이터", _Schema.put_fields)

    @ns.response(200, "매체 tracking_list 수정 성공", _Schema.msg_fields)

    def put(self, container_domain, platform_name):

        """특정 컨테이너의 한 플랫폼에 대한 매체의 tracking_id 데이터를 수정합니다"""

        body = request.json

        base_code = body['base_code']

        tracking_list = body['tracking_list']

        medium = Medium.get_by_container_and_platform(container_domain, platform_name)

        medium.update_code_and_tracking_list(base_code, tracking_list)

        return {"msg": "ok"}, 200

   

    @ns.response(200, "매체 데이터 삭제 성공", _Schema.msg_fields)

    def delete(self, container_name, medium_name):

        """특정 컨테이너의 한 플랫폼에 대한 매체의 엔티티를 삭제합니다"""

        Medium.delete_by_container_and_platform(container_name ,medium_name)

        return {"msg": "ok"}, 200

fig 6

  """설명"""를 통해 실제 api 요청을 처리하는 메소드들에 대한 설명을 삽입할 수 있다.

 

Namespace.Model()

class _Schema():

    post_fields = ns.model('컨테이너 생성시 필요 데이터', {

        'description': fields.String(description='Container Description', example='Container for tag management'),

        'domain': fields.String(description='Domain of the Container', example='samsung.com')

    })

    basic_fields = ns.model('컨테이너 기본 정보', {

        'name': fields.String(description='Container Name', example='test-container-1'),

        'domain': fields.String(description='Domain of the Container', example='samsung.com')

    })

    detail_fields = ns.inherit('컨테이너 상세 정보', basic_fields, {

        'description': fields.String(description='Container Description', example='Container for tag management')

    })

    msg_fields = ns.model('상태 코드에 따른 설명', {

        'msg': fields.String(description='상태 코드에 대한 메세지', example='처리 내용')

    })

    container_list = fields.List(fields.Nested(basic_fields))

fig 7

  입출력에 대한 스키마, 즉 예상 요청형식과 응답형식을 등록하는 데 사용된다. 

 

Namespace.expect(), Namespace.response()

  요청/응답이 특정 스키마로 수신/반환되는 걸 기대한다는 것을 알려준다. 위에 있었던 Namespace.model 객체를 등록하면 된다.

   ...

 

    @jwt_required()

    @ns.expect(_Schema.post_fields)

    @ns.response(201, '컨테이너 생성 성공', _Schema.msg_fields)

    @ns.response(400, '컨테이너 생성 실패', _Schema.msg_fields)

    def post(self):

        """새 컨테이너를 추가합니다."""

        user_code = get_jwt_identity()

        body = request.json

        domain = body['domain']

        desc = body['description']

        container = Container.save(domain=domain, description=desc, user_code=user_code)

        if not container:

            return {'msg':'이미 존재하는 도메인입니다.'}, 400

        return {'msg':'ok'}, 201  

fig 8

 

---

참고자료

• https://velog.io/@mstar228/Flask-restx-%EC%82%AC%EC%9A%A9%EB%B2%95#api

Flask-restx 사용법