Source code for zensols.db.parse

"""Simple parsing of a dynamic definition langauge (DDL) and dynamic
manipulation language (DML) files.

"""
__author__ = 'Paul Landes'

from typing import Tuple, Dict, List, Iterable, ClassVar
from dataclasses import dataclass, field
import logging
import re
import itertools as it
from pathlib import Path
from zensols.persist import persisted

logger = logging.getLogger(__name__)




@dataclass
class DynamicDataParser(object):
    """Parse a DDL/DML file meant also for prototyping.

    For example the file::

        -- meta=init_sections=create_tables,create_idx
        -- name=create_idx
        create index person_name on person(name);

        -- name=create_tables
        create table person (id int, name text, age int);

    Would have ``create_idx`` and ``create_tables`` as sections and meta data::

        {'init_sections':
         'create_tables,create_idx'}

    """
    COMMENT_PAT: ClassVar[re.Pattern] = re.compile(r'^--.*')
    SEC_START_PAT: ClassVar[re.Pattern] = re.compile(
        r'^-- name=([a-zA-Z0-9_]+)')
    META_PAT: ClassVar[re.Pattern] = re.compile(
        r'^-- meta=([a-zA-Z0-9_]+)=(.+)$')

    dd_path: Path = field()
    """The path of the file to parse."""

    def _map_section_content(self, lines: List[str]) -> str:
        return '\n'.join(lines)

    @persisted('__parse')
    def _parse(self) -> Tuple[Dict[str, str], Dict[str, str]]:
        if logger.isEnabledFor(logging.INFO):
            logger.info(f'parsing {self.dd_path}')
        secs: List[str, Tuple[str, List[str]]] = []
        sec_content: List[str] = []
        meta: Dict[str, str] = {}
        with open(self.dd_path) as f:
            line: str
            for line in f.readlines():
                line = line.rstrip()
                if len(line) == 0:
                    continue
                if re.match(self.COMMENT_PAT, line):
                    if logger.isEnabledFor(logging.DEBUG):
                        logger.debug(f'matched comment: {line}')
                    sec_start = re.match(self.SEC_START_PAT, line)
                    meta_match = re.match(self.META_PAT, line)
                    sec_content = []
                    if sec_start is not None:
                        name = sec_start.group(1)
                        secs.append((name, sec_content))
                    elif meta_match is not None:
                        meta[meta_match.group(1)] = meta_match.group(2)
                else:
                    sec_content.append(line)
        sections = {x[0]: self._map_section_content(x[1]) for x in secs}
        return sections, meta

    @property
    def sections(self) -> Dict[str, str]:
        """Return the sections of the file.

        """
        return self._parse()[0]

    @property
    def metadata(self) -> Dict[str, str]:
        """Return the meta data found int he parse object.

        """
        return self._parse()[1]



    def get_init_db_sqls(self) -> Iterable[str]:
        """Return the set of statements that create all DB objects needed to
        fully CRUD.

        """
        init_secs = self.metadata['init_sections']
        secs = init_secs.split(',')
        entries = map(lambda x: self.sections[x], secs)
        sts = map(lambda x: re.split(';[ \t\n]*', x, flags=re.MULTILINE),
                  entries)
        return filter(lambda x: len(x) > 0, it.chain(*sts))