Python(Flash)APIサーバ

今回はPython言語を用い,FlaskをAPIサーバとして用いたWebフレームワークを用いてREST APIサーバを開発した.

APIサーバ構造

Nginxを使用して複数のクライアントHTTPSを処理する

分散処理のパフォーマンスとセキュリティの問題を解決するために、WebアプリケーションFLASKとWebサーバNginxの間にWebサーバゲートウェイインターフェース(WSDI)ミドルウェアを配置します.

リクエストに応答すると、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)では、フラスコのBlueprintを使用して機能別にHTTP URL分岐処理を行い、ルーティングに対応する機能のURLを置いてルーティング処理を行う.

API Route

API Routeは次のように記述されています.
まず、__init__.pyを以下のように記述し、ルーティングをモジュール別に接続する.

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サービスは以下のように記述されている.

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) :
            ...

サービス作成者がデータベース接続オブジェクトを割り当て、各関数がこれらのオブジェクトを使用してデータベースからデータを取得できるようにします.

API Models

APIモデルは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