feat!: reimplement v1 API with FastAPI

g2p/api.py

@@ -1,12 +1,9 @@
-""" Very Basic API
-"""
+"""REST API for G2P index-preserving grapheme-to-phoneme conversion using FastAPI."""
 
-import json
-import os
+from enum import Enum
+from typing import List
 
-from flask import Blueprint, abort
-from flask_cors import CORS
-from flask_restful import Api, Resource, inputs, reqparse  # type: ignore
+from fastapi import FastAPI, HTTPException, Query
 from networkx.algorithms.dag import ancestors, descendants  # type: ignore
 from networkx.exception import NetworkXError  # type: ignore
 
@@ -17,153 +14,104 @@
 from g2p.mappings.langs import LANGS_NETWORK
 from g2p.static import __file__ as static_file
 
-
-class Ancestors(Resource):
-    def get(self, node):
-        try:
-            return sorted(ancestors(LANGS_NETWORK, node))
-        except NetworkXError:
-            abort(404)
-
-
-class Descendants(Resource):
-    def get(self, node):
-        try:
-            return sorted(descendants(LANGS_NETWORK, node))
-        except NetworkXError:
-            abort(404)
-
-
-class Langs(Resource):
-    def __init__(self):
-        # TODO: exclude parent dir and maybe null values too
-        self.AVAILABLE_MAPPINGS = [
-            json.loads(mapping.model_dump_json())
-            for mapping in sorted(MAPPINGS_AVAILABLE, key=lambda x: x.in_lang)
-        ]
-        self.parser = reqparse.RequestParser()
-        self.parser.add_argument(
-            "verbose",
-            dest="verbose",
-            type=bool,
-            location="args",
-            default=False,
-            required=False,
-            help="Return verbose mappings information",
-        )
-
-    def get(self):
-        args = self.parser.parse_args()
-        verbose = args["verbose"]
-        if verbose:
-            return self.AVAILABLE_MAPPINGS
-        else:
-            return sorted(LANGS_NETWORK.nodes)
-
-
-class Text(Resource):
-    def __init__(self):
-        self.parser = reqparse.RequestParser()
-        self.parser.add_argument(
-            "in-lang",
-            dest="in-lang",
-            type=str,
-            location="args",
-            required=True,
-            help="The input language",
-        )
-        self.parser.add_argument(
-            "out-lang",
-            dest="out-lang",
-            type=str,
-            location="args",
-            required=True,
-            help="The output language",
-        )
-        self.parser.add_argument(
-            "text",
-            dest="text",
-            type=str,
-            location="args",
-            required=True,
-            help="The text in the input language",
+# Create the v1 version of the API
+api = FastAPI(
+    title="Simple G2P API",
+    description="A simple API for the G2P module",
+    version="1.1.0",
+    contact={"email": "hello@aidanpine.ca"},
+    license_info={
+        "name": "MIT",
+        "url": "https://github.com/roedoejet/g2p/blob/main/LICENSE",
+    },
+    openapi_tags=[
+        {
+            "name": "ancestors",
+            "description": "Find which mappings can convert to a given node",
+        },
+        {
+            "name": "descendants",
+            "description": "Find which mappings can be converted to from a given node",
+        },
+        {"name": "g2p", "description": "Transduced, g2p'ed forms"},
+        {"name": "langs", "description": "Languages/mappings available for G2P"},
+    ],
+)
+
+# Get the langs
+LANGS = sorted(LANGS_NETWORK.nodes)
+Lang = Enum("Lang", [(name, name) for name in LANGS])  # type: ignore
+
+
+@api.get(
+    "/ancestors/{node}",
+    summary="get all ancestors of node",
+    tags=["ancestors"],
+    operation_id="getAncestors",
+    response_description="The valid ancestors of a node",
+)
+def get_all_ancestors_of_node(node: Lang = Query(description="language node name")) -> List[str]:
+    """Get the valid ancestors in the network's path to a given node. These
+    are all the mappings that you can convert from in order to get the
+    given node."""
+    return sorted(ancestors(LANGS_NETWORK, node.name))
+
+
+@api.get(
+    "/descendants/{node}",
+    summary="get all descendants of node",
+    tags=["descendants"],
+    operation_id="getDescendants",
+    response_description="The valid descendants of a node",
+)
+def get_all_descendants_of_node(
+    node: Lang = Query(description="language node name"),
+) -> List[str]:
+    return sorted(descendants(LANGS_NETWORK, node.name))
+
+
+@api.get(
+    "/g2p",
+    summary="get g2p'ed form",
+    tags=["g2p"],
+    operation_id="convertString",
+    response_description="The converted text",
+)
+def g2p(
+    in_lang: Lang = Query(alias="in-lang", description="input lang of string"),
+    out_lang: Lang = Query(alias="out-lang", description="output lang of string"),
+    text: str = Query(description="string to convert"),
+    index: bool = Query(False, description="return indices"),
+    debug: bool = Query(False, description="return debugging information"),
+) -> dict:
+    """Get the converted version of a string, given an input and output lang"""
+    try:
+        transducer = make_g2p(in_lang.name, out_lang.name)
+        tg = transducer(text)
+        return {
+            "input-text": tg.input_string,
+            "output-text": tg.output_string,
+            "debugger": tg.debugger if debug else debug,
+            "index": tg.edges if index else index,
+        }
+    except NoPath:
+        raise HTTPException(
+            status_code=400, detail=f"No path from {in_lang} to {out_lang}"
         )
-        self.parser.add_argument(
-            "index",
-            dest="index",
-            type=inputs.boolean,
-            location="args",
-            default=False,
-            required=False,
-            help="Return indices",
+    except InvalidLanguageCode:
+        # Actually should never happen!
+        raise HTTPException(
+            status_code=404, detail="Unknown input or output language code"
         )
-        self.parser.add_argument(
-            "debugger",
-            dest="debugger",
-            type=inputs.boolean,
-            location="args",
-            default=False,
-            required=False,
-            help="Debugging information about the transduction process",
-        )
-        self.parser.add_argument(
-            "tokenize",
-            dest="tokenize",
-            type=inputs.boolean,
-            location="args",
-            default=False,
-            required=False,
-            help="Tokenize before transducing",
-        )
-
-    def get(self):
-        args = self.parser.parse_args()
-        in_lang = args["in-lang"]
-        out_lang = args["out-lang"]
-        text = args["text"]
-        index = args["index"]
-        debugger = args["debugger"]
-        tokenize = args["tokenize"]
-        try:
-            transducer = make_g2p(in_lang, out_lang, tokenize=tokenize)
-            tg = transducer(text)
-            text = tg.output_string
-            input_text = tg.input_string
-            debugger = tg.debugger if debugger else debugger
-            index = tg.edges if index else index
-            return {
-                "input-text": input_text,
-                "output-text": text,
-                "index": index,
-                "debugger": debugger,
-            }
-        except NoPath:
-            abort(400)
-        except InvalidLanguageCode:
-            abort(404)
-
-
-def update_docs():
-    """Update the swagger documentation with all nodes from the network"""
-    swagger_path = os.path.join(os.path.dirname(static_file), "swagger.json")
-    with open(swagger_path, encoding="utf-8-sig") as f:
-        data = json.load(f)
-    data["components"]["schemas"]["Langs"]["enum"] = sorted(LANGS_NETWORK.nodes)
-    with open(swagger_path, "w", encoding="utf-8", newline="\n") as f:
-        f.write(json.dumps(data) + "\n")
-    LOGGER.info("Updated API documentation")
-
-
-g2p_api = Blueprint("resources-g2p", __name__)
-
-CORS(g2p_api)
-
-api = Api(g2p_api)
-
-api.add_resource(Ancestors, "/ancestors/<string:node>", endpoint="ancestors")
-
-api.add_resource(Descendants, "/descendants/<string:node>", endpoint="descendants")
 
-api.add_resource(Langs, "/langs", endpoint="langs")
 
-api.add_resource(Text, "/g2p", endpoint="g2p")
+@api.get(
+    "/langs",
+    summary="find all possible languages in g2p",
+    tags=["langs"],
+    operation_id="searchTable",
+    response_description="search results matching criteria",
+)
+def langs() -> List[str]:
+    """By passing in the appropriate options, you can find available mappings"""
+    return LANGS