Tiny postgresql build script

README.md

Use

Inlines

1. Create a file named "Main.sql" containing something like this:

BEGIN TRANSACTION;
--@inline {filepath0}
--@inline {filepath1}
--[...]
COMMIT;

2. In any of the imported files, use --@inline {path} to include more files. This fits a lot of workflows, but my personal folder structure looks like this:

Schemas/
- AppName/
- - Tables
- - Views
- REFS/
- - Tables
- - Data
Main.sql
- imports Permissions
- imports Schemas
Permissions.sql
Schemas.sql
- imports Schemas/REFS/Tables
- imports Schemas/Refs/Data
- imports Schemas/AppName/Tables
- imports Schemas/AppName/Views

Since the import statement is prefaced by --, it will register as a comment to sql linters and won't cause any trouble.

Conditionals

Conditionals can be specified in the CLI and supports alphanumeric defines.

You can use unary conditions, i.e.

--@if abcdefg
-- something here
--@else
-- something else
--@end

by specifying abcdefg=true, abcdefg=false, or even just abcdefg to check if it's defined. Currently there is no else-if block, but nested conditions function:

--@if something
-- dead block
--@else
--@if something2
-- something 2
--@else
-- something 3
--@end
--@end

Numerics, text, booleans, and plain defines are all supported:

./build.py letters=abc numbers=123 bool=false define

--@if letters < def
-- this would go to the output
--@end
--@if bool == false
-- this would go to the output
--@end
--@if numbers < 200
-- this would go to the output
--@end
--@if define
-- this would go to the output
--@end

These will function recursively, so

--@if someflag
--@inline file1
--@else
--@inline file2
--@end

is fully functional and intended.

Dynamic Values

At any point in a file, use @$NAME_HERE to insert a value defined in CLI args (see Conditionals for how to do that) Both keys and values can be any string of capital and/or lower case letters and underscores. Technically you can use bools/ints/anything python can cast to a string in this too, but I wouldn't exactly call that "intended behavior."

Example:

CREATE TABLE @$Schema_Name.@$TaBlE___NAME (
 value text not null
);

Called with

./sqlbuild.py Schema_Name=SchemaNameHere TaBlE___NAME=tableName__Here

will bundle to

CREATE TABLE SchemaNameHere.tableName__Here (
 value text not null
);

CSV Tables

In any included sql file, insert the line --@csv_insert TABLENAME CSVPATH to insert values from a specified csv file inline. Example:

Row_A,Row_B,Row_C,WhateverRowTitle
val1,NULL,15,22.96
NULL,15,22.96,val1

--@csv_insert TABLENAME_HERE example.csv

This will generate the following sql:

--example.csv
INSERT INTO TABLENAME_HERE (Row_A,Row_B,Row_C,WhateverRowTitle) VALUES
    ('val1',NULL,15,22.96),
    (NULL,15,22.96,'val1');
--end example.csv

Note that filenames are expressly and exclusively compatible with dynamic values! You can use --@csv_insert TABLENAME @$csv_argument and run with sqlbuild.py csv_argument=path/to/csv/file.csv to dynamically insert a csv table.

Limits

IF comparisons:

• These will not automatically converge types, and come with the jank of python's <>= string compares.
• In conditional blocks, comparisons are NOT REVERSIBLE! true == bool will attempt to compare a string variable named true to the string bool.

Update 1:

Added support for conditional blocks using the cli. I didn't test it particularly well. This also required me to import some modules, but it's still fully python core.

Update 0:

This script is hosted on my quick links. You can access it from anywhere with wget/curl/your favorite REST client at https://sh.crinfarr.zip/sqlbuild.py

build.py

#!/usr/bin/python
from os import SEEK_SET
import re
import sys
from io import TextIOWrapper

build = open("compiled.pgsql", "w")
setvals: dict[str, str | int | bool] = {}


def parse_arguments():
    matcher = re.compile(r"^([a-zA-Z0-9_]+)(=([a-zA-Z0-9_]+))?$")
    for arg in sys.argv[1:]:
        matched = matcher.match(arg)
        if matched is None:
            raise ValueError("Could not match args")
        key, _, val = matched.groups()
        if val is None:
            raise ValueError("Invalid val None")
        elif re.match(r"[0-9]+", val):
            setvals[key] = int(val)
        elif val.lower().strip() in ("true", "false"):
            setvals[key] = val == "true"
        elif re.match(r"^[a-zA-Z0-9]+$", val):
            setvals[key] = val
        else:
            raise ValueError(f"Invalid argument: {arg} (how did you do that?)")


def handle_inline(line: str):
    if not line.startswith("--@inline"):
        raise ValueError(f"Inline handler called with non-inline marker {line}")
    fpath = line[9:].strip()
    if (fpath is None) or (fpath == ""):
        raise ValueError(f"Inline handler called with empty path {line}")
    with open(fpath, "r") as f:
        yield f"\n--{fpath}\n"
        for line in f:
            if line.startswith("--@"):
                for cl in handle_component(line, f):
                    yield cl
            else:
                yield line
        yield f"--End {fpath}\n"


def build_conditional(condition: str):
    matcher = re.compile(r"([a-zA-Z0-9]+) ?([<>]=?|[=!]=)? ?([a-zA-Z0-9]+)?")
    matched = matcher.match(condition)
    if matched is None:
        raise ValueError(f"Could not extract conditional from {condition}")
    groups = matched.groups()
    if groups is None:
        raise ValueError(f"Invalid conditional: {condition}")
    key, op, val = groups
    if op is None and val is not None:
        raise ValueError(f"Conditional {condition} does not have a valid operator")
    if (
        key not in setvals
    ) and op is None:  # Op being none implies val is none (^ see check ^)
        return False

    def _compareHandler(v1, op, v2):
        if type(v1) is not type(v2):
            raise ValueError(f"Cannot compare {v1}({type(v1)}) and {v2}({type(v2)})")
        match op:
            case "==":
                if type(v1) in (int, str):
                    return v1 == v2
                if type(v1) is bool and v2:
                    print(
                        f"[Warn]: Unnecessary operator {key}{op}{v2}: Comparing bool to true can be simplified to --@if {key}"
                    )
                return v1 == v2
            case "!=":
                if type(v1) in (int, str):
                    return v1 != v2
                if type(v1) is bool and not v2:
                    print(
                        f"[Warn]: Unnecessary operator {key}{op}{v2}: Comparing bool to not false can be simplified to --@if {key}"
                    )
                return v1 != v2
            case "<":
                if type(v1) is bool:
                    raise ValueError("Bools cannot be numerically compared")
                return v1 < v2
            case ">":
                if type(v1) is bool:
                    raise ValueError("Bools cannot be numerically compared")
                return v1 > v2
            case "<=":
                if type(v1) is bool:
                    raise ValueError("Bools cannot be numerically compared")
                return v1 <= v2
            case ">=":
                if type(v1) is bool:
                    raise ValueError("Bools cannot be numerically compared")
                return v1 >= v2
            case _:
                raise ValueError(f"Invalid operator: {op}")

    if val is not None:
        if re.match(r"^[0-9]+$", val):
            return _compareHandler(setvals[key], op, int(val))
        elif val.lower().strip() in ("true", "false"):
            return _compareHandler(setvals[key], op, val.lower().strip() == "true")
        elif re.match(r"^[a-zA-Z0-9]+$", val):
            return _compareHandler(setvals[key], op, val)
        else:
            raise ValueError(f"Invalid value in conditional: {val}")
    else:  # unary conditional
        return key in setvals


def handle_conditional(component: str, handle: TextIOWrapper):
    if not component.startswith("--@if"):
        raise ValueError(
            f"Conditional handler called with non-conditional marker {component}"
        )
    condition: bool = build_conditional(component[5:].strip())
    if condition:  # conditional active
        yield component  # return conditonal line
        for line in handle:
            if line.startswith("--@"):  # component handler
                if line.strip() == "--@end":
                    yield line  # return end line
                    break
                elif line.strip() == "--@else":
                    for line in handle:  # inactive else block
                        if line.startswith("--@"):
                            for cl in handle_component(line, handle):
                                yield cl  # handle nested ifs and includes
                        elif line.strip() != "--@end":
                            continue  # discard
                        else:
                            return line  # end inactive else block (also end of full conditional block)
                else:
                    for cl in handle_component(line, handle):
                        yield cl  # handle nested ifs and includes
            else:
                yield line  # normal line
    else:  # conditional inactive
        for line in handle:
            if line.startswith("--@"):  # component handler
                if line.strip() == "--@end":
                    break  # end of inactive conditional block
                elif line.strip() == "--@else":
                    yield f"--@else [{component[5:].strip()}]\n"
                    for line in handle:  # active else block
                        if line.startswith("--@"):
                            for cl in handle_component(line, handle):
                                yield cl  # handle nested ifs and includes
                        elif line != "--@end":
                            yield line
                        else:
                            return line
                else:
                    continue  # refuse to handle inactive components in block
            else:
                continue


def handle_component(line: str, handle: TextIOWrapper):
    if not line.startswith("--@"):
        raise ValueError(f"Component handler called with non-component line {line}")
    match line.split(" ")[0]:
        case "--@inline":
            for line in handle_inline(line):
                yield line
        case "--@if":
            for line in handle_conditional(line, handle):
                yield line
        case "--@csv_insert":
            for line in handle_csv(line):
                yield line

float_matcher =  re.compile(r"(?<=[,\n])\d+\.\d+(?=[,\n])")
def normalize_csv_input_types(c:str):
    if c.isdigit() or (float_matcher.match(c) is not None) or c == "NULL":
       return c
    else:
        return f"'{c}'"
def handle_csv(line: str):
    (tablename, filename) = line.split(" ")[1:]
    if filename.startswith("@$"):
        tablename = str(list(handle_dynamic(filename))[0])
    yield(f"--{filename.strip("\n")}\n")
    yield f"INSERT INTO {tablename} "
    with open(filename.strip("\n"), mode="r") as f:
        f.seek(0, SEEK_SET)
        for index, line in enumerate(f):
            if line.strip().strip("\n") == "":
                continue
            if index==0:
                yield f"({line.strip("\n")}) VALUES\n" # use csv header as-is for row headers
            else:
                if index != 1:
                    yield ",\n"
                yield f"\t({",".join(map(lambda s: str(normalize_csv_input_types(s)), line.strip().strip("\n").split(",")))})"
    yield f";\n--End {filename.strip("\n")}\n"


def handle_dynamic(line: str):
    if "@$" not in line:
        raise ValueError(
            f"Dynamic substitution handler called with line {line}, which contains no dynamic calls"
        )
    matched = re.search(r"(@\$)([a-zA-Z_]+)", line)
    if matched is None:
        raise ValueError(f"Could not find dynamic in {line}")

    if matched.group(2) in setvals.keys():
        rep = setvals.get(matched.group(2))
        edited = line.replace(matched.group(0), str(rep))
        if "@$" in edited:
            for line in handle_dynamic(edited):
                yield line
        else:
            yield edited
    else:
        raise ValueError(
            f"Dynamic substitution call {matched.group(0)} does not have a defined value!"
        )


def build_file(path: str):
    yield f"--{path}\n"
    with open(path, "r") as f:
        for line in f:
            if line.startswith("--@"):
                for cl in handle_component(line, f):
                    yield cl
            elif "@$" in line:
                for dl in handle_dynamic(line):
                    yield dl
            else:
                yield line


parse_arguments()
for line in build_file("Main.sql"):
    build.write(line)
build.close()