Skip to content

Instantly share code, notes, and snippets.

@kemingy
Created May 2, 2020 09:57
Show Gist options
  • Save kemingy/48f869724f3c7ecba2c4ed132e640488 to your computer and use it in GitHub Desktop.
Save kemingy/48f869724f3c7ecba2c4ed132e640488 to your computer and use it in GitHub Desktop.
test spectree for falcon msgpack handlers
import msgpack
import httpx
if __name__ == "__main__":
url = 'http://localhost:8000/'
with httpx.Client() as client:
client.headers.update({'Content-Type': 'application/msgpack'})
packer = msgpack.Packer(
autoreset=True,
use_bin_type=True,
)
resp = client.post(url, data=packer.pack({'text': 'hello', 'limit': 5}))
print(msgpack.unpackb(resp.content))
from spectree import SpecTree, Response
from pydantic import BaseModel
import falcon
from falcon import media
from wsgiref import simple_server
# setup msgpack handler in falcon
handlers = media.Handlers({
'application/msgpack': media.MessagePackHandler(),
})
app = falcon.API(media_type='application/msgpack')
app.req_options.media_handlers = handlers
app.resp_options.media_handlers = handlers
api = SpecTree('falcon')
class Req(BaseModel):
text: str
limit: int
class Resp(BaseModel):
prob: float
class Demo:
@api.validate(json=Req, resp=Response(HTTP_200=Resp))
def on_post(self, req, resp):
data = req.context.json
print(f'{data.text} - {data.limit}')
resp.media = {'prob': len(data.text) / data.limit}
api.register(app)
app.add_route('/', Demo())
httpd = simple_server.make_server('0.0.0.0', 8000, app)
httpd.serve_forever()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment