python类PackageLoader()的实例源码

def encode(self, api_input, action):
        """Transform input to dummy zpl."""
        if not (action in DUMMY_ACTIONS):
            raise Exception(
                'action %s not in %s' % (action, ', '.join(DUMMY_ACTIONS)))

        api = DummyApi()
        if not api.validate(api_input):
            raise InvalidApiInput(
                'Input error : %s' % api.errors(api_input))
        data = api.normalize(api_input)

        data['to_address']['dept'] = data['to_address']['zip'][0:2]

        env = Environment(
            loader=PackageLoader('roulier', '/carriers/dummy/templates'),
            extensions=['jinja2.ext.with_'])

        template = env.get_template("dummy_%s.zpl" % action)
        return template.render(
            auth=data['auth'],
            service=data['service'],
            parcel=data['parcel'],
            sender_address=data['from_address'],
            receiver_address=data['to_address'])

def get_domain_template(distro, libvirt_ver, **kwargs):
    """
    Get a rendered Jinja2 domain template

    Args:
        distro(str): domain distro
        libvirt_ver(int): libvirt version
        kwargs(dict): args for template render

    Returns:
        str: rendered template
    """
    env = Environment(
        loader=PackageLoader('lago', 'providers/libvirt/templates'),
        trim_blocks=True,
        lstrip_blocks=True,
    )

    template_name = 'dom_template-{0}.xml.j2'.format(distro)
    try:
        template = env.get_template(template_name)
    except TemplateNotFound:
        LOGGER.debug('could not find template %s using default', template_name)
        template = env.get_template('dom_template-base.xml.j2')
    return template.render(libvirt_ver=libvirt_ver, **kwargs)

def get_template_engine(self):
        engine = self.get('template_engine')
        if engine is None:
            templates = self['templates']
            dirname = os.path.join(os.getcwd(), 'templates')
            if os.path.isdir(dirname):  # pragma: no cover
                if dirname not in templates:
                    templates.insert(0, dirname)
            elif os.getcwd() not in templates:
                templates.insert(0, os.getcwd())
            loader = jinja2.ChoiceLoader([
                FileSystemLoader(p) for p in templates
            ] + [jinja2.PackageLoader('nuka')])
            self['template_engine'] = jinja2.Environment(
                loader=loader,
                undefined=jinja2.StrictUndefined,
                keep_trailing_newline=True,
                autoescape=False,
            )
        return self['template_engine']

def launchCmd(self, inputs, tmpStore, stdout_file):
        mesh = inputs['mesh']

        absPath = os.path.abspath(os.path.join('/data', mesh.path))

        env = Environment(
            loader=PackageLoader('pkg_codeaster.tools', 'templates')
        )

        templatePara = env.get_template('paraview_warp_by_displacement_template.py')

        with open('disp_warp_output_script.py', 'w') as file:
            file.write(templatePara.render(input_mesh=absPath))

        return {'script_file': 'disp_warp_output_script.py'
                }

def launchCmd(self, inputs, tmpStore, stdout_file):
        mesh = inputs['mesh']

        absPath = os.path.abspath(os.path.join('/data', mesh.path))

        env = Environment(
            loader=PackageLoader('pkg_codeaster.tools', 'templates')
        )

        templatePara = env.get_template('paraview_clip_template.py')

        with open('clip_output_script.py', 'w') as file:
            file.write(templatePara.render(input_mesh=absPath))

        return {'script_file': 'clip_output_script.py'
                }

def launchCmd(self, inputs, tmpStore, stdout_file):
        mesh = inputs['mesh']

        absPath = os.path.abspath(os.path.join('/data', mesh.path))

        env = Environment(
            loader=PackageLoader('pkg_codeaster.tools', 'templates')
        )

        templatePara = env.get_template('paraview_stress_warp_template.py')

        with open('stress_warp_output_script.py', 'w') as file:
            file.write(templatePara.render(input_mesh=absPath))

        return {'script_file': 'stress_warp_output_script.py'
                }

def launchCmd(self, inputs, tmpStore, stdout_file):
        mesh = inputs['mesh']

        absPath = os.path.abspath(os.path.join('/data', mesh.path))

        env = Environment(
            loader=PackageLoader('pkg_codeaster.tools', 'templates')
        )

        templatePara = env.get_template('paraview_surfacegrid_template.py')

        with open('surfacegrid_output_script.py', 'w') as file:
            file.write(templatePara.render(input_mesh=absPath))

        return {'script_file': 'surfacegrid_output_script.py'
                }

def get_mapping(cls):
        from pyplanet.core import Controller

        mapping = dict()

        # Static core components.
        mapping['core.views'] = PackageLoader('pyplanet.views', 'templates')

        # Add app prefixes.
        for app_label, app in Controller.instance.apps.apps.items():
            template_path = os.path.join(app.path, 'templates')
            if os.path.exists(template_path):
                mapping[app_label] = FileSystemLoader(template_path)

        return mapping

def create_index(self):
        """
        Creates an HTML index page to redirect to an MkDocs generated page.
        """
        from jinja2 import Environment, PackageLoader, select_autoescape
        env = Environment(
            loader=PackageLoader('wikidoc', 'templates'),
            autoescape=select_autoescape(['html', 'xml'])
        )
        template = env.get_template('redirect.html')
        html_code = template.render(target_url=self._index)
        file_name = os.path.join(self._out_dir, "index.html")

        if not os.path.exists(file_name):
            with open(file_name, "w") as index_file:
                index_file.write(html_code)

def setup_fallback_env():
    env = jinja2.Environment(loader=jinja2.PackageLoader(__name__))
    env.lstrip_blocks = True
    env.trim_blocks = True
    return env

def render_control(self):
        host_port = self.exposed
        host_name = self.instance_name
        env = Environment(loader=PackageLoader('cthulhu', 'templates'))
        template = env.get_template('instance-control.sh')
        # TODO(sholsapp): How can we make this generic but meaningful? We always will
        # need to know a little bit about the application here, so maybe we should
        # ask for a fixture.spec or something?
        return template.render(
            docker_container=self.container,
            host_name=host_name,
            host_port=host_port,
            local_etc=self.node_etc,
            local_root=self.node_root,
            local_var=self.node_var,
            name=self.instance_name,
        )

def render_control(self):
        env = Environment(loader=PackageLoader('cthulhu', 'templates'))
        template = env.get_template('fixture-control.sh')
        return template.render(
            instances=self.instances,
        )

def __init__(self):
        super(TextFormatter, self).__init__()
        env = Environment(loader=PackageLoader('gixy', 'formatters/templates'), trim_blocks=True, lstrip_blocks=True)
        self.template = env.get_template('text.j2')

def __init__(self):
        super(ConsoleFormatter, self).__init__()
        env = Environment(loader=PackageLoader('gixy', 'formatters/templates'), trim_blocks=True, lstrip_blocks=True)
        self.template = env.get_template('console.j2')

def get_app(loop, data_queue):
    app = web.Application()
    app['sockets'] = []
    mainhandler = MainHandler()
    redishandler = RedisHanlder(data_queue)
    app.router.add_route("GET", '/', mainhandler.get)
    app.router.add_route("*", '/data', redishandler.get)
    app.router.add_static('/static/',
                          path=str(PROJ_ROOT / 'static/css'),
                          name='static')
    aiohttp_jinja2.setup(
        app, loader=jinja2.PackageLoader('src', 'static'))
    srv = await loop.create_server(app.make_handler(), "0.0.0.0", port=9999)
    return srv

def initialize(loop: asyncio.AbstractEventLoop, port: int, username: str, password: str, dbcon: DBConnection,
               active_monitor_manager: ActiveMonitorManager) -> None:
    """Initialize the webmgmt listener."""
    stats.set('num_calls', 0, 'WEBMGMT')
    app = web.Application(loop=loop, logger=log.logger,
                          middlewares=[
                              middleware.logging_middleware_factory,
                              middleware.error_handler_middleware_factory,
                              middleware.basic_auth_middleware_factory,
                          ])
    app['username'] = username
    app['password'] = password
    app['dbcon'] = dbcon
    app['active_monitor_manager'] = active_monitor_manager
    setup_routes(app)
    aiohttp_jinja2.setup(
        app,
        loader=jinja2.PackageLoader('irisett.webmgmt', 'templates'),
        filters={
            'timestamp': jinja_filters.timestamp
        },
    )

    listener = loop.create_server(app.make_handler(), '0.0.0.0', port)
    asyncio.ensure_future(listener)
    log.msg('Webmgmt listening on port %s' % port)

def create_jinja_loader(self):
        return PackageLoader(self.app_name)

def update(self, services, fqdn="flyby.example.com", resolvers=None, log_endpoint=None, log_format=None):
        logger.debug("Updating HAProxy configs...")
        resolvers = resolvers if resolvers else []
        env = Environment(loader=PackageLoader('flyby', 'config'))
        template = env.get_template('haproxy.cfg.j2')
        tempate_params = dict(
            fqdn=fqdn,
            services=self._filter_services(services),
            resolvers=resolvers,
            log_endpoint=log_endpoint,
            log_format=log_format
        )
        c = template.render(**tempate_params)
        if self.config != c:
            logger.debug("Changed configs identified.")
            self.config = c
            self._run()
            logger.debug("HAProxy configs successfully updated.")
        else:
            logger.debug("HAProxy configs up-to-date. Nothing to do.")

def __init__(self, protocol, config):
        self.env = Environment(loader=PackageLoader('piqueserver.web'))
        self.protocol = protocol
        root = Resource()
        root.putChild(b'json', JSONPage(self))
        root.putChild(b'', StatusPage(self))
        root.putChild(b'overview', MapOverview(self))
        site = server.Site(root)

        logging = config.get('logging', False)
        site.noisy = logging
        if not logging:
            site.log = lambda _: None

        protocol.listenTCP(config.get('port', 32886), site)

def __init__(self, template_fn='standard.html'):
        self.output_dir = config.output_dir
        self.path = "./"
        # Initiate jinja template
        env = Environment(
            loader=PackageLoader('sequana', 'resources/templates/')
        )
        self.template = env.get_template(template_fn)
        self._init_report()

def __init__(self, pkg_data):
        """Initialize."""
        self.cwd = os.getcwd()
        self.name = pkg_data.name
        self.outdir = os.path.abspath(pkg_data.outdir)
        self.tmpdir = tempfile.mkdtemp(suffix=self.suffix)
        self.templates = Environment(loader=PackageLoader(self.template_name))
        self.pkg_data = pkg_data

def __init__(self, controller_or_app):
        self.app = controller_or_app.app if hasattr(controller_or_app, 'app') else controller_or_app

        self.env = jinja2.Environment(
            loader=jinja2.ChoiceLoader(
                (
                    jinja2.PackageLoader('pycommunicate'),
                    jinja2.FileSystemLoader(self.app.template_directory)
                )
            )
        )

        def add_includes():
            return self.render_includes()

        self.env.globals.update({
            'add_includes': add_includes
        })

def __init__(self, args):
        self.args = args

        with open(args.input, 'r') as f:
            self.json = json.load(f)

        self._validate_config()
        self._build_api()

        self.name = self.json["Name"]

        self.jinja_env = Environment(loader=PackageLoader('downtoearth', 'templates'))

def render_template(template_name, **kwargs):
    """
    Simple utility function to render out a specified template, using
        **kwargs to fill in variables.

    Args:
        template_path (str): The directory where we can find the template.
        template_name (str): The actual name of the template we want to
                                render.
        **kwargs (dict):     Key Value pairs of any variables we want rendered
                                out into the template.

    Raises:
        AncillaryFileNotFound:      If we cannot find the template.
        AncillaryUndefinedError:    If we run across an undefined variable.

    """

    # Attempt to load a Tempalte file from within the 'Zapper' package
    #   and raise an IOError if I'm unable to find it.
    try:
        env = Environment(loader=PackageLoader('zapper', 'templates'))
        template = env.get_template(template_name)
    except TemplateNotFound:
        raise IOError('Unable to find template {} in zapper!'
                      .format(template_name))

    # Attempt to render our template, and raise a Value Error if we
    #   run into any undefined variables.
    try:
        template_data = template.render(**kwargs)
    except UndefinedError as e:
        raise ValueError('Undefined variable found in {}! Error: {}'
                         .format(template_name, e))

    return template_data

def _generate_data(output_file, context, template_file, default_template):
    if not template_file:
        env = Environment(loader=PackageLoader("proxmoxdeploy.cloudinit"))
        template = env.get_template(default_template)
    else:
        template = Template(template_file.read())

    with open(output_file, "w") as output:
        output.write(template.render(context=context))

def calibrate():
    print("Configuring Wordpress installer")

    log_location = os.path.dirname(__file__) + '/resources/wordpress_installer.log'
    try:
        os.remove(log_location)
    except Exception as e:
        pass

    env = Environment(loader=PackageLoader('wordpress_installer', '/resources/templates'))
    log_config_template = env.get_template('logging_config.ini.j2') 
    log_config = log_config_template.render(LOG_LOCATION=log_location)

    log_config_location = os.path.dirname(__file__) + "/resources/logging_config.ini"

    with open(log_config_location, "w") as fh:
        fh.write(log_config)

    print("Writing Files Complete!")

def create_wordpress_conf(user_dir, db_conf):
    """
    Used to generate a new wordpress configuration file from a jinja2 template.
    Pulls the configuration keys from the wordpress API and injects them into the template.
    Injects the database configuration returned from create_wordpress_database into database details of the template.
    Writes the newly templated configuration file into the wordpress directory.
    """
    logger.debug("Generating wordpress configuration")

    env = Environment(loader=PackageLoader('wordpress_installer', '/resources/templates'))
    template = env.get_template('wp-config.php.j2')     

    def get_wordpress_conf_keys():
        logger.debug("Fetching wordpress configuration")
        response = requests.get("https://api.wordpress.org/secret-key/1.1/salt/")
        return response.text

    wordpress_config = template.render(USER_DIR=user_dir,
                                                DB_NAME=db_conf["db"],
                                                DB_USER=db_conf["user"],
                                                DB_PASSWORD=db_conf["password"],
                                                DB_HOST=db_conf["host"],
                                                KEYS=get_wordpress_conf_keys())
    logger.debug("Wordpress configuration rendered from template, writing to file")

    with open(user_dir + "/public_html/wordpress/wp-config.php", "w") as fh:
        fh.write(wordpress_config)

def gen_dockerfiles(envs):
    tpl_env = Environment(loader=PackageLoader('ibench', 'docker'))
    for env in envs:
        with open(dockerfileName(env),'w') as df:
            df.write(tpl_env.get_template('Dockerfile.tpl').render(env))

#os_default = ['centos','ubuntu']

def set_package_loader(app):
    aiohttp_jinja2.setup(app, loader=jinja2.PackageLoader('tests.contrib.aiohttp.app', 'templates'))

def start(self):
        """Start aiohttp web server."""

        self.app = web.Application()

        aiohttp_jinja2.setup(
            self.app,
            loader=jinja2.PackageLoader('aiohttp_server', 'templates'),
            # enable_async=False,
            auto_reload=False)
        aiohttp_session.setup(self.app, aiohttp_session.SimpleCookieStorage())
        self.add_routes()
        self.handler = self.app.make_handler()
        self.f = self.loop.create_server(self.handler,
                                         host='0.0.0.0',
                                         port=self.port)
        # Event loop is already running, so we await create server instead
        # of run_until_complete
        self.srv = await self.f