DRAFT: Use FastAPI to put asyncio shims on top of OL API calls

[cclauss]

Related to #5590
Related to #5910
Related to #5914

DRAFT: Do not merge.

NOTE: FastAPI/main.py.disabled --> FastAPI/main.py after FastAPI is added to our requirements.

Use FastAPI to put a Python asyncio shim on top of calls to https://openlibrary.org.

Why?

• Start experimenting with asyncio for our API responses.
• Auto-creates OpenAPI (Swagger) docs including "Try it out" mode.
• Auto-validates both inputs and outputs against json schemas with Pydantic.

mkdir fastapi_hack
cd fastapi_hack
python3.9 -m venv .venv
source .venv/bin/activate
pip install --upgrade pip
pip install "fastapi[all]"

Put the contents of main.py.disabled into main.py

uvicorn main:app --reload

This starts an asyncio-based webserver that will auto-reload as changes are saved to main.py.

Try:

• http://127.0.0.1:8000/books/OL7353617M
• Other URLs in the comments in main.py
• http://127.0.0.1:8000/docs for Swagger docs with Try it out mode.
• http://127.0.0.1:8000/redoc for auto-generated API docs

Technical

Testing

See comments at the top of FastAPI/test_fastapi_hack.py.disabled

Also, when the uvicorn server is running go to http://127.0.0.1:8000/docs and do Try it out under various endpoints.

Screenshot

Autogenerated content:

http://127.0.0.1:8000/docs

---

---

http://127.0.0.1:8000/redoc

Stakeholders

@bennypowers

[RayBB]

Won't have time to check it out this week but I've been thinking about making OpenAPI docs forever and this would be fantastic! May also expose some inconsistencies in our APIs (like when we return 400 txt responses instead of json)

[bennypowers]

Yeah this is 🔥

We can use this to generate a graphQL api using graphql mesh

From there I want to both vend ready-made GraphQL-querying custom elements and create a PWA for library builders using Apollo Elements

[cclauss]

• FastAPI
  • Demo is awesome
  • Maybe a first value-yielding step might be to try to use FastAPI+friends to generate a static html/js site for the docs, which we then server from Open Library? We would the pydantic error messages though.
  • Other thoughts/concerns:
    • fastapi functions are kind of duplicates of the webpy ones ; can we codegen or bring them closer? Web.py --> FastAPI codegen for 87 functions from 22 files #5910
    • How to add more docs/helpful context to parameters, functions, etc ;
    • Can we use pydantic in some way to improve the API experience, without getting all of fastapi? Similar to how we did for imports?

[cclauss]

Put this code at the bottom of openlibrary/plugins/openlibrary/api.py and then make minor modifications and add docstrings. Next step: Add URL params and query strings. Fix for mutable default arguments (lists).

import inspect
import sys
from typing import Iterator

def enhance_query_arg(arg: str, default_type: str = "Optional[str]") -> list[str]:
    """
    >>> [enhance_query_arg(arg) for arg in (" x ", " x = -1234 ", "x=-5", "x=-67.89")]
    ['x', 'x: int = -1234', 'x: int = -5', 'x: float = -67.89']
    >>> [enhance_query_arg(arg) for arg in ("x=[ ]", "x=None", 'x=""','x="text"')]
    ['x: list = [ ]', 'x: Optional[str] = None', 'x: str = ""', 'x: str = "text"']
    >>> [enhance_query_arg(arg) for arg in ("x=true", "x=FALSE", '  x  =  tRUE  ')]
    ['x: bool = True', 'x: bool = False', 'x: bool = True']
    """
    key, _, value = (item.strip() for item in arg.partition("="))
    if not value:
        return key
    type_hint = {
        "[]": "list",
        "''": "str",
        '""': "str",
        "None": default_type,
        "True": "bool",
        "False": "bool",
    }.get(value.title(), "")
    if type_hint:
        value = value.title()
    elif value.lstrip("-").isdigit():
        type_hint = "int"
    elif value.lstrip("-").replace(".", "", 1).isdigit():
        type_hint = "float"
    elif any(value.startswith(q) and value.endswith(q) for q in ("'", '"')):
        type_hint = "str"
    elif value.startswith("[") and value.endswith("]"):
        type_hint = "list"
    return f"{key}: {type_hint or default_type} = {value}"


def get_query_args(code: CodeType, method_call: str = "web.input(") -> list[str]:
    """
    Look thru the lines of code looking for `method_call` and if found, extract
    the parameters and enhance them with Python type hints and default values.
    NOTE: Failure case if a method parameter is a tuple!
    """
    code = "".join(line.strip() for line in inspect.getsourcelines(code)[0])
    _, _, code = code.partition(method_call)  # Only the text after method_call
    if not code:
        return []
    return [enhance_query_arg(arg) for arg in code.split(")")[0].split(",")]


def get_url_path(path: str, path_args: list[str]):
    if not path_args:
        return path
    parts = [part for part in path.split("/") if part]
    args = path_args[:]  # Make our own copy so we can pop from it
    for i, part in enumerate(parts):
        if "(" in part:
            assert ")" in part, "Opening and closing parens must in the same part!"
            parts[i] = "{%s}" % args.pop(0)
    return "/" + "/".join(parts)


get_fmt = '''@app.get("{path}")
async def {classname}_get({method_params}) -> str:
    """
    {docstring}
    """
    {body_code}
'''


def webpy_to_fastapi() -> Iterator[str]:
    """
    Codegen FastAPI async functions from the web.py GET functions in this file.

    >>> list(webpy_to_fastapi()) != []  # Ensure that paths and docstrings are correct.
    True
    """
    for classname, cls in inspect.getmembers(sys.modules[__name__], inspect.isclass):
        if not getattr(cls, "path", None):
            continue
        get = getattr(cls, "GET", getattr(cls, "get", None))
        if not get:
            continue

        code = get.__code__  # @decorators may block access to __code__ and __doc__
        assert code.co_varnames[0] == "self"
        path_args = list(code.co_varnames[1: code.co_argcount])
        query_args = get_query_args(code)

        url_base = 'return requests.get(f"https://openlibrary.org'
        url_path = get_url_path(cls.path, path_args) if path_args else cls.path
        url_params = ", params=locals()" if query_args else ""
        body_code = f'{url_base}{url_path}"{url_params}).json()'

        d = {
            "path": cls.path,
            "classname": cls.__name__,
            "method_params": ", ".join(path_args + query_args),
            "docstring": inspect.cleandoc(get.__doc__),
            "body_code": body_code,
        }
        yield get_fmt.format(**d)


if __name__ == "__main__":
    from doctest import testmod

    testmod()  # Run our doctests before running codegen.
    print("Start")
    functions = sorted(webpy_to_fastapi())
    print("\n\n".join(functions))
    print(f"Finished generating {len(functions)} functions.")

Generated code:

Start
@app.get("/_tools/amazon_search")
async def amazon_search_api_get(title: str = "", author: str = "") -> str:
    """
    This function is amazon_search_api.get()
    """
    return requests.get(f"https://openlibrary.org/_tools/amazon_search", params=locals()).json()


@app.get("/authors/OL(\d+)A)/works")
async def author_works_get(key, limit: int = 50, offset: int = 0) -> str:
    """
    This function is author_works.get()
    """
    return requests.get(f"https://openlibrary.org/authors/{key}/works", params=locals()).json()


@app.get("/availability/v2")
async def book_availability_get(type: str = "", ids: str = "") -> str:
    """
    This function is book_availability.get()
    """
    return requests.get(f"https://openlibrary.org/availability/v2", params=locals()).json()


@app.get("/browse")
async def browse_get(q: str = "", page: int = 1, limit: int = 100, subject: str = "", work_id: str = "", _type: str = "", sorts: str = "") -> str:
    """
    This function is browse.get()
    """
    return requests.get(f"https://openlibrary.org/browse", params=locals()).json()


@app.get("/observations")
async def public_observations_get(olid: list = []) -> str:
    """
    This function is class public_observations.get()
    """
    return requests.get(f"https://openlibrary.org/observations", params=locals()).json()


@app.get("/prices")
async def price_api_get(isbn: str = "", asin: str = "") -> str:
    """
    This function is price_api.get()
    """
    return requests.get(f"https://openlibrary.org/prices", params=locals()).json()


@app.get("/sponsorship/eligibility/(.*)")
async def sponsorship_eligibility_check_get(_id, patron: Optional[str] = None, scan_only: bool = False) -> str:
    """
    This function is sponsorship_eligibility_check.get()
    """
    return requests.get(f"https://openlibrary.org/sponsorship/eligibility/{_id}", params=locals()).json()


@app.get("/works/OL(\d+)W/bookshelves")
async def work_bookshelves_get(work_id) -> str:
    """
    This function is work_bookshelves.get()
    """
    return requests.get(f"https://openlibrary.org/works/{work_id}/bookshelves").json()


@app.get("/works/OL(\d+)W/editions")
async def work_editions_get(key, limit: int = 50, offset: int = 0) -> str:
    """
    This function is work_editions.get()
    """
    return requests.get(f"https://openlibrary.org/works/{key}/editions", params=locals()).json()


@app.get("/works/OL(\d+)W/observations")
async def patrons_observations_get(work_id) -> str:
    """
    This function is class patrons_observations.get()
    """
    return requests.get(f"https://openlibrary.org/works/{work_id}/observations").json()

Finished generating 10 functions.

[cclauss]

Update: This was done in #5910...

This is nasty regex clutter... @app.get("(/authors/OL\d+A)/works")

Would it break things if we manually transform it to /authors/OL(\d+)A/works?

[cclauss]

I updated the code above to add type hints, doctests, and only passing of locals() when there are query parameters.

[cclauss]

• Web.py --> FastAPI codegen for 87 functions from 22 files #5910 Codegens the shims in one of our API files. The idea would be that each API file could codegen its own functions. A central process could request that the codegen to done and consolidate the results from many files.

[cclauss]

Closing in favor of #5910 which automates the code generation.