Azure Function으로 RestAPI 간단하게 만들기

Azure Function, MSSQL 로 간단한 Http Trigger function application을 만들어보자

VS 에서 Function APP을(advanced로 만들고) requirements.txt에 필요한 라이브러리들을 설정한다.

# DO NOT include azure-functions-worker in this file
# The Python Worker is managed by Azure Functions platform
# Manually managing azure-functions-worker may cause unexpected issues

azure-functions
pymssql
sqlalchemy
pytz
azure-identity
pandas

백앤드 SQL server (Azure SQL과 연동을 하기 위한) 연동을 위한 매핑 클래스와 쿼리 메소드 (test api)

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, SMALLINT, BIGINT, String, Float, DateTime, INT
from sqlalchemy import sql
from sqlalchemy import *
from hd_dtw_prod_func_restapi.funcconfig import funcconfig
from sqlalchemy.orm import sessionmaker 
from sqlalchemy.pool import NullPool

funcconfig = funcconfig()

# 매핑 선언
Base = declarative_base()


# Mapping 클래스 생성
# ORM
class testtable(Base):
    # 식별하기 쉽게 테이블명으로 지정
    __tablename__ = 'testtable'

    id = Column(String)
    c1 = Column(INT)
    c2 = Column(INT)
    c3 = Column(INT)
    c4 = Column(INT)
    c5 = Column(INT)
    
    def __init__(self, id, c1, c2, c3, c4, c5):
        self.id = id
        self.c1=c1
        self.c2=c2
        self.c3=c3
        self.c4=c4
        self.c5=c5
    
def test_api(idd):
    import pandas as pd

    try:
        db_engine = create_engine('mssql+pymssql://{}:{}@{}/{}'.format(funcconfig.MSSQL_USER, funcconfig.MSSQL_PASSWORD,
                                  funcconfig.MSSQL_HOST, funcconfig.MSSQL_DATABASE), echo=True, poolclass=NullPool)
        Session = sessionmaker(bind=db_engine)
        session = Session()
        session.commit()
        
    except:
        session.rollback()
        raise

    finally:

        df = pd.read_sql(session.query(testtable).filter_by(id = idd).statement, session.bind)
        import json
    session.close()
    return df

최종적으로는 main.py 에 간단한 http request에 대한 걸과를 반환해주는 어플리케이션이다

import logging
from sqlalchemy.orm import query_expression
from sqlalchemy.sql.elements import Null
import azure.functions as func
import json
from datetime import datetime, timedelta
import random
import logging
import os 
import sys
from funcapi_test_HttpTrigger1 import mapper
import pytz
import datetime as dt
from funcapi_test_HttpTrigger1.funcconfig import funcconfig

funcconfig = funcconfig()


def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info("Function HTTP Trigger for Restful API Service")

    id = req.params.get('id')
    if not id:
        try:
            req_body = req.get_json()
        except ValueError:
            pass
    id_list = ['a', 'b', 'c']

    if id not in id_list:
        logging.info("invalid id")
        return func.HttpResponse("{} is an invalid ID".format(id), status_code = 400)

    else:
        query_result = mapper.test_api(idd = id)
    

        return func.HttpResponse(body = json.dumps({"results": {"id":  str(query_result["id"]),
                                                                "c1":  float(query_result["c1"]),
                                                                "c2":  float(query_result["c2"]),
                                                                "c3":  float(query_result["c3"]),
                                                                "c4":  float(query_result["c4"]),
                                                                "c5":  float(query_result["c5"])}}))