Last active
February 1, 2020 07:09
-
-
Save devdave/be785c777619bdea03fceae47d6ce467 to your computer and use it in GitHub Desktop.
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
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 characters
""" | |
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/") |
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 characters
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 | |
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 characters
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 | |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Just making a quick note