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

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

https://ooknimm.tistory.com/2 글에서 이어집니다.

 

thread local

Django와 달리 Flask는 현재 요청 정보가 담겨있는 request 객체를 함수의 인수로 전달하지 않고 import 하여 사용합니다.

from flask import Flask, request

app = Flask(__name__)

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

import만 하면 어느 모듈에서든 접근이 가능합니다. 요청이 발생하면 Flask 내부에서 request 객체를 만들고 전역 변수에 할당하기 때문입니다. 이때 중요한 것은 Flask는 하나의 Python 프로세스에서 하나 이상의 애플리케이션을 가질 수 있습니다. 즉, 여러 스레드로 애플리케이션을 실행할 수 있습니다. 그래서 request 같은 전역 변수는 스레드 간 공유되지 않도록 분리되어야 합니다. 스레드마다 독립적인 변수를 갖는 저장소를 thread local이라고 합니다. Flask는 thread local로 스레드별 context를 분리하였고, current_app, g, request 등의 변수를 thread local object라고 합니다.

 

Contextvar

스레드뿐 아니라 asycnio의 태스크별로 독립적인 local 상태를 관리, 저장, 엑세스하기 위해 제공되는 Python 내장 기능입니다.

set, get, reset 메소드를 사용하여 변수를 저장하고 액세스 할 수 있습니다.

from contextvar import Contextvar

context = Contextvar("my_context_name")
token = context.set("foo")
bar = context.get()
context.reset(token)

print(bar)
# >>> foo

flask는 매 요청마다 contextvar에 Application context와 Request context를 set 하고 요청 종료 시 reset 합니다.

이제 context.py를 만들어 이를 흉내 내어 보겠습니다.

# context.py

# 앱 컨텍스트
_cv_app: ContextVar[AppContext] = ContextVar("flask.app_ctx")

# 요청 컨텍스트
_cv_request: ContextVar[RequestContext] = ContextVar("flask.request_ctx")

 

ApplicationContext

# context.py

# 앱 컨텍스트
_cv_app: ContextVar[AppContext] = ContextVar("flask.app_ctx")

class AppContext:
    def __init__(self, app: "FakeFlask") -> None:
        self.app = app
        self.g = AppCtxGlobals()
    def push(self) -> None:
        self.token = _cv_app.set(self)
        print(self.token)
    
    def pop(self) -> None:
        print("pop app ctx")
        _cv_app.reset(self.token)

AppContext라는 클래스입니다. push 메소드로 _cv_app contextvar에 set하고 pop 메소드로 reset 합니다. 외부에서 push와 pop 메소드를 호출하게 될 텐데 바로 FakeFlask에서 요청이 발생하면 push를 하고 요청이 종료되면 pop을 수행하게 될 것입니다. self.app과 self.g는 thread local object로 제공될 것입니다. 

g

Flask는 사용자 정의 전역 속성을 사용을 위해 g 객체를 제공합니다. g 객체는 AppContext 생성 시 만들어지며 어떠한 데이터든 속성으로 관리할 수 있습니다.

# context.py

class AppCtxGlobals: 
    def __getattr__(self, name: str) -> Any:
        try:
            self.__dict__[name]
        except KeyError:
            print(f"KeyError {name}")
    def __setattr__(self, name: str, value: Any):
        self.__dict__[name] = value

    def get(self, name: str, default: Any) -> Any:
        return self.__dict__.get(name, default)
    
    def pop(self, name: str, default = None) -> Any:
        if not default:
            return self.__dict__.pop(name)
        else:
            return self.__dict__(name, default)

Proxy 패턴을 사용해서 어떠한 속성이든 다이내믹하게 관리할 수 있습니다. 사실 __getattr__, __setattr__을 사용하지 않더라도 속성 동적 할당이 가능합니다만 예외 처리, 기능 확장을 생각한다면 해당 메소드를 구현하는 것이 좋습니다.

 

RequestContext

# context.py

# 요청 컨텍스트
_cv_request: ContextVar[RequestContext] = ContextVar("flask.request_ctx")

class RequestContext:
    def __init__(self, app: "FakeFlask", environ: Dict[str, Any], path: str, query: Dict[str, str]) -> None:
        self.app = app
        self.request = Request(environ=environ, path=path, query=query)

    def push(self) -> None:
        self.token = _cv_request.set(self)
        print(self.token)
    def pop(self) -> None:
        print("pop req ctx")
        _cv_request.reset(self.token)

요청에 대한 정보가 담긴 RequestContext 클래스입니다. ApplicationContext와 마찬가지로 push, pop 메소드로 Contextvar에 접근합니다. self.request는 thread local object로 제공될 것입니다. 

Request

class Request:
    def __init__(self, environ: Dict[str, Any], path: str, query: Dict[str, str]) -> None:
        self.method = environ.get("REQUEST_METHOD", "GET")
        self.path = path
        self.query = query
        self.stream = cast(IO[bytes], environ["wsgi.input"])

    def get_json(self) -> Any:
        print(f"request data stream: {self.stream}")
        return json.loads(self.stream.read())

request의 method, path, body 데이터를 제공합니다. wsgi는 request 소켓 객체를 environ["wsgi.input"]으로 할당하여 전달하기 때문에 이를 사용해 request body를 읽을 수 있습니다.

 

proxy

이제 ApplicationContext, RequestContext의 속성인 self.app(current_app), self.g(g), self.request(request)를 thread local object로 제공하기 위해서 proxy 패턴과 디스크립터를 사용합니다.

디스크립터는 클래스 속성 훅입니다. 클래스의 속성에 접근할 때 이를 가로챕니다.

# context.py

class ProxyLookup:
    def __init__(self, f: Callable) -> None:
        if f:
            def bind_f(_, obj):
                return partial(f, obj)
        else:
            bind_f = None
        self.bind_f: Optional[Callable] = bind_f
    def __set_name__(self, _, name: str) -> None:
        self.name = name
    def __get__(self, instance: "LocalProxy", _):
        obj = instance.get_current_obj()
        if self.bind_f:
            return self.bind_f(instance, obj)
        return getattr(obj, self.name)


class LocalProxy:
    def __init__(self, ctx: ContextVar, name: str) -> None:
        def get_current_obj() -> Any:
            get_name = attrgetter(name)
            obj = ctx.get()
            return get_name(obj)
        object.__setattr__(self, "get_current_obj", get_current_obj)

    __doc__ = ProxyLookup(__doc__)
    __repr__ = ProxyLookup(repr)
    __str__ = ProxyLookup(str)
    __getattr__ = ProxyLookup(getattr)
    __setattr__ = ProxyLookup(setattr)
    __delattr__ = ProxyLookup(delattr)
    __dir__ = ProxyLookup(dir)
    
request: Request = LocalProxy(_cv_request, "request")
g: AppCtxGlobals = LocalProxy(_cv_app, "g")
current_app = LocalProxy(_cv_app, "app")

ProxyLookup가 디스크립터이고 LocalProxy가 proxy 패턴을 구현한 클래스입니다. LocalProxy 생성자에 context와 context에 저장되어 있는 thread local object의 이름을 할당하여 생성하면 해당 변수에 접근할 때 context내의 지정된 속성에 접근하게 됩니다.

LocalProxy의 몇 가지 매직 메소드에 디스크립터를 할당하여 외부에서는 LocalProxy의 존재를 모르도록 하였습니다. 특히 __setattr__,  __getattr__,  __delattr__를 후킹하여 thread local object의 속성에 접근할 수 있습니다.

ProxyLookup은 LocalProxy의 get_current_obj 메소드를 사용하여 thread local object를 접근하는 것이 중요합니다.

 

App 적용

이제 context를 app에 적용해 봅니다.

1. request가 발생하면 AppContext, RequestContext를 생성, push 하고 request가 종료될 때 reset 합니다.

2. 외부에서 context 모듈을 직접 호출하지 않도록 app에 import 합니다.

# app.py 

from context import RequestContext, AppContext, request as request, g as g, current_app as current_app

class FakeFlask:
        .
        .
        .
    def __create_context(self, endpoint: str, query: Dict[str, str]) -> Tuple[AppContext, RequestContext]:
        app_ctx = AppContext(self)
        app_ctx.push()
        req_ctx = RequestContext(self, self.environ, endpoint, query)
        req_ctx.push()
        return app_ctx, req_ctx

    def __pop_context(self, app_ctx: AppContext, req_ctx: RequestContext) -> None:
        app_ctx.pop()
        req_ctx.pop()
    
    def __iter__(self) -> Iterable[bytes]:
        endpoint, query = self.__parse_url()
        app_ctx, req_ctx = self.__create_context(endpoint, query)
        controller = self.__get_controller(endpoint=endpoint)
        status_code, response_headers, response = self.__get_response(controller)
        self.__pop_context(app_ctx, req_ctx)
        self.start_response(status_code, response_headers)
        yield response.encode("utf-8")

 

테스트

sample app을 만들어서 g, request, current_app을 테스트해봅니다. 

# sample.py

app = FakeFlask()

@app.route("/")
def home():
    g.foo = "bar"
    home_service()
    return "hello world"

# sample_service.py

from app import g, request, current_app
import time

def home_service():
    print(f"request: {request}")
    print(f"g: {g}")
    print(f"current_app: {current_app}")
    print(request.path)
    print(g.foo)
    time.sleep(3)

thread local object 정보를 출력하고 3초간 슬립 하는 api 함수입니다. 거의 동시에 요청하여 스레드마다 thread local object가 다른지 테스트할 것입니다.

 

thread local임을 테스트하기 위해 uwsgi를 실행할 때 threads 옵션을 2 이상으로 설정해야 합니다.

uwsgi --http :9000 --wsgi sample.sample:app --threads 2

 

이제 터미널을 두 개 띄어 curl localhost:9000을 입력하면 uwsgi에서 다음과 같이 출력됩니다.

request: <context.Request object at 0x10cf23ad0>
g: <context.AppCtxGlobals object at 0x10cf23950>
current_app: <app.FakeFlask object at 0x10ce43210>
/
bar
request: <context.Request object at 0x10cf38310>
g: <context.AppCtxGlobals object at 0x10cf381d0>
current_app: <app.FakeFlask object at 0x10ce43210>
/
bar

request, g, current_app을 보면 요청마다 다른 객체임을 확인할 수 있습니다.

 

---

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