How to use yaml_handler method in dbt-osmosis

Best Python code snippet using dbt-osmosis_python

jekyllnow_handlers.py

Source:jekyllnow_handlers.py

1import os2from base.handlers.github_handler import GithubHandler3from base.handlers.path_handlers import PathHandler4from base.handlers.yaml_handlers import YAMLHandler5from jeklog.handlers.page_handler import AbstractPageHandler6from jekyllnow.constants import JekyllNow7from theJekyllProject.choices import BlogTemplates8from theJekyllProject.dbio import RepoDbIO, SiteDataDbIO9class JekyllNowHandler:10    def __init__(self, user, repo):11        self.user = user12        self.repo = repo13        self.gh_handler = GithubHandler(user, repo)14        self.path = PathHandler(user, repo).create_repo_path()15        self.yaml_handler = YAMLHandler()16    def perform_initial_tasks(self):17        """18        Perform initial tasks. Tasks include:19            * get user token20            * create repo21            * clone repo to the particular location22            * change config of the git repo23            * commit all changes24            * push code to remote25        """26        user_token = self.gh_handler.get_user_token()27        self.gh_handler.create_remote_repo(user_token)28        repo = self.gh_handler.clone_repo(JekyllNow.FORKED_URL, self.path)29        self.gh_handler.change_config(repo)30        self.gh_handler.gh_pages_branch(repo)31        repo_obj = RepoDbIO().get_obj({32            'user': self.user,33            'main': True,34            'repo': self.repo35        })36        self.update_baseurl(repo_obj)37        self.gh_handler.commit_all_changes(repo, 'Intial commit')38        self.gh_handler.push_code(repo)39        self.fill_page_database(repo_obj)40        self.update_template_name(repo_obj)41    def update_baseurl(self, repo_obj):42        """43        update baseurl in config file.44        """45        config_path = os.path.join(self.path, '_config.yml')46        yaml_dict = self.yaml_handler.read_yaml_file(config_path, True)47        if repo_obj.repo != (str(self.user.username) + '.github.io'):48            new_yaml = self.yaml_handler.change_yaml(yaml_dict, {49                'baseurl': "/{}".format(str(repo_obj.repo))})50            self.yaml_handler.write_dict_yaml(config_path, new_yaml)51    def update_template_name(self, repo):52        """53        update the template name to jekyllnow54        """55        RepoDbIO().update_obj(repo, {'template': BlogTemplates.JEKYLL_NOW})56    def fill_page_database(self, repo):57        """58        This method will be used to fill the page database by iterating59        over the pages in the database.60        """61        AbstractPageHandler(self.path).read_pages(repo, 'md', (62            'README.md', '404.md'))63    def perform_site_data(self, data_dict):64        """65        Perform all the site data operations here.66        """67        repo = RepoDbIO().get_obj({68            'user': self.user,69            'main': True,70        })71        config_path = os.path.join(self.path, '_config.yml')72        site_data = SiteDataDbIO().get_obj({'repo': repo})73        if site_data:74            SiteDataDbIO().update_obj(site_data, data_dict)75        else:76            SiteDataDbIO().save_db_instance(data_dict)77        self.del_key(data_dict, 'repo')78        yaml_dict = self.yaml_handler.read_yaml_file(config_path, True)79        new_yaml = self.yaml_handler.change_yaml(yaml_dict, data_dict)80        self.yaml_handler.write_dict_yaml(config_path, new_yaml)81        repo = self.gh_handler.get_repo_from_path(self.path)82        self.gh_handler.commit_all_changes(repo, 'Intial commit')83        self.gh_handler.push_code(repo)84    def del_key(self, dict, key):...

op_yaml.py

Source:op_yaml.py

...27        """åyamlæä»¶åå¥æ°æ®"""28        with open(self.file, encoding=encoding, mode='w') as f:29            return yaml.dump(data, stream=f, allow_unicode=True)30config = os.path.abspath(os.path.join(os.path.dirname(__file__), "../Config/config.yaml"))31conf = yaml_handler(config).read_yaml()["conf"]32mysql = yaml_handler(config).read_yaml()["mysql"]33variable = os.path.abspath(os.path.join(os.path.dirname(__file__), "../TestCase/test_data/{0}".format(conf['variable'])))34var = yaml_handler(variable).read_yaml()["local_variable"]35if __name__ == '__main__':...

main.py

Source:main.py

1#!/usr/bin/env python32import logging3import os4import signal5from iombian_yaml_handler import IoMBianYAMLHandler6from reply_server import ReplyServer7from sub_client import SubClient8logging.basicConfig(format='%(asctime)s %(levelname)-8s - %(name)-16s - %(message)s', level=logging.INFO)9logger = logging.getLogger(__name__)10YAML_FILE_PATH = "/boot/config/parameters.yml"11SERVER_PORT = 555512CLIENT_PORT = 555613RESET_EVENT = "long_long_click"14def config_update_callback():15    logger.info("Rebooting the system")16    os.system('reboot')17def button_event_callback(event):18    logger.debug(f"'{event}' event received")19    if event == RESET_EVENT:20        yaml_handler.reset()21def signal_handler(sig, frame):22    logger.info("Stopping IoMBian Config File Handler")23    server.stop()24    client.stop()25if __name__ == "__main__":26    logger.info("Starting IoMBian Config File Handler")27    yaml_handler = IoMBianYAMLHandler(YAML_FILE_PATH)28    yaml_handler.on_config_update(config_update_callback)29    yaml_handler.load_file()30    server = ReplyServer(yaml_handler, port=SERVER_PORT)31    server.start()32    client = SubClient(on_message_callback=button_event_callback, port=CLIENT_PORT)33    client.start()34    signal.signal(signal.SIGINT, signal_handler)35    signal.signal(signal.SIGTERM, signal_handler)...

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.