Skip to content

Instantly share code, notes, and snippets.

@rbellamy

rbellamy/postsplit.py

Created November 6, 2020 19:01
Show Gist options
  • Save rbellamy/10b6f44f657f4ea0bd9e624d9912e6ae to your computer and use it in GitHub Desktop.
Save rbellamy/10b6f44f657f4ea0bd9e624d9912e6ae to your computer and use it in GitHub Desktop.
Download ZIP
postway and postsplit - a FlyWay-like tool for raw psql and a DAG-based splitter for pg_dump
Raw
postsplit.py
#!/usr/bin/env python
import argparse
import logging
import os
import re
from enum import Enum
from itertools import groupby, chain
# import matplotlib.pyplot as plt
import networkx as nx
class ParentName(Enum):
none = 1
before = 2
after = 3
def __str__(self):
return self.name
@staticmethod
def from_string(s):
try:
return ParentName[s]
except KeyError:
raise ValueError()
class PostSplit:
"""A class object for the PG dump split utility script.
Split pg_dump files to schema/schema_type file hierarchy
Use with files produced by pg_dump -s
Original author: kmatt - from https://gist.github.com/kmatt/2572360
https://gist.github.com/rbellamy/e61bbe89abe97afb0fd6150190856f4f
"""
def __init__(self):
self.parser = argparse.ArgumentParser(
description='A script for parsing a text pg_dump file and creating either a) per-object files, or b) '
'per-object files in a form consumable by postway/postway')
self.logger = logging.getLogger('pg_dump_split')
self.logger.setLevel(logging.DEBUG)
self.console_handler = logging.StreamHandler() # sys.stderr
self.console_handler.setLevel(
logging.CRITICAL) # set later by set_log_level_from_verbose() in interactive sessions
self.console_handler.setFormatter(
logging.Formatter('[%(levelname)s] %(message)s'))
self.logger.addHandler(self.console_handler)
self.create_public_schema_sql = '''
CREATE SCHEMA public;
GRANT ALL ON SCHEMA public TO postgres;
GRANT ALL ON SCHEMA public TO PUBLIC;
COMMENT ON SCHEMA public IS 'standard public schema';
'''
self.args = None
def _get_sections(self):
infile = self.args.infile
with open(infile) as f:
groups = groupby(f, lambda x: x.startswith('-- TOC'))
for key, value in groups:
if key:
yield chain([next(value)], (next(groups)[1])) # all lines up to -- TOC
def _add_edges_by_parent_name(self, graph):
"""assign edges based on parent_name inspection"""
for parent_n, parent_d in graph.nodes(data=True):
if parent_d['schema_type'] == 'table':
parent_schema = parent_d['schema']
parent_object_name = parent_d['object_name']
for child_n, child_d in graph.nodes(data=True):
child_schema_type = child_d['schema_type']
child_schema = child_d['schema']
child_object_name = child_d['parent_name']
if 'parent_name' in child_d and \
parent_schema == child_schema and \
parent_object_name == child_d['parent_name']:
self.logger.debug(
'[toc table <- {}] edge: {}.{}({}) <- {}.{}({})'.format(child_schema_type,
parent_schema,
parent_object_name,
parent_n,
child_schema,
child_object_name,
child_n))
graph.add_edge(child_n, parent_n)
return graph
def _add_edges_by_sql(self, parent_schema_types, child_schema_types, graph):
"""
- if parent 'object_name' is in 'sql' of child, then parent is dependency of child.
- if 'myent_pkg.r' is in 'check_missing_goals_setup' function, then 'myent_pkg'(TOC == 1765) is a dependency of
'check_missing_goals_setup'(TOC == 4054)
>>> graph.add_edge(1765, 4054)
:param parent_schema_types:
:type parent_schema_types:
:param child_schema_types:
:type child_schema_types:
:param graph:
:type graph:
:return:
:rtype:
"""
for outer_n, outer_d in graph.nodes(data=True):
if outer_d['schema_type'] in parent_schema_types:
parent_schema = outer_d['schema']
parent_object_name = outer_d['object_name']
for inner_n, inner_d in graph.nodes(data=True):
if inner_d['schema_type'] in child_schema_types:
child_schema = inner_d['schema']
child_object_name = inner_d['object_name']
child_sql = inner_d['sql']
if self._is_dependent_sql(parent_schema,
parent_object_name,
child_schema,
child_object_name,
child_sql):
self.logger.debug('[{} <- {}] edge: {}.{}({}) <- {}.{}({})'.format(
'/'.join(parent_schema_types),
'/'.join(child_schema_types),
parent_schema,
parent_object_name,
outer_n,
child_schema,
child_object_name,
inner_n))
graph.add_edge(inner_n, outer_n)
return graph
def _merge_acl(self, graph):
"""
apply schema_type == 'acl' to all types - this effectively removes the acl type
*NOTE: There is a bug - currently does not correctly match with foreign tables, sequences and views*
"""
acl_delete_list = []
for outer_n, outer_d in graph.nodes(data=True):
parent_schema = outer_d['schema']
parent_schema_type = outer_d['schema_type']
parent_object_name = outer_d['object_name']
if parent_schema_type != 'acl':
for inner_n, inner_d in graph.nodes(data=True):
if inner_d['schema_type'] == 'acl':
acl_schema = inner_d['schema']
acl_parent_schema_type = self._get_acl_parent_type(inner_d['sql'])
acl_object_name = inner_d['object_name']
if parent_schema_type == acl_parent_schema_type and \
parent_object_name == acl_object_name and \
parent_schema == acl_schema:
self.logger.debug(
'[{} <- acl] merge: {}.{}({}) <- {}.{}({})'.format(parent_schema_type,
parent_schema,
parent_object_name,
outer_n,
acl_schema,
acl_object_name,
inner_n))
graph.nodes[outer_n]['sql'] = '{}\n{}'.format(outer_d['sql'], inner_d['sql'])
acl_delete_list.append(inner_n)
graph.remove_nodes_from(acl_delete_list)
return graph
@staticmethod
def _is_dependent_sql(parent_schema, parent_object_name, child_schema, child_object_name, sql):
"""
1. If the fully-qualified parent and child names match, then no match.
2. If the schema names are different, then the parent must be matched with a fully-qualified name in the child.
3. If the schema names are the same, then the parent must be matched with a relative name, if a fully-qualified
parent with the same relative name, but different schema, hasn't already matched the child.
:param parent_schema:
:type parent_schema:
:param parent_object_name:
:type parent_object_name:
:param child_schema:
:type child_schema:
:param child_object_name:
:type child_object_name:
:param sql:
:type sql:
:return:
:rtype:
"""
parent = '{}\.{}'.format(parent_schema, parent_object_name)
child = '{}\.{}'.format(child_schema, child_object_name)
relative_regex = re.compile(r'.*[\s,;\(\)]{}[\s,;\(\)].*'.format(parent_object_name))
fully_qualified_regex = re.compile(r'.*[\s,;\(\)]{}[\s,;\(\)].*'.format(parent))
is_dependent_sql = False
for line in sql.splitlines(keepends=True):
if is_dependent_sql or parent == child:
break
if not line.startswith('-- Name'):
if parent_schema != child_schema:
is_dependent_sql = fully_qualified_regex.match(line)
else:
is_dependent_sql = relative_regex.match(line)
return is_dependent_sql
@staticmethod
def _header(owner, schema, set_role):
s0 = 'BEGIN;'
if set_role and owner:
s1 = 'SET LOCAL ROLE {};'.format(owner)
else:
s1 = ''
s2 = 'SET LOCAL check_function_bodies = false;'
s3 = 'SET SEARCH_PATH TO {}, pg_catalog, sys, dbo;'.format(schema)
return '{}\n{}\n{}\n{}\n'.format(s0, s1, s2, s3)
@staticmethod
def _footer():
return 'COMMIT;'
def _drop_trigger(self, schema, table_name, trigger_name):
self.logger.debug('table_name: {}, trigger_name: {}'.format(table_name, trigger_name))
s1 = 'DROP TRIGGER IF EXISTS {} ON {}.{};'.format(trigger_name.lower(), schema, table_name)
s2 = 'DROP TRIGGER IF EXISTS "{}" ON {}.{};'.format(trigger_name.upper(), schema, table_name)
return '{}\n{}\n'.format(s1, s2)
@staticmethod
def _drop_foreign_table(object_name):
s1 = 'DROP FOREIGN TABLE IF EXISTS {};'.format(object_name)
return s1
@staticmethod
def _inject_create_or_replace(schema_type, sql):
if schema_type != 'type':
sql = re.sub(r'CREATE\s{}'.format(schema_type.upper()),
'CREATE OR REPLACE {}'.format(schema_type.upper()),
sql)
return sql
@staticmethod
def _fixup_quoted_create(schema, schema_type, object_name, sql):
sql = re.sub(r'CREATE(.*{}.*)(\"{}\")'.format(schema_type.upper(), object_name),
'CREATE\\1{}'.format(object_name.lower()),
sql)
return sql
@staticmethod
def _inject_postway_separator(schema_type, object_name, sql):
indices = [s.start() for s in re.finditer('CREATE OR REPLACE', sql)]
if len(indices) == 2:
sql = '{}/\n{}'.format(sql[:indices[1]], sql[indices[1]:])
if schema_type == 'package':
indices = [s.start() for s in re.finditer('ALTER PACKAGE'.format(object_name), sql)]
if len(indices) == 1:
sql = '{}/\n{}'.format(sql[:indices[0]], sql[indices[0]:])
return sql
@staticmethod
def _get_acl_parent_type(sql):
parent_types = ['aggregate', 'foreign_table', 'function', 'materialized_view',
'package', 'procedure', 'schema', 'server', 'table', 'trigger', 'type', 'view']
acl_sql = ''.join([line for line in sql.splitlines(keepends=True) if not line.startswith('-- Name')])
return next((pt for pt in parent_types if pt.upper() in acl_sql), None)
def _write_file(self, outdir, schema, schema_type, fname, sql):
self.logger.info('Schema: {}; Type: {}; Name: {}'.format(schema, schema_type, fname))
sqlpath = os.path.join(outdir, schema, schema_type)
if not os.path.exists(sqlpath):
print('*** mkdir {}'.format(sqlpath))
os.makedirs(sqlpath)
sqlf = os.path.join(sqlpath, fname)
self.logger.debug('sqlf: {}'.format(sqlf))
sql = re.sub(r'\n{3,}', r'\n\n', sql.strip())
open(sqlf, 'w').write(sql)
def _get_file_name(self, postway, version, versioned, name):
fname = '{}.sql'.format(name)
if versioned:
version += 1
fname = 'V{}__{}.sql'.format(version, name)
elif postway:
fname = 'R__{}.sql'.format(name)
return fname, version
def parse_arguments(self):
"""
Parse command line arguments.
Sets self.args parameter for use throughout class/script.
"""
self.parser.add_argument('-i', '--infile', required=True,
help='The dump file created using pg_dump in text mode.')
self.parser.add_argument('-o', '--outdir',
help="The directory to use as the parent for the type'd directories. Created if it "
"doesn't exist. Defaults to the current directory.")
self.parser.add_argument('--include-parent-name', type=ParentName.from_string, choices=list(ParentName),
default=ParentName.none,
help='When naming files for objects that have parents (triggers, fk, pk, etc), use '
'the parent name in the file name.')
self.parser.add_argument('--postway', action='store_true',
help='Create postway-compliant migration scripts.')
self.parser.add_argument('--postway-versioned-only', action='store_true',
help='Create postway-compliant VERSIONED migration scripts ONLY. Normally, postsplit '
'creates both versioned and repeatable scripts, with '
'functions/procedures/packages etc. being repeatable and '
'tables/indexs/constraints etc. being versioned. However, EDB EPAS '
'"check_function_bodies=false" DOES NOT WORK for packages. This means that if '
'there are packages that have dependencies, they will fail to build unless they '
'just happen to sort correctly.')
self.parser.add_argument('--postway-version', default=1,
help='Start with postway-version for migration scripts.')
self.parser.add_argument('--plot', action='store_true',
help='Show plot.')
self.parser.add_argument('-V', '--version', action='version', version='%(prog)s 1.0.0',
help='Print the version number of postsplit.')
self.parser.add_argument('-v', '--verbose', action='count', help='verbose level... repeat up to three times.')
self.args = self.parser.parse_args()
return self
def set_log_level_from_verbose(self):
if not self.args.verbose:
self.console_handler.setLevel('ERROR')
elif self.args.verbose == 1:
self.console_handler.setLevel('WARNING')
elif self.args.verbose == 2:
self.console_handler.setLevel('INFO')
elif self.args.verbose >= 3:
self.console_handler.setLevel('DEBUG')
else:
self.logger.critical('UNEXPLAINED NEGATIVE COUNT!')
return self
def build_graph(self):
outdir = self.args.outdir
if outdir == '':
outdir = os.path.dirname(self.args.infile)
graph = nx.DiGraph()
type_regex = re.compile(
r'-- Name: ([-\w\s\.\$\^]+)(?:\([-\w\s\[\],.\*\"]*\))?; Type: ([-\w\s]+); Schema: ([-\w]+); Owner: ([-\w]*)(?:; Tablespace: )?([-\w]*)\n',
flags=re.IGNORECASE)
toc_regex = re.compile(r'-- TOC entry (\d*) \(class (\d+) OID (\d+)\)\n', flags=re.IGNORECASE)
dep_regex = re.compile(r'-- Dependencies: (.*)')
name_regex = re.compile(r'(\w+)\s+(\w+)')
user_mapping_name_regex = re.compile(r'USER\sMAPPING\s([\w\s]+)')
name, schema_type, schema, owner, tablespace = [''] * 5
toc_id = 0
sql = ''
type_line = ''
# build the graph based on TOC
for sec in self._get_sections():
# remove all lines with just `--`
section = [x for x in list(sec) if x != '--\n']
toc_id = toc_regex.search(section[0]).group(1)
graph.add_node(toc_id, schema='', schema_type='', name='', sql='')
if section[1].startswith('-- Dependencies'):
dep_ids = dep_regex.search(section[1]).group(1).split()
# edges defined by TOC dependencies
for dep_id in set(dep_ids):
# dep_id is a dependency of toc_id
self.logger.debug('TOC edge: {}({}) <- {}({})'.format('unknown',
dep_id,
'unknown',
toc_id))
graph.add_edge(toc_id, dep_id)
type_line = section[2]
else:
type_line = section[1]
name, schema_type, schema, owner, tablespace = type_regex.search(type_line).groups()
# ignore the schema_version table if building for postway
if self.args.postway and 'schema_version' in name:
continue
schema_type = schema_type.replace(' ', '_').lower()
if schema_type == 'user_mapping':
parent_name = ''
name = user_mapping_name_regex.search(name).group(1).replace(' ', '_').lower()
object_name = name
else:
name_match = name_regex.match(name)
if name_match:
parent_name = name_regex.search(name).group(1)
object_name = name_regex.search(name).group(2)
if self.args.include_parent_name == ParentName.none:
name = name_regex.search(name).group(2)
else:
if self.args.include_parent_name == ParentName.before:
name = '{}_{}'.format(name_regex.search(name).group(1), name_regex.search(name).group(2))
else:
name = '{}_{}'.format(name_regex.search(name).group(2), name_regex.search(name).group(1))
else:
parent_name = ''
object_name = name
name = name.lower()
if self.args.postway:
name = '{}_{}'.format(name, schema_type)
if schema == '-':
schema = 'public'
# section = anything that doesn't start with:
# 1. -- TOC
# 2. -- Dependencies
# 3. SET search_path
sql = [y for y in section if not (y.startswith('-- TOC') or
y.startswith('-- Dependencies') or
y.startswith('SET search_path'))]
self.logger.debug(
'Owner: {}; Schema: {}; Type: {}; Name: {}; Parent Name: {}; Object Name: {}'.format(owner, schema,
schema_type, name,
parent_name,
object_name))
graph.nodes[toc_id]['schema'] = schema
graph.nodes[toc_id]['schema_type'] = schema_type
graph.nodes[toc_id]['name'] = name
graph.nodes[toc_id]['parent_name'] = parent_name
graph.nodes[toc_id]['object_name'] = object_name
graph.nodes[toc_id]['owner'] = owner
graph.nodes[toc_id]['sql'] = ''.join(sql)
# prune the graph - remove all nodes that have no name attribute
prune_list = [pr for pr, d in graph.nodes(data=True) if 'name' not in d or d['name'] is None or d['name'] == '']
graph.remove_nodes_from(prune_list)
# find edges where type is a dependency of type/function/procedure/package
graph = self._add_edges_by_sql(['type'], ['type', 'function', 'procedure', 'package'], graph)
# find edges where table/view is a dependency of view
graph = self._add_edges_by_sql(['table', 'view'], ['view'], graph)
# find edges where table is a dependency of index
graph = self._add_edges_by_sql(['table'], ['index'], graph)
# find edges where table is a dependency of constraints/fk_constraints
graph = self._add_edges_by_parent_name(graph)
# merge acl into parent script
graph = self._merge_acl(graph)
# add public schema
# graph.add_node(0, schema='public', schema_type='schema', name='public_schema',
# sql=self.create_public_schema_sql)
return graph
def prepare_and_write(self, toc_id, postway, version, versioned_only, outdir, node):
self.logger.debug('{}: {}'.format(toc_id, node))
owner, schema, schema_type, name, parent_name, object_name, sql = map(node.get,
('owner', 'schema', 'schema_type', 'name',
'parent_name', 'object_name', 'sql'))
if name is not None and name > '':
self.logger.info(
'Owner: {}; Schema: {}; Type: {}; Name: {}; Parent Name: {}; Object Name: {}'.format(owner, schema,
schema_type, name,
parent_name,
object_name))
if postway and schema_type != 'schema':
# sql = self._inject_postway_separator(schema_type, object_name, sql)
set_role = True
sql = self._fixup_quoted_create(schema, schema_type, object_name, sql)
if schema_type in ['function', 'procedure', 'package', 'view', 'trigger', 'foreign_table']:
fname, version = self._get_file_name(postway, version, versioned_only, name)
if schema_type == 'trigger':
sql = '{}\n{}'.format(self._drop_trigger(schema, parent_name, object_name), sql)
elif schema_type == 'foreign_table':
sql = '{}\n{}'.format(self._drop_foreign_table(object_name), sql)
else:
sql = self._inject_create_or_replace(schema_type, sql)
else:
fname, version = self._get_file_name(postway, version, True, name)
sql = '{}\n{}\n{}'.format(self._header(owner, schema, set_role), sql, self._footer())
else:
fname, version = self._get_file_name(False, version, False, name)
self._write_file(outdir, schema, schema_type, fname, sql)
return version
if __name__ == '__main__':
postsplit = PostSplit().parse_arguments().set_log_level_from_verbose()
outdir = postsplit.args.outdir
if outdir is None or outdir == '':
outdir = os.path.dirname(postsplit.args.infile)
print('Start parsing {}.'.format(postsplit.args.infile))
postsplit.logger.debug('args: {}'.format(postsplit.args))
g = postsplit.build_graph()
# if postsplit.args.plot:
# print('Plot the graph.')
# nx.draw(g, with_labels=True, font_weight='bold')
# plt.show()
print('Sort the DAG.')
try:
s = list(reversed(list(nx.topological_sort(g))))
postsplit.logger.debug(s)
print('Write the files.')
p = postsplit.args.postway
p_v = postsplit.args.postway_version # postway version - used in version migration scripts
p_v_o = postsplit.args.postway_versioned_only
for n in s:
p_v = postsplit.prepare_and_write(n, p, p_v, p_v_o, outdir, g.nodes[n])
print('Done processing {}.'.format(postsplit.args.infile))
except Exception as err:
c = nx.find_cycle(g)
postsplit.logger.debug(c)
postsplit.logger.error(err)
exit(1)
Raw
postway.py
#!/usr/bin/env python
import argparse
import collections.abc
import logging
import ntpath
import os
import re
import subprocess
import zlib
from collections import namedtuple
from enum import Enum
from glob import glob
from timeit import default_timer as timer
import psycopg2
from packaging import version
from psycopg2._psycopg import AsIs
from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
from psycopg2.extensions import make_dsn
from psycopg2.extras import LoggingConnection
class OrderedSet(collections.abc.MutableSet):
def __init__(self, iterable=None):
self.end = end = []
end += [None, end, end] # sentinel node for doubly linked list
self.map = {} # key --> [key, prev, next]
if iterable is not None:
self |= iterable
def __len__(self):
return len(self.map)
def __contains__(self, key):
return key in self.map
def add(self, key):
if key not in self.map:
end = self.end
curr = end[1]
curr[2] = end[1] = self.map[key] = [key, curr, end]
def discard(self, key):
if key in self.map:
key, prev, next = self.map.pop(key)
prev[2] = next
next[1] = prev
def __iter__(self):
end = self.end
curr = end[2]
while curr is not end:
yield curr[0]
curr = curr[2]
def __reversed__(self):
end = self.end
curr = end[1]
while curr is not end:
yield curr[0]
curr = curr[1]
def pop(self, last=True):
if not self:
raise KeyError('set is empty')
key = self.end[1][0] if last else self.end[2][0]
self.discard(key)
return key
def __repr__(self):
if not self:
return '%s()' % (self.__class__.__name__,)
return '%s(%r)' % (self.__class__.__name__, list(self))
def __eq__(self, other):
if isinstance(other, OrderedSet):
return len(self) == len(other) and list(self) == list(other)
return set(self) == set(other)
class PostWayCommand(Enum):
baseline = 1
version = 2
clean = 3
migrate = 4
validate = 5
info = 6
def __str__(self):
return self.name
@staticmethod
def from_string(s):
try:
return PostWayCommand[s]
except KeyError:
raise ValueError()
class PostWay:
"""
PostWay is a wholly new derivative of Flyway from BoxFuse. It's a derivative in that it derives it's workflow and
the schema_version table directly from Flyway. It's wholly new in that the code and focus are specific to
EDB/Postgres, and the method for executing SQL scripts uses native EDB/Postgres tools - specifically psql.
Management routines are managed via psycopg2. Migration scripts are run through psql.
"""
SchemaRecord = namedtuple('SchemaRecord',
'name, script, baseline_version, max_current_version, max_schema_version, '
'repeatable_schema_migrations, versioned_schema_migrations, '
'diff_repeatable_migrations, diff_versioned_migrations, changed_repeatable_migrations')
MigrationRecord = namedtuple('MigrationRecord', 'script, version, description, checksum')
ChangedMigrationRecord = namedtuple('ChangedMigrationRecord',
'script, version, description, old_checksum, new_checksum')
ExecutingMigrationRecord = namedtuple('ExecutingMigrationRecord', 'schema, script, version, description, checksum')
def __init__(self):
self.parser = argparse.ArgumentParser(
add_help=False,
description='''
PostWay is a wholly new derivative of Flyway from BoxFuse. It's a derivative in that it derives it's workflow and
the schema_version table directly from Flyway. It's wholly new in that the code and focus are specific to
EDB/Postgres, and the method for executing SQL scripts uses native EDB/Postgres tools - specifically psql.
Management routines are managed via psycopg2. Migration scripts are run through psql.
''')
self.logger = logging.getLogger('postway')
self.logger.setLevel(logging.DEBUG)
self.console_handler = logging.StreamHandler() # sys.stderr
self.console_handler.setLevel(
logging.CRITICAL) # set later by set_log_level_from_verbose() in interactive sessions
self.console_handler.setFormatter(logging.Formatter('[%(levelname)s] %(message)s'))
self.logger.addHandler(self.console_handler)
self.args = None
self.conn = None
self.is_db = False
self.schema_records = None
self.is_max_schema_version = False
self.is_baselined = False
self.is_validated = False
self.versioned_prefix = 'V'
self.repeatable_prefix = 'R'
self.migration_separator = '__'
self.migration_suffix = '.sql'
self.baseline_version = 1
self.baseline_description = '<< Flyway Baseline >>'
self.max_schema_version_sql = '''
SELECT version
FROM %(nspname)s.schema_version
WHERE installed_rank = (SELECT MAX(installed_rank) FROM %(nspname)s.schema_version WHERE version IS NOT NULL);
'''
self.get_repeatable_migrations_sql = '''
SELECT script, version, description, checksum
FROM %(nspname)s.schema_version
WHERE version IS NULL
ORDER BY installed_rank ASC
'''
self.get_versioned_migrations_sql = '''
SELECT script, version, description, checksum
FROM %(nspname)s.schema_version
WHERE version IS NOT NULL AND script != %(description)s
ORDER BY installed_rank ASC
'''
self.schema_exists_sql = '''
SELECT EXISTS(
SELECT 1 FROM pg_class c
JOIN pg_namespace n ON c.relnamespace = n.oid
WHERE n.nspname = %(schema)s
AND c.relkind = 'r'
AND c.oid NOT IN (SELECT inhrelid FROM pg_inherits)
);
'''
self.schema_version_exists_sql = '''
SELECT EXISTS(
SELECT 1 FROM pg_class c
JOIN pg_namespace n ON c.relnamespace = n.oid
WHERE n.nspname = %(schema)s
AND c.relkind = 'r'
AND c.oid NOT IN (SELECT inhrelid FROM pg_inherits)
AND c.relname = 'schema_version'
);
'''
self.create_schema_version_sql = '''
--
-- Copyright 2010-2017 Boxfuse GmbH
--
-- Licensed under the Apache License, Version 2.0 (the 'License');
-- you may not use this file except in compliance with the License.
-- You may obtain a copy of the License at
--
-- http://www.apache.org/licenses/LICENSE-2.0
--
-- Unless required by applicable law or agreed to in writing, software
-- distributed under the License is distributed on an 'AS IS' BASIS,
-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-- See the License for the specific language governing permissions and
-- limitations under the License.
--
CREATE TABLE %(nspname)s.schema_version (
installed_rank SERIAL PRIMARY KEY,
version VARCHAR(50),
description VARCHAR(200) NOT NULL,
type VARCHAR(20) NOT NULL,
script VARCHAR(1000) NOT NULL,
checksum NUMBER,
installed_by VARCHAR(100) NOT NULL,
installed_on TIMESTAMP NOT NULL DEFAULT now(),
execution_time INTEGER NOT NULL,
success BOOLEAN NOT NULL
) WITH (
OIDS=FALSE
);
CREATE INDEX schema_version_s_idx ON %(nspname)s.schema_version (version, success);
'''
self.is_baselined_sql = 'SELECT EXISTS(SELECT 1 FROM %(nspname)s.schema_version WHERE script = %(script)s);'
self.baselined_version_sql = 'SELECT version FROM %(nspname)s.schema_version WHERE script = %(script)s;'
self.insert_schema_version_sql = '''
INSERT INTO %(nspname)s.schema_version (version, description, "type", script, checksum, installed_by, execution_time, success)
VALUES (%(version)s, %(description)s, %(type)s, %(script)s, %(checksum)s, %(installed_by)s, %(execution_time)s, %(success)s);
'''
self.drop_schema_sql = 'DROP SCHEMA IF EXISTS %(nspname)s CASCADE;'
self.db_no_more_connections_sql = "UPDATE pg_database SET datallowconn = 'false' WHERE datname = %(dbname)s;"
self.db_disconnect_sql = '''
SELECT pg_terminate_backend(pid)
FROM pg_stat_activity
WHERE datname = %(dbname)s;
'''
self.db_drop_sql = 'DROP DATABASE IF EXISTS %(dbname)s'
self.db_create_sql = '''
CREATE DATABASE %(dbname)s WITH OWNER=%(owner)s LC_COLLATE='C' LC_CTYPE='C' TEMPLATE='template0';
'''
@property
def _dsn(self):
return make_dsn(host=self.args.host,
port=self.args.port,
dbname=self.args.dbname,
user=self.args.user,
password=self.args.password)
def _get_search_path(self):
with self.conn:
with self.conn.cursor() as curs:
curs.execute('SHOW search_path;')
search_path = curs.fetchone()
return search_path
def _set_search_path(self, schema):
search_path = self._get_search_path()
self.logger.debug('old search_path: {}'.format(search_path[0]))
with self.conn:
with self.conn.cursor() as curs:
curs.execute('SET search_path=%(schema)s,%(search_path)s',
{'schema': schema, 'search_path': AsIs(', '.join(search_path))})
self.logger.debug('new search_path: {}'.format(self._get_search_path()[0]))
def _is_schema(self, schema):
with self.conn:
with self.conn.cursor() as curs:
curs.execute(self.schema_exists_sql, {'schema': schema})
res = curs.fetchone()
is_schema_version = res[0]
self.logger.warning('schema exists: {}'.format(is_schema_version))
return is_schema_version
def _is_schema_version(self, schema):
with self.conn:
with self.conn.cursor() as curs:
curs.execute(self.schema_version_exists_sql, {'schema': schema})
res = curs.fetchone()
is_schema_version = res[0]
self.logger.warning('schema_version exists: {}'.format(is_schema_version))
return is_schema_version
def _schema_baseline(self, schema, script):
with self.conn:
with self.conn.cursor() as curs:
curs.execute(self.baselined_version_sql, {'nspname': AsIs(schema), 'script': script})
res = curs.fetchone()
if res is not None:
baselined_version = res[0]
self.logger.warning('Schema is baselined')
return baselined_version
else:
self.logger.warning('Schema is not baselined')
return 0
def _get_repeatable_schema_migrations(self, schema):
with self.conn:
with self.conn.cursor() as curs:
curs.execute(self.get_repeatable_migrations_sql, {'nspname': AsIs(schema)})
res = curs.fetchall()
ret = list(map(self.MigrationRecord._make, res))
return ret
def _get_versioned_schema_migrations(self, schema, baseline_description):
with self.conn:
with self.conn.cursor() as curs:
curs.execute(self.get_versioned_migrations_sql,
{'nspname': AsIs(schema), 'description': baseline_description})
res = curs.fetchall()
# noinspection PyTypeChecker
ret = list(map(self.MigrationRecord._make, [(r[0], version.parse(r[1]), r[2], r[3]) for r in res]))
return ret
def _get_current_migrations(self, migration_base_directory, schema, repeatable_prefix, versioned_prefix,
migration_separator, migration_suffix):
max_current_version = 0
repeatable_migrations = []
versioned_migrations = []
repeatable_migrations = postway._get_repeatable_migrations(migration_base_directory, schema, repeatable_prefix,
migration_separator, migration_suffix)
versioned_migrations = postway._get_versioned_migrations(migration_base_directory, schema, versioned_prefix,
migration_separator, migration_suffix)
if versioned_migrations:
max_current_version = max(versioned_migrations, key=lambda mi: mi.version).version
return max_current_version, repeatable_migrations, versioned_migrations
def _get_schema_migrations(self, schema, baseline_description, baseline_version, user):
repeatable_schema_migrations = self._get_repeatable_schema_migrations(schema)
versioned_schema_migrations = self._get_versioned_schema_migrations(schema, baseline_description)
return repeatable_schema_migrations, versioned_schema_migrations
def _get_repeatable_migrations(self, migration_base_directory, schema, prefix, separator, suffix):
files = self._get_files(os.path.join(migration_base_directory, schema),
'{}*{}*{}'.format(prefix, separator, suffix))
migration_files = [self._repeatable_file_parts(schema, f, separator, prefix, suffix) for f in files]
return sorted(migration_files, key=lambda fi: self._path_leaf(fi.script))
def _get_versioned_migrations(self, migration_base_directory, schema, prefix, separator, suffix):
files = self._get_files(os.path.join(migration_base_directory, schema),
'{}*{}*{}'.format(prefix, separator, suffix))
migration_files = [self._versioned_file_parts(f, separator, prefix, suffix) for f in files]
return sorted(migration_files, key=lambda fi: fi.version)
def _repeatable_file_parts(self, schema, file_path, separator, prefix, suffix):
description, file_parts = self._migration_file_parts(file_path, separator, suffix)
checksum = self._crc(file_path)
return self.MigrationRecord(file_path, None, description, checksum)
def _versioned_file_parts(self, file_path, separator, prefix, suffix):
description, file_parts = self._migration_file_parts(file_path, separator, suffix)
ver = version.parse(file_parts[0].replace(prefix, '').replace('_', '.'))
checksum = self._crc(file_path)
return self.MigrationRecord(file_path, ver, description, checksum)
def _migration_file_parts(self, file_path, separator, suffix):
file_name = self._path_leaf(file_path)
file_parts = file_name.split(separator)
description = file_parts[1].replace(suffix, '').replace('_', ' ')
return description, file_parts
def _create_schema_version(self, schema):
if self._is_schema_version(schema):
return True
else:
with self.conn:
with self.conn.cursor() as curs:
self.logger.warning('Creating schema_version in {}'.format(schema))
curs.execute(self.create_schema_version_sql, {'nspname': AsIs(schema)})
return True
def _execute_migration(self, migration, user, bindir, host, port, dbname, password):
penv = os.environ.copy()
penv['PGPASSWORD'] = password
psql = ['{}/psql'.format(bindir), '-h', host, '-p', '{}'.format(port), '-d', dbname, '-U', user]
with self.conn:
with self.conn.cursor() as curs:
start, end, success = self._execute_psql(penv, psql, migration.script)
if not success:
exit(1)
execution_time = end - start
curs.execute(self.insert_schema_version_sql,
{'nspname': AsIs(migration.schema),
'version': None if migration.version is None else '{}'.format(migration.version),
'description': migration.description,
'type': 'SQL',
'script': migration.script,
'checksum': migration.checksum,
'installed_by': user,
'execution_time': int(round(execution_time * 1000)),
'success': True})
def _execute_psql(self, penv, psql, script):
success = True
error_regex = re.compile(r'psql\.bin:(.*):([\d]+):\sERROR:\s+(.*)')
warning_regex = re.compile(r'psql\.bin:(.*):([\d]+):\sNOTICE:\s+(.*)')
if self.args.verbose is not None and self.args.verbose >= 3:
psql.extend(['-e'])
psql.extend(['-f', script])
self.logger.warning(subprocess.list2cmdline(psql))
start = timer()
end, output = self._execute_process(psql, penv)
error_matches = error_regex.search(output)
if error_matches:
success = False
self.logger.error(
'Error found at {start}-{end}: {match}'.format(start=error_matches.start(),
end=error_matches.end(),
match=error_matches.group()))
warning_matches = warning_regex.search(output)
if warning_matches:
self.logger.warning(
'Warning found at {start}-{end}: {match}'.format(start=warning_matches.start(),
end=warning_matches.end(),
match=warning_matches.group()))
return start, end, success
def _execute_clean_db(self, user, host, port, dbname, password):
self._execute_dropdb(user, host, port, dbname, password)
self._execute_createdb(user, host, port, dbname, password)
def _execute_dropdb(self, user, host, port, dbname, password):
start = timer()
db_dsn = make_dsn(host=host,
port=port,
dbname='postgres',
user=user,
password=password)
with psycopg2.connect(db_dsn, connection_factory=LoggingConnection) as conn:
conn.initialize(self.logger)
with conn.cursor() as curs:
self.logger.warning('Setting database {} to accept no more connections.'.format(dbname))
curs.execute(self.db_no_more_connections_sql, {'dbname': dbname})
with conn.cursor() as curs:
self.logger.warning('Disconnecting database {}.'.format(dbname))
curs.execute(self.db_disconnect_sql, {'dbname': dbname})
self.logger.warning('Dropping database {}.'.format(dbname))
connect = psycopg2.connect(db_dsn, connection_factory=LoggingConnection)
connect.initialize(self.logger)
connect.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
cursor = connect.cursor()
cursor.execute(self.db_drop_sql, {'dbname': AsIs(dbname)})
end = timer()
return start, end
def _execute_createdb(self, user, host, port, dbname, password):
start = timer()
db_dsn = make_dsn(host=host,
port=port,
dbname='postgres',
user=user,
password=password)
self.logger.warning('Creating database {}.'.format(dbname))
connect = psycopg2.connect(db_dsn, connection_factory=LoggingConnection)
connect.initialize(self.logger)
connect.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
cursor = connect.cursor()
cursor.execute(self.db_create_sql, {'dbname': AsIs(dbname), 'owner': user})
end = timer()
return start, end
def _execute_process(self, process, penv):
try:
stdout = subprocess.check_output(process, stderr=subprocess.STDOUT, env=penv)
end = timer()
output = stdout.decode('utf-8')
self.logger.debug(output)
except subprocess.CalledProcessError as db_exc:
output = db_exc.output.decode('utf-8')
self.logger.error(output)
raise
return end, output
# noinspection PyTypeChecker
def _find_changed_migrations(self, expected, difference):
changed = []
for exp in expected:
for dif in difference:
if exp.script == dif.script:
changed.append(self.ChangedMigrationRecord._make(list(exp) + [dif.checksum]))
return OrderedSet(changed)
@staticmethod
def _crc(file_path):
crc = 0
with open(file_path, "rb") as f:
data = f.read()
crc = zlib.crc32(data) & 0xffffffff
return crc
@staticmethod
def _path_leaf(path):
head, tail = ntpath.split(path)
return tail or ntpath.basename(head)
@staticmethod
def _get_files(path, glob_pattern):
files = [y for x in os.walk(path) for y in
glob(os.path.join(x[0], glob_pattern))]
return files
def set_log_level_from_verbose(self):
if not self.args.verbose:
self.console_handler.setLevel('ERROR')
elif self.args.verbose == 1:
self.console_handler.setLevel('WARNING')
elif self.args.verbose == 2:
self.console_handler.setLevel('INFO')
elif self.args.verbose >= 3:
self.console_handler.setLevel('DEBUG')
else:
self.logger.critical('UNEXPLAINED NEGATIVE COUNT!')
return self
def parse_arguments(self):
"""
Parse command line arguments.
Sets self.args parameter for use throughout class/script.
"""
self.parser.add_argument('-m', '--migration-base-directory',
default=os.getcwd(),
help='The base directory in which to look for schema-named directories for migration '
'scripts.')
self.parser.add_argument('command',
default=PostWayCommand.validate,
type=PostWayCommand.from_string,
choices=list(PostWayCommand),
help='The PostWay command to execute.')
self.parser.add_argument('-h', '--host', default='localhost', help='The EPAS host.')
self.parser.add_argument('--port', default=5444, help='The EPAS port.')
self.parser.add_argument('-U', '--user', required=True, help='The EPAS username.')
self.parser.add_argument('-W', '--password', required=True, help='The EPAS password.')
self.parser.add_argument('-d', '--dbname', required=True, help='The EPAS database.')
self.parser.add_argument('-s', '--schema', dest='schemas', action='append',
help='The EPAS schema(s). If none, postway assumes a full DB run, iterating '
'over each *_schema.sql file in the "{MIGRATION_DIRECTORY}/public/schema" '
'directory. Each use appends a schema to the list.')
self.parser.add_argument('-b', '--bindir',
required=True,
help='The path to the EPAS bin directory where psql is located. Required to actually '
'run the migration scripts.')
self.parser.add_argument('-V', '--version', action='version', version='%(prog)s 1.0.0',
help='Print the version number of postway.')
self.parser.add_argument('-v', '--verbose', action='count', help='verbose level... repeat up to three times.')
self.parser.add_argument('--help', action='help', default=argparse.SUPPRESS,
help=argparse._('show this help message and exit'))
self.args = self.parser.parse_args()
return self
def get_schemas(self):
if not self.schema_records:
schemas = self.args.schemas
directory = self.args.migration_base_directory
suffix = self.migration_suffix
files = self._get_files(os.path.join(directory, 'public', 'schema'), '*_schema{}'.format(suffix))
public = self.SchemaRecord('public', None, None, None, None, None, None, None, None, None)
if not schemas:
self.schema_records = [self.SchemaRecord._make(
[self._path_leaf(f).replace('_schema.sql', ''), f, None, None, None, None, None, None, None, None])
for f in files]
self.schema_records.extend([public])
self.is_db = True
else:
self.schema_records = [self.SchemaRecord._make(
[self._path_leaf(f).replace('_schema.sql', ''), f, None, None, None, None, None, None, None, None])
for s in schemas for f in files if s in f]
if 'public' in schemas:
self.schema_records.extend([public])
return self
def connect(self):
if not self.conn:
self.conn = psycopg2.connect(self._dsn, connection_factory=LoggingConnection)
self.conn.initialize(self.logger)
return self
def do_clean(self, user, bindir, host, port, dbname, password):
if self.is_db:
self._execute_clean_db(user, host, port, dbname, password)
self.connect()
for sr in self.schema_records:
if sr.name != 'public':
with self.conn:
with self.conn.cursor() as curs:
self.logger.warning('Dropping {} schema'.format(sr.name))
curs.execute(self.drop_schema_sql, {'nspname': AsIs(sr.name)})
self.logger.warning('Creating {} schema'.format(sr.name))
penv = os.environ.copy()
penv['PGPASSWD'] = password
psql = ['{}/psql'.format(bindir), '-h', host, '-p', '{}'.format(port), '-d', dbname, '-U', user]
start, end, success = self._execute_psql(penv, psql, sr.script)
print('{} cleaned'.format(sr.name))
if not success:
exit(1)
def do_baseline(self, baseline_version, baseline_description, user):
self.connect()
if not self.is_baselined:
schema_records = []
for sr in self.schema_records:
schema_baseline_version = -1
if self._is_schema(sr.name) and self._create_schema_version(sr.name):
schema_baseline_version = self._schema_baseline(sr.name, baseline_description)
if schema_baseline_version == 0:
with self.conn:
with self.conn.cursor() as curs:
self.logger.warning(
'Inserting baseline version for : {}'.format(sr.name, baseline_version))
schema_baseline_version = baseline_version
curs.execute(self.insert_schema_version_sql,
{'nspname': AsIs(sr.name),
'version': baseline_version,
'description': baseline_description,
'type': 'SQL',
'script': baseline_description,
'checksum': 0,
'installed_by': user,
'execution_time': 0,
'success': True})
schema_records.extend(self.SchemaRecord(sr.name, sr.script, baseline_version, sr.max_current_version,
sr.max_schema_version,
sr.repeatable_schema_migrations, sr.versioned_schema_migrations,
sr.diff_repeatable_migrations, sr.diff_versioned_migrations,
sr.changed_repeatable_migrations))
print('Baseline schema_version for {}: {}'.format(sr.name, schema_baseline_version))
if schema_records:
self.schema_records = schema_records
self.is_baselined = True
def get_max_schema_version(self, baseline_version, baseline_description, user):
self.connect()
if not self.is_max_schema_version:
schema_records = []
for sr in self.schema_records:
max_schema_version = 0
self.do_baseline(baseline_version, baseline_description, user)
with self.conn:
with self.conn.cursor() as curs:
curs.execute(self.max_schema_version_sql, {'nspname': AsIs(sr.name)})
res = curs.fetchone()
max_schema_version = version.parse(res[0])
schema_records.append(self.SchemaRecord(sr.name, sr.script, sr.baseline_version, sr.max_current_version,
max_schema_version,
sr.repeatable_schema_migrations, sr.versioned_schema_migrations,
sr.diff_repeatable_migrations, sr.diff_versioned_migrations,
sr.changed_repeatable_migrations))
print('Max schema_version for {}: {}'.format(sr.name, max_schema_version))
if schema_records:
self.schema_records = schema_records
self.is_max_schema_version = True
def migrate(self, baseline_version, baseline_description, user, bindir, host, port, dbname, password,
migration_base_directory, repeatable_prefix, versioned_prefix, migration_separator,
migration_suffix):
self.connect()
print('Beginning migrations.')
self.validate(baseline_version, baseline_description, user, migration_base_directory, repeatable_prefix,
versioned_prefix, migration_separator, migration_suffix)
diff_versioned_migrations = []
diff_repeatable_migrations = []
for sr in self.schema_records:
if sr.diff_versioned_migrations:
# noinspection PyTypeChecker
diff_versioned_migrations.extend(
map(self.ExecutingMigrationRecord._make,
[(sr.name, d_v.script, d_v.version, d_v.description, d_v.checksum)
for d_v in sr.diff_versioned_migrations]))
if sr.diff_repeatable_migrations:
# noinspection PyTypeChecker
diff_repeatable_migrations.extend(
map(self.ExecutingMigrationRecord._make,
[(sr.name, d_r.script, d_r.version, d_r.description, d_r.checksum)
for d_r in sr.diff_repeatable_migrations]))
versioned_migrations = sorted(diff_versioned_migrations, key=lambda v: v.version)
repeatable_migrations = sorted(diff_repeatable_migrations, key=lambda r: r.script)
versioned = 0
repeatable = 0
try:
for m in versioned_migrations:
self._execute_migration(m, user, bindir, host, port, dbname, password)
versioned += 1
for m in repeatable_migrations:
self._execute_migration(m, user, bindir, host, port, dbname, password)
repeatable += 1
finally:
print('Executed - Versioned: {}, Repeatable: {}'.format(versioned, repeatable))
def validate(self, baseline_version, baseline_description, user, migration_base_directory, repeatable_prefix,
versioned_prefix, migration_separator, migration_suffix):
"""
Determine if the migration is valid by checking the following metadata about the current migration:
1. Max applied version: the max version in the schema_version table.
2. Max current version: the max version of the scripts.
3. The number of versioned scripts that need to be applied:
a. new versioned scripts.
b. throw an error if a versioned script has been changed (checksum is different).
4. The number of repeatable migration scripts that need to be applied:
a. new repeatable scripts.
b. old scripts that have changed (checksum is different).
5. The number of unchanged versioned scripts.
6. The number of unchanged repeatable scripts.
:param sr:
:type sr:
:param baseline_version:
:type baseline_version:
:param baseline_description:
:type baseline_description:
:param user:
:type user:
:param migration_base_directory:
:type migration_base_directory:
:param repeatable_prefix:
:type repeatable_prefix:
:param versioned_prefix:
:type versioned_prefix:
:param migration_separator:
:type migration_separator:
:param migration_suffix:
:type migration_suffix:
:return:
:rtype:
"""
self.connect()
print('Validating migrations.')
if not self.is_validated:
self.get_max_schema_version(baseline_version, baseline_description, user)
schema_records = []
for sr in self.schema_records:
max_current_version, \
repeatable_migrations, \
versioned_migrations = self._get_current_migrations(migration_base_directory, sr.name,
repeatable_prefix,
versioned_prefix, migration_separator,
migration_suffix)
repeatable_schema_migrations, \
versioned_schema_migrations = self._get_schema_migrations(sr.name, baseline_description,
baseline_version, user)
repeatable_expected = OrderedSet(repeatable_schema_migrations)
repeatable_found = OrderedSet(repeatable_migrations)
versioned_expected = OrderedSet(versioned_schema_migrations)
versioned_found = OrderedSet(versioned_migrations)
diff_repeatable_migrations = OrderedSet(
sorted(repeatable_found - repeatable_expected, key=lambda r: r.script))
diff_versioned_migrations = OrderedSet(
sorted(versioned_found - versioned_expected, key=lambda v: v.version))
changed_repeatable_migrations = self._find_changed_migrations(repeatable_expected,
diff_repeatable_migrations)
changed_versioned_migrations = self._find_changed_migrations(versioned_expected,
diff_versioned_migrations)
schema_records.append(self.SchemaRecord(sr.name, sr.script, sr.baseline_version, max_current_version,
sr.max_schema_version, repeatable_schema_migrations,
versioned_schema_migrations, diff_repeatable_migrations,
diff_versioned_migrations, changed_repeatable_migrations))
if bool(changed_versioned_migrations):
self.logger.error('Applied versioned migrations changed for {} - THIS SHOULD NEVER HAPPEN!!!')
self.logger.error(
'YOU MUST REVERT THE CHANGES TO THESE SCRIPTS, AND CREATE NEW VERSIONED MIGRATIONS.')
self.logger.error(sr.name, changed_versioned_migrations)
exit(1)
print('{} max applied: {}, current: {} version.'
.format(sr.name, sr.max_schema_version, max_current_version))
print('versioned scripts ready to run for {}: {}'
.format(sr.name, len(diff_versioned_migrations)))
print('repeatable scripts ready to run for {}: {} ({} changed)'
.format(sr.name, len(diff_repeatable_migrations), len(changed_repeatable_migrations)))
self.logger.warning('{} versioned: {}'.format(sr.name, diff_versioned_migrations))
self.logger.warning('{} repeatable: {}'.format(sr.name, diff_repeatable_migrations))
self.logger.warning('{} changed repeatable: {}'.format(sr.name, changed_repeatable_migrations))
if schema_records:
self.schema_records = schema_records
self.is_validated = True
def info(self, baseline_version, baseline_description, user, migration_base_directory, repeatable_prefix,
versioned_prefix, migration_separator, migration_suffix):
self.connect()
self.validate(baseline_version, baseline_description, user, migration_base_directory, repeatable_prefix,
versioned_prefix, migration_separator, migration_suffix)
versioned_schema_migrations_count = 0
diff_versioned_migrations_count = 0
repeatable_schema_migrations_count = 0
changed_repeatable_migrations_count = 0
diff_repeatable_migrations_count = 0
for sr in self.schema_records:
versioned_schema_migrations_count += len(sr.versioned_schema_migrations)
print('{} VERSIONED MIGRATIONS - APPLIED: {}'.format(sr.name, len(sr.versioned_schema_migrations)))
for v_s in sr.versioned_schema_migrations:
self.logger.info(v_s)
diff_versioned_migrations_count += len(sr.diff_versioned_migrations)
print('{} VERSIONED MIGRATIONS - PENDING: {}'.format(sr.name, len(sr.diff_versioned_migrations)))
for d_v in sr.diff_versioned_migrations:
self.logger.info(d_v)
repeatable_schema_migrations_count += len(sr.repeatable_schema_migrations)
print('{} REPEATABLE MIGRATIONS - APPLIED: {}'.format(sr.name, len(sr.repeatable_schema_migrations)))
for r_s in sr.repeatable_schema_migrations:
self.logger.info(r_s)
changed_repeatable_migrations_count += len(sr.changed_repeatable_migrations)
print(
'{} REPEATABLE MIGRATIONS - CHANGED/PENDING: {}'.format(sr.name, len(sr.changed_repeatable_migrations)))
for c_r in sr.changed_repeatable_migrations:
self.logger.info(c_r)
diff_repeatable_migrations_count += len(sr.diff_repeatable_migrations)
print('{} REPEATABLE MIGRATIONS - PENDING: {}'.format(sr.name, len(sr.diff_repeatable_migrations)))
for d_r in sr.diff_repeatable_migrations:
self.logger.info(d_r)
if self.is_db:
print('TOTAL VERSIONED MIGRATIONS - APPLIED: {}'.format(versioned_schema_migrations_count))
print('TOTAL VERSIONED MIGRATIONS - PENDING: {}'.format(diff_versioned_migrations_count))
print('TOTAL REPEATABLE MIGRATIONS - APPLIED: {}'.format(repeatable_schema_migrations_count))
print('TOTAL REPEATABLE MIGRATIONS - CHANGED/PENDING: {}'.format(changed_repeatable_migrations_count))
print('TOTAL REPEATABLE MIGRATIONS - PENDING: {}'.format(diff_repeatable_migrations_count))
if __name__ == '__main__':
postway = PostWay().parse_arguments().set_log_level_from_verbose().get_schemas()
postway.logger.debug('args: {}'.format(postway.args))
m_dir = postway.args.migration_base_directory
b_dir = postway.args.bindir
u = postway.args.user
pwd = postway.args.password
h = postway.args.host
p = postway.args.port
d = postway.args.dbname
v_prefix = postway.versioned_prefix
r_prefix = postway.repeatable_prefix
sep = postway.migration_separator
suf = postway.migration_suffix
bs_ver = postway.baseline_version
bs_desc = postway.baseline_description
c = postway.args.command
if c == PostWayCommand.clean:
postway.do_clean(u, b_dir, h, p, d, pwd)
elif c == PostWayCommand.baseline:
postway.do_baseline(bs_ver, bs_desc, u)
elif c == PostWayCommand.version:
postway.get_max_schema_version(bs_ver, bs_desc, u)
elif c == PostWayCommand.migrate:
postway.migrate(bs_ver, bs_desc, u, b_dir, h, p, d, pwd, m_dir, r_prefix, v_prefix, sep, suf)
elif c == PostWayCommand.validate:
postway.validate(bs_ver, bs_desc, u, m_dir, r_prefix, v_prefix, sep, suf)
elif c == PostWayCommand.info:
postway.info(bs_ver, bs_desc, u, m_dir, r_prefix, v_prefix, sep, suf)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment