How to use _get_config_template method in molecule

Best Python code snippet using molecule_python

config_tasks.py

Source:config_tasks.py

...10class CommonParameters(NamedTuple):11    id_source: str12    source_name: str13    extraction_type: str14def _get_config_template(extraction_type: str) -> ExtractionTemplate:15    return get_table_item('#extraction_config#', extraction_type)16def _resolve_dynamic_parameters(17        config_kwargs: TemplateParameters,18        common_parameters: CommonParameters19):20    resolved = {}21    for entry_name, entry in config_kwargs.items():22        if isinstance(entry, dict):23            if 'dynamic' in entry:24                if entry['dynamic'] is True:25                    dynamic_type = entry['dynamic_type']26                    dynamic_kwargs = entry.get('dynamic_kwargs', {})27                    dynamic_kwargs['common_parameters'] = common_parameters28                    resolving_fn = getattr(dynamic_extraction_kwargs, dynamic_type)29                    entry = resolving_fn(**dynamic_kwargs)30                    resolved[entry_name] = entry31                    continue32            entry = _resolve_dynamic_parameters(entry, common_parameters)33        resolved[entry_name] = entry34    return resolved35def _build_source_config(36        id_source: str,37        source_name: str38):39    item = get_table_item('#id_source#', id_source)40    if item is None:41        raise RuntimeError(f'id_source: {id_source} is not registered with the system')42    source = item['sources'].get(source_name)43    if source is None:44        raise RuntimeError(f'source: {source_name} is not configured for: {id_source}')45    source_config = source['config']46    config_values = {x: y for x, y in source_config.items()}47    return config_values48def _inject_runtime_parameters(49        runtime_parameters: Sequence[RuntimeParameter],50        resolved_parameters: Mapping[str, TemplatePrimitive],51        extraction_params: Mapping[str, TemplatePrimitive]52):53    parameters = deepcopy(resolved_parameters)54    for runtime_parameter in runtime_parameters:55        insert_path = runtime_parameter['insert_path']56        parameter_name = runtime_parameter['parameter_name']57        parameter_value = extraction_params[parameter_name]58        parameters = insert_json(parameters, insert_path, parameter_value)59    return parameters60def _build_template_parameter_set(61        parameter_set: ParameterSet,62        extraction_params,63        common_params: CommonParameters64):65    static_parameters = parameter_set.get('static_parameters', {})66    resolved_parameters = _resolve_dynamic_parameters(static_parameters, common_params)67    runtime_parameters = parameter_set.get('runtime_parameters', [])68    completed_parameters = _inject_runtime_parameters(runtime_parameters, resolved_parameters, extraction_params)69    return completed_parameters70def _build_extraction_config(71        id_source: str,72        source_name: str,73        extraction_type: str,74        extraction_params: Mapping[str, Any]75) -> ExtractionConfig:76    commons = CommonParameters(id_source, source_name, extraction_type)77    template = _get_config_template(extraction_type)78    if not template:79        raise UnknownExtractionTypeException(f'could not load an extraction template for {extraction_type}')80    extractor_template = template['extractor']81    processor_template = template['processor']82    return ExtractionConfig(83        id_source=id_source,84        source_name=source_name,85        extraction_type=extraction_type,86        extractor_class=extractor_template['extractor_class'],87        extractor_params=_build_template_parameter_set(extractor_template, extraction_params, commons),88        processor_class=processor_template['processor_class'],89        processor_params=_build_template_parameter_set(processor_template, extraction_params, commons),90        extracted_object_type=template['extracted_object_type'],91        extraction_params=_build_template_parameter_set(template['extraction'], extraction_params, commons)...

config.py

Source:config.py

...36        base = self.get(self.parameters.template)37        for key in value.split("."):38            base = base[key]39        return base40def _get_config_template() -> Dict:41    ##########################42    api_versions = Choice(choices={"v1"})43    server_template = {44        "host": str,45        "port": str,46        "api_version": Optional(api_versions, default="v1", allow_missing=True),47    }48    ##########################49    auth_modes = Choice(choices={"username", "api", "token"})50    auth_user_template = {51        "username": str,52        "password": str,53        "authentication_mode": Optional(54            auth_modes, default="username", allow_missing=True55        ),56    }57    auth_api_template = {58        "api_key": str,59        "authentication_mode": Optional(60            auth_modes, default="api_key", allow_missing=True61        ),62    }63    token_template = {64        "token": str,65        "authentication_mode": Optional(66            auth_modes, default="token", allow_missing=True67        ),68    }69    auth_template = OneOf([auth_user_template, auth_api_template, token_template])70    ###########################71    context_template = {"namespace": str}72    ###########################73    template = {74        "server": server_template,75        "auth": auth_template,76        "context": context_template,77    }78    return template79def get_config_parameters():80    default_values = {81        # "server.host": "localhost",82        # "server.port": "8000",83        # "context.namespace": "default",84    }85    redacted_fields = {86        "auth.password": True,87        "auth.api_key": True,88    }89    template = _get_config_template()90    return ConfuseParameters(template, default_values, redacted_fields)...

__init__.py

Source:__init__.py

1from pathlib import Path2import jinja23import gliderport4TEMPLATE_DIR = Path(gliderport.__path__[0]) / "preset/template"5def _get_config_template(job_name):6    config = TEMPLATE_DIR / f"{job_name}.config_template.jinja2"7    if not config.exists():8        raise FileNotFoundError(f"preset {config} not found.")9    return config10def _get_sky_template(job_name):11    config = TEMPLATE_DIR / f"{job_name}.sky_template.jinja2"12    if not config.exists():13        raise FileNotFoundError(f"preset {config} not found.")14    return config15def _rander_jinja2(template_path, **kwargs):16    with open(template_path) as f:17        template = jinja2.Template(f.read())18        content = template.render(kwargs, undefined=jinja2.StrictUndefined)19    return content20def rander_preset_config(name, output_path, **kwargs):21    """Get preset config yaml dict."""22    path = _get_config_template(name)23    content = _rander_jinja2(path, **kwargs)24    with open(output_path, "w") as f:25        f.write(content)26    return27def rander_preset_sky(name, output_path, **kwargs):28    """Get preset sky yaml dict."""29    path = _get_sky_template(name)30    content = _rander_jinja2(path, **kwargs)31    with open(output_path, "w") as f:32        f.write(content)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.