Skip to content

Instantly share code, notes, and snippets.

@aodag
Created June 16, 2019 16:00
Show Gist options
  • Save aodag/c2a91ea2342bbebdfe0f2e4e72e7e9c0 to your computer and use it in GitHub Desktop.
Save aodag/c2a91ea2342bbebdfe0f2e4e72e7e9c0 to your computer and use it in GitHub Desktop.
request引数のgenericsで指定したTypeからmarshmallowスキーマを生成して設定するwsgi wrapper
import json
import inspect
from datetime import datetime
from dataclasses import dataclass, asdict
from typing import TypeVar, Generic, Optional, Callable, Dict, Any, Type, Iterable, Sequence
from typing_extensions import Protocol
import webob
import webtest
import marshmallow_dataclass
from marshmallow import Schema
T = TypeVar("T")
class Request(Generic[T], webob.Request):
def __init__(self, environ, data: T, schema: Schema) -> None:
super().__init__(environ)
self._data = data
self.schema = schema
@property
def data(self) -> T:
return self._data
TypedApp = Callable[[Request[T]], Dict[str, Any]]
class WSGIWrap(Generic[T]):
def __init__(self, app: TypedApp[T]) -> None:
self._app = app
self._type: Type[T] = inspect.getfullargspec(app).annotations["request"].__args__[0]
self._schema = marshmallow_dataclass.class_schema(self._type)()
def __call__(self, environ, start_response) -> Iterable[bytes]:
j = json.load(environ["wsgi.input"])
data = self._schema.load(j).data
print(data)
req = Request(environ, data, self._schema)
result = self._app(req)
start_response("200 OK", [('Content-type', 'application/json')])
return [json.dumps(result).encode("utf-8")]
@dataclass
class Person:
first_name: str
last_name: str
def app(request: Request[Person]) -> dict:
d = request.data
if d is None:
return dict()
return asdict(d)
wsgi_app = WSGIWrap(app)
test_app = webtest.TestApp(wsgi_app)
res = test_app.post_json("/", params={"first_name": "Atsushi", "last_name": "Odagiri"})
print(res)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment