IFastAPI¶
[ ]:
import os, asyncio, uvicorn, fastapi, pydantic, IPython, ujson, tornado.ioloop
Hubba Hubba¶
[ ]:
port = 8000
app_args = {}
uvi_args = dict(port=port)
IN_HUB = "JUPYTERHUB_SERVICE_PREFIX" in os.environ
prefix = "/"
if IN_HUB:
prefix = f"""{os.environ["JUPYTERHUB_SERVICE_PREFIX"]}proxy/{port}"""
app_args.update(openapi_prefix=prefix)
uvi_args.update(root_path=prefix)
Make an App¶
[ ]:
app = fastapi.FastAPI(title="IFastAPI", **app_args)
Make a route¶
[ ]:
@app.get("/")
def read_root():
return {"Hello": "World"}
Serve in-loop with uvicorn
¶
[ ]:
config = uvicorn.Config(app, **uvi_args)
server = uvicorn.Server(config=config)
The actual serving will live in an asyncio.Task
[ ]:
if __name__ == "__main__":
task = asyncio.ensure_future(server.serve())
Inspecting the route¶
[ ]:
def show_route(route, height="400px"):
url = f"{prefix}{route}" if IN_HUB else f"http://localhost:{port}{route}"
display(IPython.display.Markdown(f"[`{url}`]({url})"))
display(IPython.display.IFrame(url, width="100%", height=height))
show_route("/", height="40px")
Live: Modelling the domain¶
[ ]:
class Item(pydantic.BaseModel):
"""
a lovely model
"""
name: str
price: float
is_offer: bool = None
IPython.display.JSON(ujson.loads(Item.schema_json()))
[ ]:
DB = {}
Live: using the domain¶
[ ]:
@app.get("/items/{item_id}", response_model=Item)
def read_item(item_id: int, q: str = None) -> Item:
return DB[item_id]
@app.put("/items/{item_id}")
def create_item(item_id: int, item: Item):
DB[item_id] = item
return {"item_name": item.name, "item_id": item_id}
# openapi gets cached, force reloading it
app.openapi_schema = None
Showing Docs¶
[ ]:
show_route("/docs")
Showing (More) Docs¶
[ ]:
show_route("/redoc")
All done¶
[ ]:
if __name__ == "__main__":
tornado.ioloop.IOLoop.current().add_callback(server.shutdown)