in msm_plugin/lib/management.py [0:0]
def generate_deployment_script(
schema_project_path: str, version: str, overwrite_existing: bool = False) -> str:
"""Generate the deployment script for a release
Args:
schema_project_path: The path to the schema project folder.
version: The version to generate the deployment script for.
overwrite_existing: Whether existing files should be overwritten. Defaults to False.
Returns:
The file name path of the generated script
"""
project_settings = get_project_settings(schema_project_path)
schema_name = project_settings.get("schemaName")
schema_file_name = project_settings.get("schemaFileName")
# Build deployment file path that will be used for output and check that it does not exist
deployment_file_path = os.path.join(
schema_project_path, "releases", "deployment",
f"{schema_file_name}_deployment_{version}.sql")
if os.path.exists(deployment_file_path) and not overwrite_existing:
raise ValueError(
f"The file {deployment_file_path} already exists. Please explicitly allow to replace existing files.")
# Build the version file path that is used as a source and check that it does exist
version_file_path = os.path.join(
schema_project_path, "releases", "versions",
f"{schema_file_name}_{version}.sql")
if not os.path.exists(version_file_path):
raise ValueError(
f"The file {version_file_path} does not exist exists. Please make "
f"sure to prepare the release {version} first.")
# Get the version script file section
target_version_sections = get_file_sections(version_file_path)
version_as_ints = lib.core.convert_version_str_to_list(version)
# Ensure there are no missing updates between 2 released versions
updatable_versions = get_updatable_versions(schema_project_path)
# Get the list of released version and build list of updatable versions
released_versions = get_released_versions(schema_project_path)
i = 0
updatable_versions = []
updatable_versions_sections = {}
while i + 1 < len(released_versions) and released_versions[i] < version_as_ints:
version_from = '%d.%d.%d' % tuple(released_versions[i])
version_to = '%d.%d.%d' % tuple(released_versions[i + 1])
updatable_versions.append(version_from)
updatable_versions_sections[version_from] = {
"version_from": released_versions[i],
"version_to": released_versions[i + 1],
"sections": get_file_sections(os.path.join(
schema_project_path, "releases", "updates",
f"{schema_file_name}_{version_from}_to_{version_to}.sql")),
}
i += 1
if len(released_versions) == 0:
raise ValueError(
"Please prepare a version for release before generating a deployment script.")
# If there is exactly one released versions yet, use the version SQL script as deployment script
if len(released_versions) == 1:
shutil.copyfile(version_file_path, deployment_file_path)
return deployment_file_path
# Prepare the schema version template file
template_folder = os.path.join(Path(__file__).parent.parent, "templates")
schema_deployment_template_file_path = os.path.join(
template_folder, "scripts", "schema_deployment_a.b.c.sql")
if not os.path.exists(schema_deployment_template_file_path):
raise ValueError(
f"The schema release version template file `{schema_deployment_template_file_path}` does not exist.")
with open(schema_deployment_template_file_path, "r") as f:
# Remove copyright line and replace placeholders
schema_deployment_script = Template("".join(f.readlines()[1:]))
schema_deployment_script = schema_deployment_script.safe_substitute({
"license": get_license_text(project_settings=project_settings),
"schema_name": schema_name,
"version_target": version,
"version_comma_str": ", ".join(str(number) for number in version_as_ints),
"section_130_creation_of_helpers":
target_version_sections.get("130", {}).get("sql_content", "").strip("\n"),
"section_140_non_idempotent_schema_objects": indent(
target_version_sections.get("140", {}).get("sql_content", "").strip("\n"), " "),
"section_150_idempotent_schema_objects":
target_version_sections.get("150", {}).get("sql_content", "").strip("\n"),
"section_170_authorization": indent(
target_version_sections.get("170", {}).get("sql_content", "").strip("\n"), " "),
"section_190_removal_of_helpers":
target_version_sections.get("190", {}).get("sql_content", "").strip("\n"),
"updatable_versions": ", ".join(f'"{v}"' for v in updatable_versions),
})
matches = re.finditer(
MSM_LOOP_UPDATABLE_VERSIONS_REGEX, schema_deployment_script, re.MULTILINE | re.DOTALL)
# for match_id, match in enumerate(matches, start=1):
# print("Match {match_id} was found at {start}-{end}: {match}".format(
# match_id=match_id, start=match.start(), end=match.end(), match=match.group()))
# for group_id in range(0, len(match.groups())):
# group_id = group_id + 1
# print("Group {group_id} found at {start}-{end}: {group}".format(group_id=group_id,
# start=match.start(group_id), end=match.end(group_id), group=match.group(group_id)))
for match in reversed(list(matches)):
loop_content_template = match.group(3)
loop_content = ""
needs_indent = match.group(2)
if needs_indent is None or needs_indent == "":
needs_indent = 0
else:
needs_indent = int(needs_indent)
for version_from in updatable_versions_sections:
loop_version_content = loop_content_template
sections_to_replace = reversed(list(
re.finditer(MSM_SECTION_PLACEHOLDER_REGEX, loop_content_template)))
for s in sections_to_replace:
section_id = s.group(1)
sql_content = updatable_versions_sections[version_from]["sections"].get(
section_id, {}).get("sql_content", "").strip("\n")
sql_content_indent = indent(
sql_content, " " * needs_indent) if needs_indent else sql_content
sql_content_indent += "\n"
# If the sql_content has no SQL statements, do not insert
if sql_content_has_no_statement(sql_content_indent):
sql_content_indent = ""
loop_version_content = (
loop_content_template[:s.start()]
+ sql_content_indent
+ loop_content_template[s.end():])
loop_content += Template(loop_version_content).safe_substitute({
"version_from": '%d.%d.%d' % tuple(updatable_versions_sections[version_from]["version_from"]),
"version_to": '%d.%d.%d' % tuple(updatable_versions_sections[version_from]["version_to"]),
})
# Check if the loop_content is empty (TODO: check if it contains no statements) and if so
# insert a placeholder
if len(updatable_versions_sections) > 0 and loop_content == "" and needs_indent > 0:
loop_content = " " * needs_indent + "DO NONE;\n"
if sql_content_has_no_statement(loop_content):
loop_content = ""
schema_deployment_script = (
schema_deployment_script[:match.start()]
+ loop_content
+ schema_deployment_script[match.end():])
# Remove empty sections
schema_deployment_script_sections = get_script_sections(
schema_deployment_script)
remove_empty_sections(
sections=schema_deployment_script_sections,
keep_even_if_empty=["license", "003"])
write_sections_to_file(
file_path=deployment_file_path,
sections=schema_deployment_script_sections,
overwrite_existing=overwrite_existing)
# with open(deployment_file_path, "w") as f:
# f.write(schema_deployment_script)
return deployment_file_path