Upgrade / Evolution for CI/CD

evolution.py

import re
import os
import glob
import collections
import contextlib

PatchEntry = collections.namedtuple('PatchEntry', 'version,path')

patch_key = lambda x: x.version

@contextlib.contextmanager
def in_dir(dirname):
    origdir = os.path.abspath(os.curdir)
    try:
        os.chdir(dirname)
        yield dirname
    finally:
        os.chdir(origdir)


def list_patches(path, pattern=r'^(?P<version_id>\d{4}-\d{3})-.*\.sql$'):
    
    def file_filter(file_list):
        for fname in file_list:
            m = re.match(pattern, os.path.basename(fname))
            if m:
                yield PatchEntry(m['version_id'], fname)

    return [f for f in file_filter(glob.glob(path + '/*'))]


def filter_patches(patch_list, current_version):
    filtered_list = [f for f in patch_list if f.version > current_version]
    return filtered_list

def combine_patches(patch_list):
    def get_patch(patch):
        with open(patch.path, 'r') as f:
            return f.read()

    full_patch = "\n".join(
        [get_patch(patch) for patch in sorted(patch_list, key=patch_key)])
    return full_patch


def build_patch(path, current_version, patch_filename=None,
                name_pattern=r'^(?P<version_id>\d{4}-\d{3})-.*\.sql$'):
    patch_list = list_patches(path, name_pattern)
    filtered_list = filter_patches(patch_list, current_version)
    patch_content = combine_patches(filtered_list)
    if patch_filename is not None:
        with open(patch_filename, 'w') as f:
            f.write(patch_content)
    return patch_content