Last active
July 21, 2023 12:15
-
-
Save tomschr/695e4a04d50ef129285ae70f0c6f628a to your computer and use it in GitHub Desktop.
Revisions
-
tomschr revised this gist
Aug 4, 2021 . 1 changed file with 6 additions and 0 deletions.There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode charactersOriginal file line number Diff line number Diff line change @@ -56,6 +56,12 @@ 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, -
Aug 4, 2021 . 1 changed file with 4 additions and 25 deletions.There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode charactersOriginal file line number Diff line number Diff line change @@ -93,18 +93,6 @@ def search_in_database(url: Union[str, yarl.URL]) -> Dict[str, Optional[int]]: 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/ @@ -122,14 +110,6 @@ async def get_rating(request: web.Request) -> web.StreamResponse: # 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")} @@ -150,11 +130,10 @@ async def post_rateing(request: web.Request) -> web.StreamResponse: # 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( @@ -179,7 +158,7 @@ async def post_rateing(request: web.Request) -> web.StreamResponse: ) # 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}") -
Aug 4, 2021 . 1 changed file with 43 additions and 20 deletions.There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode charactersOriginal file line number Diff line number Diff line change @@ -54,21 +54,36 @@ logging.basicConfig(level=logging.DEBUG) routes = web.RouteTableDef() AVAILABLE_PRODUCTS = ("sle-ha", "sle-hpc", "sles", "suma", "slesforsap", ...) DATABASE = [ 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: @@ -85,7 +100,7 @@ async def handle(request: web.Request) -> web.StreamResponse: """ Example """ name = request.match_info.get("name", "Anonymous") text = f"""Hello, {name}""" return web.Response(text=text) @@ -98,25 +113,27 @@ async def handle(request: web.Request) -> web.StreamResponse: # /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 print(" rel_url=", url, dir(url)) print(" fragment=", url.fragment) print(" raw_parts=", url.raw_parts) print(" ", dir(request)) for key, item in dict( product=product, release=release, guide=guide, topic=topic ).items(): print(f" {key}={item}") 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") @@ -126,13 +143,17 @@ async def get_rating(request: web.Request) -> web.StreamResponse: @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 product = request.match_info.get("product", None) release = request.match_info.get("release", None) sp = request.match_info.get("sp", None) url = request.rel_url if request.content_type != "application/json": @@ -144,16 +165,18 @@ async def post_rateing(request: web.Request) -> web.StreamResponse: 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=str(url), rate=rate) @@ -169,13 +192,13 @@ async def on_shutdown(app: web.Application) -> None: 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() -
Aug 4, 2021 . 1 changed file with 11 additions and 63 deletions.There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode charactersOriginal file line number Diff line number Diff line change @@ -35,6 +35,7 @@ * 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 @@ -64,47 +65,19 @@ ] def search_in_database(url) -> dict: """ 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) # Only for test purposes: @routes.get("/") @routes.get("/{name}") @@ -125,6 +98,7 @@ async def handle(request: web.Request) -> web.StreamResponse: # /sle-pos/11-SP3/html/SLEPOS-imgsrv12/index.html async def get_rating(request: web.Request) -> web.StreamResponse: """ Async function when a GET with given parameters """ product = request.match_info.get("product") release = request.match_info.get("release") @@ -136,8 +110,6 @@ async def get_rating(request: web.Request) -> web.StreamResponse: print(" rel_url=", url, dir(url)) print(" fragment=", url.fragment) print(" raw_parts=", url.raw_parts) print(" ", dir(request)) for key, item in dict(product=product, release=release, guide=guide, topic=topic).items(): print(f" {key}={item}") @@ -150,38 +122,14 @@ async def get_rating(request: web.Request) -> web.StreamResponse: # 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: """ """ # 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 product = request.match_info.get("product", None) release = request.match_info.get("release", None) sp = request.match_info.get("sp", None) -
Aug 4, 2021 .There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode charactersOriginal file line number Diff line number Diff line change @@ -0,0 +1,233 @@ #!/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/...) 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 = [ 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 get_db_entries(product: str, release: int, sp: Optional[int]): """ Get database entry or None, if no entry is available :param product: the abbreviated product string :param release: the release (12, 15, ...) :param sp: the optional SP or None """ sp_or_empty = "" if sp is None else f"SP{sp}" # Filter the database against product and release def takewhile(): yield from (item for item in DATABASE if (product==item["product"]) and \ (release==item["release"]) ) # If sp is None, return a generator of all releases if sp is None: yield from takewhile() else: # If sp is not None, return a specific release sp = int(sp) # print(f"Trying to find /{product}/{release}/{sp}...") for item in takewhile(): if sp==item["sp"]: yield item # If there is no match, an empty generator will be returned def search_in_database(url): """ Search in the database for given URL """ for item in DATABASE: u = item["url"] if u == str(url): return item return dict(rate=None) # Only for test purposes: @routes.get("/") @routes.get("/{name}") async def handle(request: web.Request) -> web.StreamResponse: """ Example """ name = request.match_info.get('name', "Anonymous") text = f"""Hello, {name}""" return web.Response(text=text) @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: """ """ 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 print(" rel_url=", url, dir(url)) print(" fragment=", url.fragment) print(" raw_parts=", url.raw_parts) # print(" values()=", request.values()) # print(" headers=", request.headers) print(" ", dir(request)) for key, item in dict(product=product, release=release, guide=guide, topic=topic).items(): print(f" {key}={item}") 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: """ Returns JSON payload from parsed URL HINTS: * Trailing slash is unimportant * Separator between release and SP can be "-" or "/" """ product = request.match_info.get("product") release = int(request.match_info.get("release", "0").replace("/", "")) sp = request.match_info.get("sp", None) sp = None if sp is None else sp.replace("/", "") url = request.rel_url entries = list(get_db_entries(product, release, sp)) print("entries:", entries) if not entries: return web.HTTPNotFound( text=f"Rating not found for {product}/{release}/SP{sp}", reason=f"No rating found" ) print("request URL:", request.url) return web.json_response(entries) # @routes.post(r"/{product}/{release:\d+}{sep:[/-]}SP{sp:\d/?}") @routes.post(r"/{product}/{release}/html/{guide}/{topic}") @routes.post(r"/{product}/{release}/single-html/{guide}/") async def p_rateing(request: web.Request) -> web.StreamResponse: product = request.match_info.get("product", None) release = request.match_info.get("release", None) sp = request.match_info.get("sp", None) url = 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=str(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()