master_parse/__init__.py
def generate_commands(
        self, file_name: str, params: Params
    ) -> Tuple[str, Sequence[Command]]:
        """Generates file content for DBC and ipynb use.

        Note:
            If file for output already exists in the current working directory
            it will be overwritten.

        Args:
            file_name (str): The name of the input file that has been exported
            from DBC in py format.

        Returns:
            tuple[str, list[Command]]: A tuple containing the header and a list
            of commands.
        """
        current = ParseState()

        def extend_content(cell_state):
            debug(
                f'Notebook "{file_name}": Cell at line '
                + f"{cell_state.starting_line_number} matches "
                + f"labels {[l.value for l in cell_state.command_labels]}."
            )
            if cell_state.command_code is None:
                cell_state.command_code = self.base_notebook_code
            else:  # Remove %sql, %fs, etc and MAGIC
                pat = Parser.code_to_pattern[current.command_code]
                cell_state.command_content = _transform_match(
                    cell_state.command_content, [pat]
                )
                cell_state.command_content = _replace_text(
                    cell_state.command_content, [_magic_replace]
                )

            if cell_state.command_code not in Parser.master_code:
                # non base notebook commands are inlined automatically
                cell_state.command_labels.add(CommandLabel.INLINE)

            cmd = Command(
                part=self.part,
                code=cell_state.command_code,
                labels=cell_state.command_labels,
                content=cell_state.command_content,
            )

            commands.append(cmd)

        _, file_extension = os.path.splitext(file_name)
        file_extension = file_extension.lower()

        assert (
            file_extension in Parser.extension_to_code
        ), f"Bad file extension for {file_name}"
        self.base_notebook_code = Parser.extension_to_code[file_extension]
        base_comment = _code_to_comment[self.base_notebook_code]

        with codecs.open(
            file_name, "r", _file_encoding_in, errors=_file_encoding_errors
        ) as dbcExport:
            file_contents = dbcExport.readlines()

        commands = []
        current.reset()
        self.part = 0

        line_number_inc = 1
        if len(file_contents) > 0 and _databricks.match(file_contents[0]):
            self.header = file_contents[0]
            # Strip first line if it's the Databricks' timestamp
            file_contents[:1] = []
            line_number_inc += 1

        for i, line in enumerate(file_contents):
            line_number = i + line_number_inc

            if current.starting_line_number is None:
                current.starting_line_number = line_number

            if _command.match(line):
                # New cell. Flush the current one.
                if len(current.command_content) > 0:
                    extend_content(current)

                current.reset()
                continue

            if _new_part.match(line):
                self.part += 1

            # Does this line match any of our special comment strings?
            for pat, label in Parser.pattern_to_label:
                if pat.match(line):
                    if label in DEPRECATED_LABELS:
                        print(
                            f'*** "{file_name}", line {line_number}: '
                            + f'WARNING: "{label.value}" is deprecated.'
                        )
                    current.command_labels.add(label)

            # Does this line match any of the Databricks magic cells?
            for pat, code in Parser.pattern_to_code:
                line2 = line.rstrip()
                m = pat.match(line2)
                if m:
                    if current.command_code is not None:
                        msg = (
                            f'"{file_name}", line {line_number}: '
                            + "multiple magic strings"
                        )
                        raise Exception(msg)
                    current.command_code = code

            line = line.rstrip("\r\n")
            # Lines with only MAGIC do not reimport
            if len(line) <= 8 and line[-5:] == "MAGIC":
                # Keep empty lines
                current.command_content.append(base_comment + " MAGIC ")
            else:
                current.command_content.append(line)

        # Flush anything left.
        if len(current.command_content) > 0:
            current.command_content.append("")
            extend_content(current)

        return self.header, commands
Generates file content for DBC and ipynb use.

        Note:
            If file for output already exists in the current working directory
            it will be overwritten.

        Args:
            file_name (str): The name of the input file that has been exported
            from DBC in py format.

        Returns:
            tuple[str, list[Command]]: A tuple containing the header and a list
            of commands.
        
Generates file content for DBC and ipynb use.
Note:
If file for output already exists in the current working directory
it will be overwritten.

file_name (str): The name of the input file that has been exported
from DBC in py format.

tuple[str, list[Command]]: A tuple containing the header and a list
of commands.
Generates file content for DBC and ipynb use . Note : If file for output already exists in the current working directory it will be overwritten . file_name ( str ) : The name of the input file that has been exported from DBC in py format . tuple [ str list [ Command ]] : A tuple containing the header and a list of commands .
def generate_commands(
        self, file_name: str, params: Params
    ) -> Tuple[str, Sequence[Command]]:
        current = ParseState()
        def extend_content(cell_state):
            debug(
                f'Notebook "{file_name}": Cell at line '
                + f"{cell_state.starting_line_number} matches "
                + f"labels {[l.value for l in cell_state.command_labels]}."
            )
            if cell_state.command_code is None:
                cell_state.command_code = self.base_notebook_code
            else:  
                pat = Parser.code_to_pattern[current.command_code]
                cell_state.command_content = _transform_match(
                    cell_state.command_content, [pat]
                )
                cell_state.command_content = _replace_text(
                    cell_state.command_content, [_magic_replace]
                )
            if cell_state.command_code not in Parser.master_code:
                cell_state.command_labels.add(CommandLabel.INLINE)
            cmd = Command(
                part=self.part,
                code=cell_state.command_code,
                labels=cell_state.command_labels,
                content=cell_state.command_content,
            )
            commands.append(cmd)
        _, file_extension = os.path.splitext(file_name)
        file_extension = file_extension.lower()
        assert (
            file_extension in Parser.extension_to_code
        ), f"Bad file extension for {file_name}"
        self.base_notebook_code = Parser.extension_to_code[file_extension]
        base_comment = _code_to_comment[self.base_notebook_code]
        with codecs.open(
            file_name, "r", _file_encoding_in, errors=_file_encoding_errors
        ) as dbcExport:
            file_contents = dbcExport.readlines()
        commands = []
        current.reset()
        self.part = 0
        line_number_inc = 1
        if len(file_contents) > 0 and _databricks.match(file_contents[0]):
            self.header = file_contents[0]
            file_contents[:1] = []
            line_number_inc += 1
        for i, line in enumerate(file_contents):
            line_number = i + line_number_inc
            if current.starting_line_number is None:
                current.starting_line_number = line_number
            if _command.match(line):
                if len(current.command_content) > 0:
                    extend_content(current)
                current.reset()
                continue
            if _new_part.match(line):
                self.part += 1
            for pat, label in Parser.pattern_to_label:
                if pat.match(line):
                    if label in DEPRECATED_LABELS:
                        print(
                            f'*** "{file_name}", line {line_number}: '
                            + f'WARNING: "{label.value}" is deprecated.'
                        )
                    current.command_labels.add(label)
            for pat, code in Parser.pattern_to_code:
                line2 = line.rstrip()
                m = pat.match(line2)
                if m:
                    if current.command_code is not None:
                        msg = (
                            f'"{file_name}", line {line_number}: '
                            + "multiple magic strings"
                        )
                        raise Exception(msg)
                    current.command_code = code
            line = line.rstrip("\r\n")
            if len(line) <= 8 and line[-5:] == "MAGIC":
                current.command_content.append(base_comment + " MAGIC ")
            else:
                current.command_content.append(line)
        if len(current.command_content) > 0:
            current.command_content.append("")
            extend_content(current)
        return self.header, commands
def generate_commands ( self , file_name : str , params : Params ) -> Tuple [ str , Sequence [ Command ] ] : current = ParseState ( ) def extend_content ( cell_state ) : debug ( f'Notebook "{file_name}": Cell at line ' + f"{cell_state.starting_line_number} matches " + f"labels {[l.value for l in cell_state.command_labels]}." ) if cell_state . command_code is None : cell_state . command_code = self . base_notebook_code else : pat = Parser . code_to_pattern [ current . command_code ] cell_state . command_content = _transform_match ( cell_state . command_content , [ pat ] ) cell_state . command_content = _replace_text ( cell_state . command_content , [ _magic_replace ] ) if cell_state . command_code not in Parser . master_code : cell_state . command_labels . add ( CommandLabel . INLINE ) cmd = Command ( part = self . part , code = cell_state . command_code , labels = cell_state . command_labels , content = cell_state . command_content , ) commands . append ( cmd ) _ , file_extension = os . path . splitext ( file_name ) file_extension = file_extension . lower ( ) assert ( file_extension in Parser . extension_to_code ) , f"Bad file extension for {file_name}" self . base_notebook_code = Parser . extension_to_code [ file_extension ] base_comment = _code_to_comment [ self . base_notebook_code ] with codecs . open ( file_name , "r" , _file_encoding_in , errors = _file_encoding_errors ) as dbcExport : file_contents = dbcExport . readlines ( ) commands = [ ] current . reset ( ) self . part = 0 line_number_inc = 1 if len ( file_contents ) > 0 and _databricks . match ( file_contents [ 0 ] ) : self . header = file_contents [ 0 ] file_contents [ : 1 ] = [ ] line_number_inc += 1 for i , line in enumerate ( file_contents ) : line_number = i + line_number_inc if current . starting_line_number is None : current . starting_line_number = line_number if _command . match ( line ) : if len ( current . command_content ) > 0 : extend_content ( current ) current . reset ( ) continue if _new_part . match ( line ) : self . part += 1 for pat , label in Parser . pattern_to_label : if pat . match ( line ) : if label in DEPRECATED_LABELS : print ( f'*** "{file_name}", line {line_number}: ' + f'WARNING: "{label.value}" is deprecated.' ) current . command_labels . add ( label ) for pat , code in Parser . pattern_to_code : line2 = line . rstrip ( ) m = pat . match ( line2 ) if m : if current . command_code is not None : msg = ( f'"{file_name}", line {line_number}: ' + "multiple magic strings" ) raise Exception ( msg ) current . command_code = code line = line . rstrip ( "\r\n" ) if len ( line ) <= 8 and line [ - 5 : ] == "MAGIC" : current . command_content . append ( base_comment + " MAGIC " ) else : current . command_content . append ( line ) if len ( current . command_content ) > 0 : current . command_content . append ( "" ) extend_content ( current ) return self . header , commands
Generates file content for DBC and ipynb use.
Generates file content for DBC and ipynb use .
"""Generates file content for DBC and ipynb use.

        Note:
            If file for output already exists in the current working directory
            it will be overwritten.

        Args:
            file_name (str): The name of the input file that has been exported
            from DBC in py format.

        Returns:
            tuple[str, list[Command]]: A tuple containing the header and a list
            of commands.
        """ # Remove %sql, %fs, etc and MAGIC # non base notebook commands are inlined automatically # Strip first line if it's the Databricks' timestamp # New cell. Flush the current one. # Does this line match any of our special comment strings? # Does this line match any of the Databricks magic cells? # Lines with only MAGIC do not reimport # Keep empty lines # Flush anything left.
self
file_name
str
params
Params