Skip to content

Instantly share code, notes, and snippets.

@devdave

devdave/book.py

Last active Feb 1, 2020
Embed
What would you like to do?
Example code for having routed classes in Flask. These are different than View Classes or Blueprints as it allows exposing individual methods to the routing system
"""
Example code
============
"""
import typing as T
from flask import request, redirect
if T.TYPE_CHECKING:
from ..lib.cls_endpoint import CLSEndpointFlask
from .. import app
@app.add_class("/book")
class Book(object):
def __init__(self):
"""
This is dangerous and not advised for production because flask is threaded.
I just did this as an example BUT it could be used for cases where you have
stateless variables/constants you want to reuse between methods.
"""
self.books = ["foo", "bar", "ich", "bein"]
@app.expose("/", defaults={"book_id":-1}) # URL == /book/
def do_list(self, book_id = None): #Just like normal flask, the function/method name can be arbitrary
booklist = [f"""<li><a href="/book/{book_id}/show">{book_name}</a></li>""" for book_id, book_name in enumerate(self.books)]
booklist = "\n".join(booklist)
response = f"""
<ol>
{booklist}
</ol>
<br>
<form method="post" action="/book/add">
<legend> Add a new book</legend>
<label>Book Name
<input type="text" name="book_name" autofocus></input>
</label>
<button type="submit">Submit</button>
</form>
"""
return response
@app.expose("/<int:book_id>/show", methods=["GET"]) # URL == "/book/###/show/"
def show_book(self, book_id=None):
response = f"""
<h1>Book: {self.books[book_id]}</h1>
<a href="/book/{book_id}/remove">Remove</a><br>
<form method="POST" action="/book/{book_id}/replace">
<label>Book Name
<input type="text" name="book_name" value="{self.books[book_id]}" autofocus></input>
</label>
<button type="submit">Submit</button>
</form>
"""
return response
@app.expose("/add", defaults={"book_id":-1}, methods=["POST"]) # URL == "/book/add"
def add_a_book(self, book_id=None):
self.books.append(request.form['book_name'])
return redirect("/book/")
@app.expose("/<int:book_id>/replace", methods=["POST"]) # URL /book/###/replace
def replace_book(self, book_id):
self.books[book_id] = request.form['book_name']
return redirect("/book/")
@app.expose("/<int:book_id>/remove", methods=["GET"]) # URL = "/book/###/remove
def remove_book(self, book_id):
self.books.pop(book_id)
return redirect("/book/")
from flask import Flask
from werkzeug.routing import Submount
import inspect
import functools
import typing as T
from dataclasses import dataclass
@dataclass
class ExposedRule:
uri:str
route_kwargs:dict
ExposedClassDefinition = T.TypeVar("ExposedClassDefinition")
ExposedClassInstance = T.TypeVar("ExposedClassInstance")
EXPOSED_ID = "_EXPOSED"
PREFILTER_ID = "_PREFILTER"
POSTFILTER_ID = "_POSTFILTER"
class CLSEndpointFlask(Flask):
"""
Adds logic to allow class routing
> @app.add_class("/foo")
> class Foo(object):
>
> @app.expose("/bar") #http://my_website/foo/bar
> def do_Bar(self):
> return "/foo/bar says hello"
"""
_instances:T.Dict[str, ExposedClassInstance]
def __init__(self, *args, **kwargs):
Flask.__init__(self, *args, **kwargs) # TODO - Use super() instead?
# Just as a precaution to prevent the class from being garbage collected
self._instances = {}
def _get_name(self, thing:T.Callable):
return thing.__qualname__ \
if hasattr(thing, "__qualname__") else thing.__name__ \
if hasattr(thing, "__name__") else repr(thing)
def _find_exposed(self, cls_def:ExposedClassDefinition, identifier:str=EXPOSED_ID)->T.Iterator:
"""
Originally I used inspect.getmembers but I kept running into a problem where reserved words (ex 'update')
weren't being picked up.
:param cls_def: A class defintion but as written this could be any type of basic object.
:param identifier: The attribute name to look for.
:return:
"""
# skip methods and functions that start with _, assuming they are reserved
names = [name for name in dir(cls_def) if name.startswith("_") is False]
for name, thing in inspect.getmembers(cls_def):
if hasattr(thing, EXPOSED_ID) and (inspect.ismethod(thing) or inspect.isfunction(thing)):
yield self._get_name(thing), thing
def _find_member(self, cls_def:ExposedClassDefinition, identifier:str)->T.Union[bool, T.Callable]:
for name, member in inspect.getmembers(cls_def):
if hasattr(member, identifier) and (inspect.isfunction(member) or inspect.ismethod(member)):
return member
return False
def expose(self, uri:str, **route_args:T.Dict[str, T.Any]):
"""
Behaves similar to flask.route except it just adds a custom attribute "_EXPOSED" to
the decorated method or function
:param uri:
:param route_args:
:return:
"""
def processor(method:T.Callable):
setattr(method, EXPOSED_ID, ExposedRule(uri, route_args))
return method
return processor
def set_prefilter(self, func):
setattr(func, PREFILTER_ID, True)
return func
def _find_prefilter(self, clsdef):
return self._find_member(clsdef, PREFILTER_ID)
def set_postfilter(self, func):
setattr(func, POSTFILTER_ID, True)
return func
def _find_postfilter(self, clsdef):
return self._find_member(clsdef, POSTFILTER_ID)
def add_class(self, class_uri:str, **rule_kwargs:T.Dict[str, T.Any]) -> T.Callable:
"""
Instantiates a generic class Foo(object) and wraps its exposed methods into a werkzeug.Submount
TODO find a way to make this smaller/shorter
:param class_uri: The base URI for a class view ( eg "/base" is prepended to its exposed methods)
:param rule_kwargs: Currently not used and not sane for use
:return:
"""
def decorator(cls_def:ExposedClassDefinition) -> ExposedClassDefinition:
"""
:param cls_def: A class with atleast one method decoratored by CLSEndpointFlask.expose
:return: Returns cls_def
"""
cls_name = cls_def.__name__
assert cls_name not in self._instances, \
f"There is already a {cls_name} in add_class"
cls_obj = self._instances[cls_name] = cls_def()
prefilter = self._find_prefilter(cls_obj)
postfilter = self._find_postfilter(cls_obj)
endpoints = {name:func for name, func in self._find_exposed(cls_obj)}
# {getattr(method, "__qualname__", getattr(method, "__name__")):method
# for (name, method) in inspect.getmembers(cls_obj)
# if inspect.ismethod(method) and hasattr(method, "_EXPOSED")
# }
# Does the class have pre or post filter methods?
if prefilter or postfilter:
# TODO to shorten this scope down, this decorator could be made into a staticmethod of CLSEndpointFlask
def view_method_decorator(method, *args, **kwargs)->str:
if prefilter is not False:
prefilter()
if postfilter is not False:
response = method(*args, **kwargs)
method_name = getattr(method, "__qualname__", getattr(method, "__name__", repr(method)))
if method_name is None:
_, method_name = getattr(method, "__qualname__").rsplit(".",1) # type: str
return postfilter(method_name, response)
else:
return method(*args, **kwargs)
for name in list(endpoints.keys()):
orig_method_rule = getattr(endpoints[name], "_EXPOSED")
partial_method = functools.partial(view_method_decorator, endpoints[name])
# To make it easier for debugging, have the common attributes like __qualname__ copied over to the partial
partial_method = functools.update_wrapper(partial_method, endpoints[name])
setattr(partial_method, "_EXPOSED", orig_method_rule)
endpoints[name] = partial_method
else:
# acknowledge nothing needs to be done to the endpoints
pass
#time to build the rules
rules = []
for endpoint, view_func in endpoints.items(): # type: str, T.Callable
exposed_rule = getattr(view_func, EXPOSED_ID) # type: ExposedRule
methods = exposed_rule.route_kwargs.pop("methods", ["GET"]) # type: T.List[str]
if methods is None:
methods = getattr(view_func, "methods", None) or ("GET",)
rule = self.url_rule_class(exposed_rule.uri, methods=methods, endpoint=endpoint, **exposed_rule.route_kwargs)
rules.append(rule)
if endpoint not in self.view_functions:
self.view_functions[endpoint] = view_func
else:
raise AssertionError(
f"View function {view_func} mapping is overwriting an existing endpoint function: {endpoint}"
)
sub_rule = Submount(class_uri, rules)
self.url_map.add(sub_rule)
return cls_def
return decorator
from __future__ import annotations
"""
This is an example of using CLSEndPointFlask to create a self-contained class.
Personally with MVC I prefer "fat" models to contain my
"""
#stdlib
import typing as T
from logging import getLogger
#framework
from flask import jsonify
#User application
if T.TYPE_CHECKING:
from logging import Logger
from ..lib.cls_endpoint import CLSEndpointFlask
from .. import app # type: CLSEndpointFlask
log = getLogger(__name__) # type: Logger
class SomeGenericPrefilter(object):
"""
Intended to demonstrate that the prefilter method/hook/function/thing can be inherited from a subclass
This could be a reusable authentication class
"""
def _prefilter(self) -> T.NoReturn:
"""
This could be an opportunity to throw a redirect to a login page
or do some pre-validation checks via Flask's request object.
A bit touchier, the prefilter could also make changes to the instance variable `self`
for use with the subsequent methods. I am apprehensive about doing this because I
believe flask is threaded in production so whatever is assigned would need to be
stateless.
:return:
"""
log.debug("Prefilter called")
class ConvertDict2JSON(object):
"""
Same as SomeGenericPrefilter, just a demonstration of a subclass/mixin pattern for use
with the example Page controller below.
"""
def _postfilter(self, method_name, response):
# grab from flask import request to change the content-type to application/json, text/json,
# or whatever is the appropriate type
return jsonify(response) if isinstance(response, dict) else response
@app.add_class("/page")
class Page(SomeGenericPrefilter, ConvertDict2JSON):
"""
Taken and stripped down from my private project `Codex4` which is a generational improvement on
a personal wiki application I wrote ~10 years ago and am now updating to use 3.7/3.8 additions
like f-strings, pathlib, and dataclasses.
Inheriting from SomeGenericPreFilter adds a prefilter before
every method exposed below.
Inheriting from ConvertDict2JSON adds a postfilter which checks if
the response from the exposed methods is of type dict and if so, uses
flasks's jsonify.
This is an example of having a self-contained endpoint for LCRUD (List, Create, Replace, Update, Delete)
"""
@app.expose("/", methods=["GET"])
def list_pages(self):
raw_response = f"""
<a href="./create">Create</a><br>
<h2>Example Page 1</h2>
<a href="./replace/1">Replace</a>
<a href="./update/1">Update</a>
<a href="./delete/1">Delete</a>
<br>
<h2>examples and info endpoints</h2>
<a href="./json/1">Json output</a>
<a href="./debug">Debug</a>
<a href="./class_method">Example class method</a>
"""
return raw_response
@app.expose("/create", methods=["GET"])
def create(self):
return "would create a new page"
@app.expose("/replace/<int:page_id>")
def replace(self, page_id):
return f"Would replace {page_id}"
@app.expose("/update/<int:page_id>")
def do_update(self, page_id):
"""
NOTE: inspect.getmembers doesn't appear to see/list the identifier "update" as it is perhaps seen as
a reserved word I've swapped that out for the bruteforce dir() check.
"""
return f"Would update {page_id}"
@app.expose("/json/<int:page_id>")
def will_be_jsonified(self, page_id):
"""
An example of using ConvertDict2JSON's postfilter
:param page_id:
:return:
"""
return dict(record_id=page_id, text="Hello world", type="Page type")
@app.expose("/debug")
def debug(self):
get_name = lambda func: func.__qualname__ if hasattr(func, "__qualname__") \
else \
func.__name__ if hasattr(func,"__name__") else repr(func)
response = dict()
response['rules'] = {rule.rule: rule.endpoint for rule in app.url_map._rules}
response['endpoints'] = {endpointname: get_name(target) for endpointname, target in app.view_functions.items()}
return response
@devdave

This comment has been minimized.

Copy link
Owner Author

@devdave devdave commented Jan 31, 2020

Just making a quick note

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment