Created
December 6, 2020 20:33
-
-
Save jordic/9941e073410bc0895feb7a0b71e3d016 to your computer and use it in GitHub Desktop.
guillotina asgi server side python render with zpt
This file contains 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 . import templates | |
from guillotina.entrypoint import app as guillo | |
from starlette.applications import Starlette | |
from starlette.responses import PlainTextResponse | |
from starlette.responses import JSONResponse | |
from guillotina.component import get_utility | |
from guillotina.interfaces import IApplication | |
from guillotina.utils.content import get_database | |
from guillotina.utils.content import navigate_to | |
from guillotina import task_vars | |
import logging | |
import httpx | |
import os | |
logger = logging.getLogger("writing") | |
folder = os.path.dirname(__file__) | |
template_folder = os.path.join(folder, "templates") | |
template_folder = os.path.abspath(template_folder) | |
templates.global_init(template_folder, auto_reload=True) | |
app = Starlette() | |
@app.route("/") | |
@templates.template("index.pt") | |
async def home(request): | |
root = get_utility(IApplication, name="root") | |
db = await get_database("db", root) | |
task_vars.db.set(db) | |
tm = db.get_transaction_manager() | |
await tm.begin() | |
task_vars.tm.set(tm) | |
obj = await navigate_to(db, "v1/folder") | |
await tm.abort() | |
return {"param": "world", "obj": obj} | |
@app.route("/demo") | |
async def demo(request): | |
return PlainTextResponse("demo, hola") | |
@app.route("/httpx") | |
async def calling_with_httpx(request): | |
# this uses a feature around httpx, that allos to | |
# make requests to an asgi app without network | |
async with httpx.AsyncClient( | |
app=app, base_url="http://localhost/" | |
) as client: | |
resp = await client.get("/db/v1/", auth=("root", "root")) | |
return JSONResponse(resp.json()) | |
app.add_event_handler("startup", guillo.startup) | |
app.add_event_handler("shutdown", guillo.shutdown) | |
app.mount("/", guillo) | |
# run it: G_CONFIG_FILE=config.yml uvicorn writing:app --port 8765 --reload |
This file contains 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 https://github.com/mikeckennedy/fastapi-chameleon | |
import inspect | |
import os | |
from functools import wraps | |
from typing import Optional | |
from starlette.responses import HTMLResponse | |
from chameleon import PageTemplateLoader, PageTemplate | |
class ChameleonException(Exception): | |
pass | |
__templates: Optional[PageTemplateLoader] = None | |
template_path: Optional[str] = None | |
def global_init(template_folder: str, auto_reload=False, cache_init=True): | |
global __templates, template_path | |
if __templates and cache_init: | |
return | |
if not template_folder: | |
msg = "The template_folder must be specified." | |
raise ChameleonException(msg) | |
if not os.path.isdir(template_folder): | |
msg = f"The specified template folder must be a folder, it's not: {template_folder}" | |
raise ChameleonException(msg) | |
template_path = template_folder | |
__templates = PageTemplateLoader(template_folder, auto_reload=auto_reload) | |
def clear(): | |
global __templates, template_path | |
__templates = None | |
template_path = None | |
def render(template_file: str, **template_data): | |
if not __templates: | |
raise Exception( | |
"You must call global_init() before rendering templates." | |
) | |
page: PageTemplate = __templates[template_file] | |
return page.render(encoding="utf-8", **template_data) | |
def response( | |
template_file: str, mimetype="text/html", status_code=200, **template_data | |
): | |
html = render(template_file, **template_data) | |
return HTMLResponse( | |
content=html, media_type=mimetype, status_code=status_code | |
) | |
def template(template_file: str, mimetype: str = "text/html"): | |
""" | |
Decorate a FastAPI view method to render an HTML response. | |
:param str template_file: The Chameleon template file (path relative to template folder, *.pt). | |
:param str mimetype: The mimetype response (defaults to text/html). | |
:return: Decorator to be consumed by FastAPI | |
""" | |
if not template_file: | |
raise ChameleonException("You must specify a template file.") | |
def response_inner(f): | |
@wraps(f) | |
def sync_view_method(*args, **kwargs): | |
response_val = f(*args, **kwargs) | |
return __render_response(template_file, response_val, mimetype) | |
@wraps(f) | |
async def async_view_method(*args, **kwargs): | |
response_val = await f(*args, **kwargs) | |
return __render_response(template_file, response_val, mimetype) | |
if inspect.iscoroutinefunction(f): | |
return async_view_method | |
else: | |
return sync_view_method | |
return response_inner | |
def __render_response(template_file, response_val, mimetype): | |
# sourcery skip: assign-if-exp | |
if isinstance(response_val, HTMLResponse): | |
return response_val | |
if template_file and not isinstance(response_val, dict): | |
msg = f"Invalid return type {type(response_val)}, we expected a dict or fastapi.Response as the return value." | |
raise Exception(msg) | |
model = response_val | |
html = render(template_file, **model) | |
return HTMLResponse(content=html, media_type=mimetype) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment