Source code for kuha_oai_pmh_repo_handler.genshi_loader

#!/usr/bin/env python3
# Author(s): Toni Sissala
# Copyright 2020 Finnish Social Science Data Archive FSD / University of Tampere
# Licensed under the EUPL. See LICENSE.txt for full license.
"""Load genshi templates.

Configure::

    from genshi_loader import add_template_folder, set_template_writer
    add_template_folder(settings.oai_pmh_template_folder)
    set_template_writer(handler.template_writer)

Use as decorator::

    from genshi_loader import OAITemplate

    class Handler:
    ...
        @OAITemplate('error.xml')
        async def build_error_message(self):
            return {'msg': 'there was an error'}

"""
import os.path

from genshi.template import TemplateLoader
from kuha_oai_pmh_repo_handler.constants import TEMPLATE_FOLDER

#: Template folders. There can be multiple.
FOLDERS = []


[docs]def add_template_folder(folder): """Add folder to lookup for templates. :param folder: absolute path to folder containing genshi templates. :type folder: str """ FOLDERS.append(folder)
[docs]def get_template_folder(): """Get template folder. :returns: template folders. :rtype: list """ if not FOLDERS: fallback = os.path.join(os.path.dirname(__file__), TEMPLATE_FOLDER) return [fallback] return FOLDERS
#: Template writer. Function which accepts an iterator as parameter. WRITER = []
[docs]def set_template_writer(writer): """Set template writer. :note: Supports only one template writer. :param writer: Function that writes the template. Must accept an iterator as a paramter. :type writer: :func: """ if WRITER: raise Exception("Template writer already configured") WRITER.append(writer)
[docs]def get_template_writer(): """Get template writer. :returns: template writer :rtype: function """ return WRITER[0]
[docs]class OAITemplate: """OAITemplate class. Decorate functions that should write output to genshi-templates. The decorated function must be an asynchronous function and it must return a dictionary. Example:: from genshi_loader import OAITemplate class Handler: @OAITemplate('error.xml') async def build_error_message(self): ... return {'msg': 'there was an error'} :param template_file: filename of the template to use. :type template_file: str :param template_folder: optional parameter to use a different template folder to lookup for given template_file. :type template_folder: str :raises: :exc:`ValueError` if decorated function returns invalid type. """ def __init__(self, template_file): self.template_file = template_file self.template_writer = None self.loader = None def __call__(self, ctx_func): async def wrapper(ctx): """Asynchronous wrapper to serialize XML for output. Calls :func:`ctx_func` with parameter :obj:`ctx` and gets the returned value. Passes this value to the template and serializes it as XML. Finally sends the XML-serialization to the template_writer function. :note: :func:`ctx_func` and :obj:`ctx` are transparent for this function. The only requirement regarding these objets is that :func:`ctx_func` MUST be asynchronous function and it should return a dictionary. :param ctx: Context that gets passed to the decorated function. :raises: :exc:`ValueError` for invalid return type of :func:`ctx_func` """ if self.loader is None: self.loader = TemplateLoader(get_template_folder()) template = self.loader.load(self.template_file) context = await ctx_func(ctx) if not isinstance(context, dict): raise ValueError("{} returns invalid type {}. Expecting dict".format(ctx_func, type(context))) if self.template_writer is None: self.template_writer = get_template_writer() xml = template.generate(**context).serialize(method='xml', strip_whitespace=True) self.template_writer(ctx, xml) return wrapper