We.TIL 41 : Flask로 API만들기 기초

예제로 작성된 코드입니다. 미흡한 점이 있더라도 너른 양해바랍니다.
검색기능을 간단하게 구현했습니다.

Flask란?

마이크로 웹프레임워크, 여러기능이 기본적으로 내장되어있는 풀 프레임워크인 Django와 달리 사용자가 일일히 모든것을 짜줘야한다. 편의성은 떨어지지만 그만큼 유틸성이 극대화 될 수 있으며, 필요한 기능만 구현하면 되어서 좀 더 가볍게 프로젝트를 진행할 수 있다는 장점이 있다.

플라스크로 프로젝트를 진행하면서 프로젝트 구조를 짤 때, 플라스크는 정해진 룰은 없지만 나 같은 경우 Layered Architecture형식을 따랐다.

레이어드 아키텍쳐가 뭔지 궁금하신분들은 이 게시물을 참조하자.
We.TIL 33 : Layered Architecture <- 클릭

구조도

DB와 연결해서 직접 쿼리를 보내는 역할을 하는 모델디렉토리와 연산 등 다양한 로직구현을 하는 서비스 디렉토리, 마지막으로 Request와 Response및 예외처리, 트랜잭션 처리, DB connecttion 처리를 하는 뷰 디렉토리로 구성되어있다.

각각의 디렉토리들에 담겨진 파이썬 파일들은 init.py에 담겨져서 app.py에서 모든 로직이 합쳐지게 되며, app.py는 다시 run.py를 통해 실행되게 된다.

config.py는 DB주소와 비밀번호 등 실제 서버가 구동할 때 필요한 개인적인 요소들이 담긴 파일이며, connection.py 에는 PyMysql을 통해 디비접속시 필요한 메서드가 담겨있는 파일이다.

connection.py

import pymysql

def get_connection(db):
    return pymysql.connect(
        host        = db['host'],
        port        = db['port'],
        user        = db['user'],
        passwd      = db['password'],
        db          = db['database'],
        charset     = db['charset'],
        cursorclass = pymysql.cursors.DictCursor,
        autocommit  = False
    )

config.py에 있는 db정보(여기선 db라는 arg명으로 받았다)를 기반으로
연결을 하게된다. pymysql의 경우 다양한 커서기능이 지원되지만 그 중에서도 이번 프로젝트에선 딕셔너리 형태로 데이터를 담는 경우가 많아 아예 Default를 딕셔너리 커서로 설정했다.

app.py

import config

from flask        import Flask
from flask_cors   import CORS [1]
  
from view         import create_endpoints [2]
from model        import SearchDao
from service      import SearchService

class Services: [3]
    pass

def create_app(test_config = None):
    app = Flask(__name__)
    CORS(app) [4]

    if test_config is None:
        app.config.from_pyfile('config.py')
    else:
        app.config.update(test_config)

    #SetUp Persistence Layer
    search_dao  = SearchDao() [5]
   
    #SetUp Business Layer
    services = Services
    services.search_service  = SearchService(search_dao)[6]
   
    #SetUp Presentation Layer
    create_endpoints(app, services)[7]

    return app

[ 1 ] : 우선 flask_cors를 install하고 import해서 서버를 외부접속이 가능하게 만들어준다. 명령어는 pip install flask-cors 로 실행해주면 된다.

[ 2 ] : 각각의 레이어에 해당하는 디렉토리에서 기능구현을 위해 필요한 파일들은 import한다.

[ 3 ] : 빈 클래스를 만든다. 후에 여러 함수들을 임포트하는 용도다.
빈 List나 Dictionary같은 자료구조 형태에 데이터를 append할 수 있듯이 빈 클래스에 각각의 디렉토리 함수들을 담는 원리다.

[ 4 ] : app객체를 CORS함수로 감싸서 외부접속이 가능한 객체로 바꿔준다

[ 5 ] : 모델디렉토리에서 SearchDao라는 클래스를 객체로 담아준다.

[ 6 ] : 서비스디렉토리에 SearchService 클래스에 search_dao를 파라미터로 담아넣고 이를 다시 아까 생성한 class Service라는 빈 클래스에 넣어준다.

[ 7 ] : 그런다음 엔드포인트에 앱과 Class service를 parameter로 담는다.

이와 같은 프로세스를 통해 각각의 디렉토리의 클래스들을 app.py에서 묶어줄 수 있다.

run.py

from flask_script import Manager

from app         import create_app

app     = create_app()
manager = Manager(app)

@manager.command
def run():
    app.run(
        host='0.0.0.0', 
        port=5000,
        debug=True)

if __name__ == '__main__':
    manager.run()

run.py에서는 호스트, 포트, 디버그 등을 규정해서 실제서버를 실행하는 역할을 한다. flask-script를 설치해서 manager클래스를 활용하면 플라스크앱에 다양한 옵션을 run.py내에서 규정해서 실행할 수 있게된다.

자세한건 https://flask-script.readthedocs.io/en/latest/ 공식문서를 참조하자.

search_view

from flask       import jsonify, request 
from flask.views import MethodView

import config, connection 

class SearchView(MethodView):
    def __init__(self, service): 
        self.service = service

    def get(self):
        try:
            db = connection.get_connection(config.database)
            Q     = request.args.get('q') 
            limit = int(request.args.get('r'))
            search_stores_results   = self.service.search_stores(Q, limit, db)
            search_products_results = self.service.search_products(Q, limit, db)
        
        except:
            return jsonify({'message':'FAILED'}), 400
        
        else:
            return jsonify(
                {
                'stores'   : search_stores_results,
                'products' : search_products_results
                }), 200
        
        finally:
            db.close()

플라스크도 장고와 마찬가지로 클래스 기반 뷰와 함수 기반 뷰를 가질 수 있다.
나 같은 경우에는 클래스 기반 뷰를 작성했다. 쿼리스트링은 requst.GET.get 형태로 받는 Django와 달리 flask에서는request.args.get으로 받을 수 있다.

init.py(search_view)

from .search_view  import SearchView


def create_endpoints(app, services):
    search_service  = services.search_service
   

    app.add_url_rule('/search', view_func = SearchView.as_view('search_results', search_service))

url 주소를 설정해서 라우팅과 연결하는 역할을 해준다.

search_dao

import pymysql

class SearchDao:
    def search_stores(self, Q, limit, db):
        try:
            cursor = db.cursor()
            QEURY = """
            SELECT DISTINCT
                id,
                profile_image,
                korean_name
            FROM
                sellers 
            WHERE
                korean_name LIKE %s
            LIMIT
                %s;
            """
            cursor.execute(QEURY, ('%'+Q+'%', limit))
            result = cursor.fetchall()
        
        except:
            pass

        else:
            return result
        
        finally:
            cursor.close()

    def search_products(self, Q, limit, db):
        try:
            cursor = db.cursor()

            QEURY = """
            SELECT DISTINCT
                s.korean_name as seller_name,
                p.id,
                p.sale_amount,
                pi.image_path,
                pd.name as product_name,
                pd.discount_rate,
                pd.sale_price 
            FROM 
                products p,
                product_images pi,
                product_details pd,
                sellers s
            WHERE 
                s.korean_name LIKE %s  
                OR pd.name LIKE %s 
                AND p.seller_id = s.id
                AND p.id = pi.product_id
                AND p.id = pd.product_id
                AND pi.ordering = 1
            LIMIT
                %s;
            """
            cursor.execute(QEURY, ('%'+Q+'%', '%'+Q+'%', limit))
            result = cursor.fetchall()
            
        
        except:
            pass
        
        else:
            return result
        
        finally:
            cursor.close()

pymysql은 cursor를 통해 db에 쿼리를 보내고 결과물을 뽑아낼 수 있다.
execute method와 fetch method를 통해 결과물을 담아 서비스로 보내게된다.

init.py(search_dao)

from .search_dao  import SearchDao


__all__ = [
    SearchDao
]

실제로는 SearchDao 말고 다양한 model파일이 누적되게 되므로 init.py를 통해 패키지화 시켜주면 좋다.

search_service

class SearchService:
    def __init__(self, search_dao, product_dao):
        self.search_dao = search_dao
        self.product_dao = product_dao

    def search_stores(self, Q, limit, db):
        try:
            search_stores_results = self.search_dao.search_stores(Q, limit, db)
       
        except:
            raise
       
        else:
            return search_stores_results

    def search_products(self, Q, limit, db):
        try:
            products_data = self.search_dao.search_products(Q, limit, db)
            search_products_results  = [{
                'id'               : product['id'],
                'image_url'        : product['image_path'], 
                'seller_name'      : product['seller_name'], 
                'product_name'     : product['product_name'], 
                'discount_rate'    : product['discount_rate'],
                'origin_price'     : product['sale_price'], 
                'discounted_price' : (lambda x,y : int(x*((100-y)/100)))(product['sale_price'],product['discount_rate']),
                'sale_amount'      : product['sale_amount']
            } for product in products_data] 
        
        except:
            raise
        
        else:
            return search_products_results

search_dao를 통해 얻어낸 result결과값을 실제 서버에 보낼때 필요한 데이터형태로 바꿔준다.

init.py(search_view)

from .search_service  import SearchService

__all__ = [
    SearchService
]

서치서비스 말고 다양한 서비스파일이 누적되게 되므로 init.py를 통해 패키지화 시켜주면 좋다.

모든 레이어의 내용이 작성됬다면 python run.py run 명령어를 통해 서버를 실행해 실제 기능을 동작하게 만들면 된다.