이번에는 Python 언어를 사용, 그리고 API 서버로 사용하기 좋은 웹 프레임 워크로 Flask 를 사용하여 REST API 서버를 개발한 내용이다.
API 서버 구조
-
다수의 클라이언트 HTTPS 처리를 위해 Nginx 를 사용
-
웹 어플리케이션인 FLASK 와 웹 서버인 Nginx 사이에 WSGI(Web Server GateWay Interface) 미들웨어를 두어 분산처리 성능과 보안 문제를 해결 한다.
-
요청에 대한 응답 시 Flask 는 DB 와 연동 하여 데이터를 가져오도록 한다.
API Main
API Main 은 다음과 같이 작성하였다.
import os
from dotenv import load_dotenv
from flask import Flask
...
app.register_blueprint(customer_v1, url_prefix='/customer/v1')
if __name__ == "__main__":
app.run(host=os.getenv("WEB_HOST"),port=os.getenv("WEB_PORT"))
- CORS 처리는 기본 항목이다.
- Main 이 되는(app.py) 에서는 flask 의 blueprint 를 이용, 기능 별 HTTP URL 분기 처리를 해주고 routes 에서 해당 기능 별 url 을 두어 라우팅 처리를 하였다.
API Route
API Route 는 다음과 같이 작성 하였다.
먼저 __init__.py 를 아래와 같이 작성하여 route 를 모듈 별로 연결 해 주었다.
from flask import Blueprint
customer_v1 = Blueprint('customer_v1_routes', __name__)
from .customer_v1_routes import *동일 디렉토리에 customer_v1_routes.py 모듈을 작성 라우팅 처리를 해주었다.
from . import customer_v1
from services.customer_v1_service import CustomerService
from flask import request, Response
from util.print_util import *
import json
# Get Login Auth
@customer_v1.route('/auth', methods = ['POST'])
def loginAuth():
user_info = request.get_json()
if (user_info is not None) :
if (user_info['username'] is None) :
...
customer = CustomerService()
result = customer.create_jwt_token(user_info['username'], user_info['password'])
...
# Get Customer Stats
@customer_v1.route('/stats_customer', methods = ['GET'])
def getCustomerStats() :
query_dict = request.args.to_dict()
...
# Get Customer Msg_History
@customer_v1.route('/msg_histories', methods = ['GET'])
def getMsgHistory() :
query_dict = request.args.to_dict()
...
이렇게 작성하게 될 경우 요청 로그인 Auth URL 은 /customer/v1/auth 가 된다.
API Service
주요 처리 기능인 API Service 는 다음과 같이 작성 하였다.
from models.customer_v1 import CustomerV1
from util.altibase import InitAltibase
from dotenv import load_dotenv
from datetime import datetime, timedelta
import jwt, os, json
class CustomerService :
def __init__(self):
# 알티베이스 연결 객체 생성
self.db_conn = InitAltibase()
# 서비스 객체에 연결 객체 전달
self.db = CustomerV1(self.db_conn.get_conn())
def create_jwt_token(self, username, password):
load_dotenv()
# DB 데이터 요청
user_info = self.db.get_login_info(username, password)
res_obj = dict()
...
def get_stats_data(self, query_dict):
count = self.db.get_stats_count(query_dict, False)
if (count is not None) :
...
# 메시지 이력 데이터 조회
def get_msghist_data(self, query_dict):
count = self.db.get_msghist_count(query_dict, False)
if (count is not None) :
...
서비스 생성자에서 DB 접속 객체를 할당 하도록 하고, 각 함수에서 해당 객체를 이용하여 DB 에서 데이터를 가져올 수 있도록 하였다.
API Models
API Model 은 DB 에서 데이터를 가져오긴 쿼리문들로 되어있다.
import os
import jaydebeapi as jp
from dotenv import load_dotenv
from util.query_util import set_query_sort_by, get_query_month_list
import json
class CustomerV1:
def __init__(self, conn):
self.conn = conn
def get_login_info(self, username, passwd) :
sql = " SELECT SUBS_ID, LOGIN_ID, CREATE_DATE, ...
result = None
with self.conn.cursor() as curs:
try :
curs.execute(sql)
result = curs.fetchall()
return result[0]
except Exception as msg :
return result
def get_stats_count(self, query_dict, one=False) :
sql = "SELECT COUNT(*) as Count FROM ( SELECT RA...
result = None
with self.conn.cursor() as curs:
try :
curs.execute(sql)
result = [dict((curs.description[i][0], str(value)) \
for i, value in enumerate(row)) for row in curs.fetchall()]
return (result[0] if result else None) if one else result
except Exception as msg :
return result
def get_stats_list(self, query_dict, one=False) :
...
sql = " SELECT RAISEDATE as raisedate, MSG_TYPE as msg_serv...
with self.conn.cursor() as curs:
try :
curs.execute(sql)
result = [dict((curs.description[i][0], str(value)) \
for i, value in enumerate(row)) for row in curs.fetchall()]
return (result[0] if result else None) if one else result
except Exception as msg :
return result
def get_msghist_count(self, query_dict, one=False) :
...
result = None
with self.conn.cursor() as curs:
try :
curs.execute(sql)
result = [dict((curs.description[i][0], str(value)) \
for i, value in enumerate(row)) for row in curs.fetchall()]
return (result[0] if result else None) if one else result
except Exception as msg :
return result
def get_msghist_list(self, query_dict, one=False) :
...
with self.conn.cursor() as curs:
try :
curs.execute(sql)
result = [dict((curs.description[i][0], str(value)) \
for i, value in enumerate(row)) for row in curs.fetchall()]
return (result[0] if result else None) if one else result
except Exception as msg :
return result
API Utils.. Altibase
해당 프로젝트에서 Altibase DB 와 연동이 필요 하여 JDBC 를 이용, Altibase 모듈을 별도 아래처럼 만들어 사용 하였다.
import os
import jaydebeapi as jp
from dotenv import load_dotenv
class InitAltibase:
def __init__(self):
load_dotenv()
self.host = os.getenv("ALTIBASE_HOST")
self.port = os.getenv("ALTIBASE_PORT")
self.id = os.getenv("ALTIBASE_USER")
self.password = os.getenv("ALTIBASE_PASSWORD")
self.mydb = os.getenv("ALTIBASE_DB")
# DB Connect
self.conn = jp.connect('Altibase.jdbc.driver.AltibaseDriver',
'jdbc:Altibase://'+self.host+':'+self.port+'/'+self.mydb+'',
driver_args={"user":self.id,"password":self.password},
jars = "./lib/Altibase.jar")
def get_conn(self):
return self.conn
rs = woo + sun; console.log("♥" + rs);