How to use yaml method in dbt-osmosis

Best Python code snippet using dbt-osmosis_python

test_yaml.py

Source:test_yaml.py

1import importlib2import unittest3from io import StringIO4from django.core import management, serializers5from django.core.serializers.base import DeserializationError6from django.test import SimpleTestCase, TestCase, TransactionTestCase7from .models import Author8from .tests import SerializersTestBase, SerializersTransactionTestBase9try:10    import yaml11    HAS_YAML = True12except ImportError:13    HAS_YAML = False14YAML_IMPORT_ERROR_MESSAGE = r'No module named yaml'15class YamlImportModuleMock:16    """Provides a wrapped import_module function to simulate yaml ImportError17    In order to run tests that verify the behavior of the YAML serializer18    when run on a system that has yaml installed (like the django CI server),19    mock import_module, so that it raises an ImportError when the yaml20    serializer is being imported.  The importlib.import_module() call is21    being made in the serializers.register_serializer().22    Refs: #1275623    """24    def __init__(self):25        self._import_module = importlib.import_module26    def import_module(self, module_path):27        if module_path == serializers.BUILTIN_SERIALIZERS['yaml']:28            raise ImportError(YAML_IMPORT_ERROR_MESSAGE)29        return self._import_module(module_path)30class NoYamlSerializerTestCase(SimpleTestCase):31    """Not having pyyaml installed provides a misleading error32    Refs: #1275633    """34    @classmethod35    def setUpClass(cls):36        """Removes imported yaml and stubs importlib.import_module"""37        super().setUpClass()38        cls._import_module_mock = YamlImportModuleMock()39        importlib.import_module = cls._import_module_mock.import_module40        # clear out cached serializers to emulate yaml missing41        serializers._serializers = {}42    @classmethod43    def tearDownClass(cls):44        """Puts yaml back if necessary"""45        super().tearDownClass()46        importlib.import_module = cls._import_module_mock._import_module47        # clear out cached serializers to clean out BadSerializer instances48        serializers._serializers = {}49    def test_serializer_pyyaml_error_message(self):50        """Using yaml serializer without pyyaml raises ImportError"""51        jane = Author(name="Jane")52        with self.assertRaises(ImportError):53            serializers.serialize("yaml", [jane])54    def test_deserializer_pyyaml_error_message(self):55        """Using yaml deserializer without pyyaml raises ImportError"""56        with self.assertRaises(ImportError):57            serializers.deserialize("yaml", "")58    def test_dumpdata_pyyaml_error_message(self):59        """Calling dumpdata produces an error when yaml package missing"""60        with self.assertRaisesMessage(management.CommandError, YAML_IMPORT_ERROR_MESSAGE):61            management.call_command('dumpdata', format='yaml')62@unittest.skipUnless(HAS_YAML, "No yaml library detected")63class YamlSerializerTestCase(SerializersTestBase, TestCase):64    serializer_name = "yaml"65    pkless_str = """- model: serializers.category66  pk: null67  fields:68    name: Reference69- model: serializers.category70  fields:71    name: Non-fiction"""72    mapping_ordering_str = """- model: serializers.article73  pk: %(article_pk)s74  fields:75    author: %(author_pk)s76    headline: Poker has no place on ESPN77    pub_date: 2006-06-16 11:00:0078    categories:""" + (79        ' [%(first_category_pk)s, %(second_category_pk)s]' if HAS_YAML and yaml.__version__ < '5.1'80        else '\n    - %(first_category_pk)s\n    - %(second_category_pk)s') + """81    meta_data: []82"""83    @staticmethod84    def _validate_output(serial_str):85        try:86            yaml.safe_load(StringIO(serial_str))87        except Exception:88            return False89        else:90            return True91    @staticmethod92    def _get_pk_values(serial_str):93        ret_list = []94        stream = StringIO(serial_str)95        for obj_dict in yaml.safe_load(stream):96            ret_list.append(obj_dict["pk"])97        return ret_list98    @staticmethod99    def _get_field_values(serial_str, field_name):100        ret_list = []101        stream = StringIO(serial_str)102        for obj_dict in yaml.safe_load(stream):103            if "fields" in obj_dict and field_name in obj_dict["fields"]:104                field_value = obj_dict["fields"][field_name]105                # yaml.safe_load will return non-string objects for some106                # of the fields we are interested in, this ensures that107                # everything comes back as a string108                if isinstance(field_value, str):109                    ret_list.append(field_value)110                else:111                    ret_list.append(str(field_value))112        return ret_list113    def test_yaml_deserializer_exception(self):114        with self.assertRaises(DeserializationError):115            for obj in serializers.deserialize("yaml", "{"):116                pass117@unittest.skipUnless(HAS_YAML, "No yaml library detected")118class YamlSerializerTransactionTestCase(SerializersTransactionTestBase, TransactionTestCase):119    serializer_name = "yaml"120    fwd_ref_str = """- model: serializers.article121  pk: 1122  fields:123    headline: Forward references pose no problem124    pub_date: 2006-06-16 15:00:00125    categories: [1]126    author: 1127- model: serializers.category128  pk: 1129  fields:130    name: Reference131- model: serializers.author132  pk: 1133  fields:...

_travis.py

Source:_travis.py

1import yaml2import datetime3# We need to split the incoming file4with open("index.md", 'r') as reader:5	# There are multiple yaml documents in the index.md	6	blankbit, yaml_to_load, text = reader.read().split("---")7yaml_data = yaml.load(yaml_to_load, Loader=yaml.Loader) # To deal with depreciation warnings, we need Loader=Loader8yaml_data['venue']="Foo" 9yaml_data['address']="Room 123" 10yaml_data['country']="us" 11yaml_data['language']="en" 12yaml_data['latitude']="0"13yaml_data['longitude']="0"14yaml_data['humandate']="Jan 01-02, 2020" 15yaml_data['humantime']="9:00 am - 4:30 pm" 16yaml_data['startdate']= datetime.datetime.today()17yaml_data['enddate']= datetime.datetime.today()18yaml_data['instructor']=["Foo"]19yaml_data['helper']=["Foo"] 20yaml_data['email']=["foo@bar.com"]21yaml_data['collaborative_notes'] =  "http://foo.bar"22yaml_data['eventbrite']="1234567890AB" 23with open("index.md", "w") as writer:24	writer.write("---\n") #blankbit is technically a yaml document25	yaml.dump(yaml_data, writer)26	writer.write("---\n")...

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.