Python 웹 프레임워크 만들기 - WSGI

이 글은 Python 웹 프레임워크의 내부 동작을 이해하고자 Flask의 일부 기능을 간략하게 설명합니다. 세부적인 구현은 포함되어 있지 않습니다.

 

서버단에서 웹 어플리케이션을 개발한다면 대부분의 경우 웹 프레임워크를 사용합니다. 웹 어플리케이션 개발에 필요한 여러 작업들을 처리해주어 개발자는 자신이 구현하고자 하는 비즈니스 로직에 집중할 수 있기 때문입니다. 웹 프레임워크가 수행하는 기능을 살펴보고 간단히 구현해보겠습니다.

우선 웹 프레임워크를 이해하기 위해선 알아둬야할 개념들이 있습니다. 정적 컨텐츠, 동적 컨텐츠, 웹 서버, CGI, WAS, WSGI입니다.

---

기본 개념

정적 컨텐츠

html, css, 이미지등과 같이 서버에 미리 준비되어 있는 파일 형태의 컨텐츠를 의미합니다.

 

동적 컨텐츠

사용자/상황에 따라 결과 값이 다른 컨텐츠를 의미합니다.

 

웹 서버

사용자가 요청하면 정적 컨텐츠를 제공합니다.

web server 동작방식

 

CGI

동적 컨텐츠를 제공하기 위해 웹서버 상에서 프로그램을 동작시키기 위한 인터페이스입니다. 웹 서버는 사용자의 동적 요청이 발생하면 CGI 규격을 준수한 프로그램을 실행시키고 결과를 반환합니다.

GCI 동작방식

CGI는 매 요청마다 process/thread를 생성하여 프로그램을 실행합니다. 그렇기 때문에 process/thread 생성, 프로그램 실행과 같은 추가적인 시간 소요가 발생하고 요청이 많아질수록 시스템에 부하를 줍니다.

 

WAS

동적 컨텐츠를 제공하기 위해 만들어진 Web Application Server입니다. 웹 서버 + CGI의 느낌이지만 CGI와 달리 데몬으로 미리 실행되어 있어 CGI의 단점을 보완됩니다. HTTP 서버이므로 웹 서버 없이도 동작이 가능하지만 보안, 안정성, 성능 등을 이유로 웹 서버와 같이 사용됩니다.

 

WSGI

Web Server Gateway Interface의 약자로 Python에서 WAS를 구현하기 위한 인터페이스(미들웨어)입니다. 어플리케이션을 감싸는 컨테이너로 동작 하여 웹 서버와 어플리케이션간 통신이 가능하게 합니다. Flask, Django와 같은 Python 웹 프레임 워크는 WSGI 규격을 준수하여 gunicon, uwsgi 등의 WSGI로 실행할 수 있습니다. 그래서 실제 프로덕션 환경에서는 웹 서버 <--> WSGI <--> 어플리케이션으로 구성합니다.

 

 

WSGI 규격

위에서 WSGI는 어플리케이션을 감싸는 컨테이너라고 했습니다. 요청이 들어오면 WSGI는 어플리케이션을 호출하고 결과값을 반환합니다. WSGI에서 어플리케이션이 동작하기 위해서는 어플리케이션은 다음과 같은 규칙을 지켜야합니다.

1. Callable 객체여야 한다.

2. environ, start_response(콜백 함수)라는 2개의 파라미터를 받을 수 있어야 한다.

3. Callable 객체의 반환 값으로 Iterable 객체가 반환된다.

4. WSGI는 Iterable 객체를 호출하여 비즈니스로직의 결과값을 받는다.

코드로 살펴보면 다음과 같습니다.

class FakeFlask:
    def __call__(self, environ: Dict[str, str], start_response: Callable[[str, List[Tuple[str, str]]], None]) -> "FakeFlask":
        self.environ = environ
        self.start_response = start_response
        return self

    def __iter__(self) -> Iterable[bytes]:
        response = "hello world"
        status_code = "200 OK"
        response_headers = [("Content-Type", "text/plain")]
        self.start_response(status_code, response_headers)
        yield response.encode("utf-8")

WSGI로 실행해봅니다. 이 글에서는 uwsgi를 사용했습니다. 설치가 되어있지 않다면 pip install uwsgi로 설치가 필요합니다.

uwsgi --http :9000 --wsgi-file {파일명}.py

uwsgi가 실행되고 curl로 요청하면 기대대로 응답합니다.

curl localhost:9000
>>> hello world%

이렇게 WSGI 규격을 준수하는 어플리케이션을 만들 수 있습니다.

 

 

웹 프레임워크

WSGI 규격을 알게 되었습니다. 이제 Flask를 만들어 보겠습니다.

URL 라우팅

우리가 알고 있는 Flask는 URL 라우팅을 제공합니다. URL을 비즈니스 로직 함수에 매핑하여 요청이 들어왔을 때 URL에 맞는 적절한 함수를 호출합니다.

아래와 같이 데코레이터를 사용해서 URL과 함수를 매핑합니다.

app = FakeFlask()

@app.route("/")
def home():
    return "hello world"

URL 라우팅과 데코레이터를 구현하면 다음과 같습니다.

class FakeFlask:
    def __init__(self):
        self.__route: Dict[str, Callable] = {}
	
    def __call__(self, environ: Dict[str, str], start_response: Callable[[str, List[Tuple[str, str]]], None]) -> "FakeFlask":
        self.environ = environ
        self.start_response = start_response
        return self
    
    def __iter__(self) -> Iterable[bytes]:
        endpoint, query = self.__parse_url()
        controller = self.__get_controller(endpoint=endpoint)

    def route(self, endpoint: str) -> Callable:
        def controller(func):
            self.__route[endpoint] = func
        return controller

    def __parse_url(self) -> Tuple[str, str]:
        endpoint = self.environ["PATH_INFO"]
        query = self.environ["QUERY_STRING"]
        if query:
            query = {key: value for key, value in [u.split("=") for u in query.split("&")]}
        return endpoint, query
        
    def __get_controller(self, endpoint: str) -> Optional[Callable]:
        if endpoint in self.__route:
            return self.__route[endpoint]
        return None

route 메소드를 데코레이터로 만듭니다. 데코레이터 파라미터로 엔드포인트를 받도록하여 클래스 내부에 엔드포인트와 함수를 매핑하여 저장합니다.

요청이 들어와 WSGI가 어플리케이션을 호출할 때 주입하는 environ에 URL과 쿼리스트링 정보가 포함되어 있습니다. 이를 통해 적절한 라우팅을 할 수 있습니다. 실제 웹 프레임워크의 URL 라우팅은 Path Parameter등을 고려해야합니다.

 

Get Response

이제 URL 라우팅으로 매칭된 함수를 실행하고 결과 값을 반환하는 로직을 추가하면 됩니다.

class FakeFlask:
    ...
    def __iter__(self) -> Iterable[bytes]:
        endpoint, query = self.__parse_url()
        controller = self.__get_controller(endpoint=endpoint)
        status_code, response_headers, response = self.__get_response(controller)
        self.start_response(status_code, response_headers)
        yield response.encode("utf-8")
        
    def __get_response(self, controller: Callable, *args) -> Tuple[str, List[Tuple[str, str]], str]:
        if not controller: 
            return "404 NOT FOUND", [("Content-Type", "text/plain")], "NOT FOUND"
        try:
            response_body = controller(*args)
        except:
            return "500 INTERNAL ERROR", [("Content-Type", "text/plain")], "INTERNAL ERROR"
        return "200 OK", [("Content-Type", "text/plain")], response_body

요청 URL이 존재하지 않으면 404 NOT FOUND를 반환하고 그게 아니라면 200 OK와 결과값을 반환합니다.

 

Application

이제 어플리케이션을 작성할 수 있습니다.

# sample.py
from app import FakeFlask

app = FakeFlask()

@app.route("/")
def home():
    return "hello world"

@app.route("/ping")
def ping():
    return "pong"

따로 파일을 만들어서 FakeFlask를 import하여 사용했습니다.

uwsgi --http :9000 --wsgi sample:app

curl localhost:9000/
>>> hello world%

기대대로 값이 반환됩니다. 

---

전체 코드는 https://github.com/ooknimm/fake-flask에서 확인할 수 있습니다.