Proof-of-concept to retrieve (GET) and store (POST) ratings from documentation URLs similar to doc.suse.com

docrate.py

#!/usr/bin/env python3
"""
Proof-of-concept to retrieve (GET) and store (POST) ratings from
documentation URLs similar to doc.suse.com.

Requirements
------------

* aiohttp
* Python >=3.6, preferably a more recent version


Retrieve information from database
----------------------------------

    * Return JSON object:
      $ curl http://localhost:8080/sle-ha/15-GA/single-html/SLE-HA-pmremote-quick/
      {"rate": 5}
    

Store information into database
-------------------------------

    $ curl -X POST -H "Content-Type: application/json" \
        -d '{"rate": 5}'  \
        http://localhost:8080/sle-ha/15-SP1/html/SLE-HA-all/foo.html
    $ curl http://localhost:8080/sle-ha/15-SP1/html/SLE-HA-all/foo.html
    {"rate": 5}

TODOs
-----

* Connect to a real database
* Improve logging
* Correct error codes when something goes wrong
* Allow application/x-www-form-urlencoded" as content type?
* Be relaxed when using double slashes (http://localhost:8080//... vs. http://localhost:8080/...)
* Detect SUMA URLs


See also
--------

 * aiohttp documentation:  https://docs.aiohttp.org
 * curl POST examples: https://gist.github.com/subfuzion/08c5d85437d5d4f00e58
"""

from aiohttp import web
import json
import logging
from typing import Dict, Union, Optional


logging.basicConfig(level=logging.DEBUG)
routes = web.RouteTableDef()

AVAILABLE_PRODUCTS = ("sle-ha", "sle-hpc", "sles", "suma", "slesforsap", ...)
DATABASE = [
    # The keys are a moving target and to be defined.
    # Theoretically, we don't need product, release,
    # and sp. It was just from a former idea so I left it to have some "flesh".
    #
    # We save only the part of the part of the URL after the host name to
    # make comparison a bit easier and avoid host name variations
    dict(
        product="sle-hpc",
        release=15,
        sp=3,
        rate=10,
        url="/sle-hpc/15-SP3/html/hpc-guide/cha-slurm.html",
    ),
    dict(
        product="sle-ha",
        release=15,
        sp=0,
        rate=5,
        url="/sle-ha/15-GA/single-html/SLE-HA-pmremote-quick/",
    ),
    dict(
        product="sle-ha",
        release=15,
        sp=0,
        rate=2,
        url="/sle-ha/15-GA/html/SLE-HA-all/art-sleha-pmremote-quick.html",
    ),
]


def search_in_database(url: Union[str, yarl.URL]) -> Dict[str, Optional[int]]:
    """
    Search in the database for given URL

    :param url: the URL
    """
    for item in DATABASE:
        u = item["url"]
        if u == str(url):
            return item
    return dict(rate=None)


@routes.get(r"/{product}/{release}/html/{guide}/{topic}")
@routes.get(r"/{product}/{release}/single-html/{guide}/")
# /sle-ha/15-GA/single-html/SLE-HA-pmremote-quick/
# /external-tree/en-us/suma/4.1/suse-manager/installation/install-intro.html
# /external-tree/en-us/suma/4.0/suse-manager/retail/retail-introduction.html
# /sle-pos/11-SP3/html/SLEPOS-imgsrv12/index.html
async def get_rating(request: web.Request) -> web.StreamResponse:
    """
    Async function when a GET event happen
    """
    product = request.match_info.get("product")
    release = request.match_info.get("release")
    guide = request.match_info.get("guide")
    topic = request.match_info.get("topic")

    # see https://docs.aiohttp.org/en/stable/web_reference.html#aiohttp.web.Request
    url = request.rel_url

    found = search_in_database(str(url))
    data = {"rate": found.get("rate")}

    return web.json_response(data)
    # another alternative would be simple text:
    # return web.Response(text=f"Found rating {found['rate']}", content_type="text/plain")


@routes.post(r"/{product}/{release}/html/{guide}/{topic}")
@routes.post(r"/{product}/{release}/single-html/{guide}/")
async def post_rateing(request: web.Request) -> web.StreamResponse:
    """
    Async function called when a POST event happen.
    
    If every parameter was okay, rating is included into the database.
    """
    # The product, release guide, and topic are not needed, but it is
    # "required" to form the URL.
    # Haven't found an easy way to make a "get everything after /" request
    # You can extract the variable with, for example:
    # product = request.match_info.get("product", None)

    url = str(request.rel_url)

    if request.content_type != "application/json":
        return web.HTTPNotAcceptable(
            text=f"Unsupported content_type {request.content_type}"
        )

    # Check if we got the right product abbreviation:
    if product not in AVAILABLE_PRODUCTS:
        return web.HTTPNotAcceptable(text=f"product {product!r} not known")

    rating = await request.json()
    print(f"Received {rating}")

    if rating.get("rate") is None:
        return web.HTTPNotAcceptable(text=f"Rating not provided")
    try:
        rate = rating.get("rate")
        rate = int(rate)
    except ValueError:
        return web.HTTPNotAcceptable(
            text=f"Invalid rating, expected a number, but got {rate!r}"
        )

    # All is good, so include it into our database:
    data = dict(product=product, release=release, url=url, rate=rate)
    DATABASE.append(data)

    return web.Response(text=f"Added new rating entry with {data}")


async def on_shutdown(app: web.Application) -> None:
    print("\n*** Closing server...")


def main():
    """
    Main entry point
    """
    app = web.Application()
    app.add_routes(routes)
    app.on_shutdown.append(on_shutdown)
    return web.run_app(app)


if __name__ == "__main__":
    print("DATABASE:", DATABASE)
    main()