Python(Flask) API 서버

1

Python Api Server

목록 보기
2/2
post-thumbnail

이번에는 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
   
profile
rs = woo + sun; console.log("♥" + rs);

0개의 댓글