djrobstep/migra
1
1
Fork 0
mirror of https://github.com/djrobstep/migra.git synced 2024-09-11 05:45:51 +03:00
Code Issues Projects Releases Wiki Activity

reformat with black

Browse Source
This commit is contained in:
Robert Lechte 2018-07-16 22:01:35 +10:00
parent 4d2f0512a2
commit 749d79f324
6 changed files with 83 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
migra/__init__.py
View File

@ -6,5 +6,9 @@ from .migra import Migration
from .command import do_command
__all__ = [
'Migration', 'Changes', 'Statements', 'UnsafeMigrationException', 'do_command'
"Migration",
"Changes",
"Statements",
"UnsafeMigrationException",
"do_command",
]

41
migra/changes.py
View File

@ -6,17 +6,17 @@ from functools import partial
from collections import OrderedDict as od
THINGS = [
'schemas',
'enums',
'sequences',
'constraints',
'functions',
'views',
'indexes',
'extensions',
'privileges',
"schemas",
"enums",
"sequences",
"constraints",
"functions",
"views",
"indexes",
"extensions",
"privileges",
]
PK = 'PRIMARY KEY'
PK = "PRIMARY KEY"
def statements_for_changes(
@ -91,8 +91,10 @@ def statements_for_changes(
if not after:
break
elif after == before: # this should never happen because there shouldn't be circular dependencies
raise ValueError('cannot resolve dependencies') # pragma: no cover
elif (
after == before
): # this should never happen because there shouldn't be circular dependencies
raise ValueError("cannot resolve dependencies") # pragma: no cover
return statements
@ -109,7 +111,11 @@ def get_enum_modifications(tables_from, tables_target, enums_from, enums_target)
_, _, c_modified, _ = differences(t_before.columns, v.columns)
for k, c in c_modified.items():
before = t_before.columns[k]
if c.is_enum == before.is_enum and c.dbtypestr == before.dbtypestr and c.enum != before.enum:
if (
c.is_enum == before.is_enum
and c.dbtypestr == before.dbtypestr
and c.enum != before.enum
):
pre.append(before.change_enum_to_string_statement(t))
post.append(before.change_string_to_enum_statement(t))
for e in enums_to_change.values():
@ -143,13 +149,12 @@ def get_schema_changes(tables_from, tables_target, enums_from, enums_target):
class Changes(object):
def __init__(self, i_from, i_target):
self.i_from = i_from
self.i_target = i_target
def __getattr__(self, name):
if name == 'schema':
if name == "schema":
return partial(
get_schema_changes,
self.i_from.tables,
@ -158,21 +163,21 @@ class Changes(object):
self.i_target.enums,
)
elif name == 'non_pk_constraints':
elif name == "non_pk_constraints":
a = self.i_from.constraints.items()
b = self.i_target.constraints.items()
a_od = od((k, v) for k, v in a if v.constraint_type != PK)
b_od = od((k, v) for k, v in b if v.constraint_type != PK)
return partial(statements_for_changes, a_od, b_od)
elif name == 'pk_constraints':
elif name == "pk_constraints":
a = self.i_from.constraints.items()
b = self.i_target.constraints.items()
a_od = od((k, v) for k, v in a if v.constraint_type == PK)
b_od = od((k, v) for k, v in b if v.constraint_type == PK)
return partial(statements_for_changes, a_od, b_od)
elif name == 'views_and_functions':
elif name == "views_and_functions":
av = self.i_from.views.items()
bv = self.i_target.views.items()
af = self.i_from.functions.items()

38
migra/command.py
View File

@ -11,7 +11,7 @@ from .statements import UnsafeMigrationException
@contextmanager
def arg_context(x):
if x == 'EMPTY':
if x == "EMPTY":
yield None
else:
@ -20,36 +20,36 @@ def arg_context(x):
def parse_args(args):
parser = argparse.ArgumentParser(description='Generate a database migration.')
parser = argparse.ArgumentParser(description="Generate a database migration.")
parser.add_argument(
'--unsafe',
dest='unsafe',
action='store_true',
help='Prevent migra from erroring upon generation of drop statements.',
"--unsafe",
dest="unsafe",
action="store_true",
help="Prevent migra from erroring upon generation of drop statements.",
)
parser.add_argument(
'--schema',
dest='schema',
"--schema",
dest="schema",
default=None,
help='Restrict output to statements for a particular schema',
help="Restrict output to statements for a particular schema",
)
parser.add_argument(
'--create-extensions-only',
dest='create_extensions_only',
action='store_true',
"--create-extensions-only",
dest="create_extensions_only",
action="store_true",
default=False,
help='Only output "create extension..." statements, nothing else.',
)
parser.add_argument(
'--with-privileges',
dest='with_privileges',
action='store_true',
"--with-privileges",
dest="with_privileges",
action="store_true",
default=False,
help='Also output privilege differences (ie. grant/revoke statements)',
help="Also output privilege differences (ie. grant/revoke statements)",
)
parser.add_argument('dburl_from', help='The database you want to migrate.')
parser.add_argument("dburl_from", help="The database you want to migrate.")
parser.add_argument(
'dburl_target', help='The database you want to use as the target.'
"dburl_target", help="The database you want to use as the target."
)
return parser.parse_args(args)
@ -73,7 +73,7 @@ def run(args, out=None, err=None):
print(m.sql, file=out)
except UnsafeMigrationException:
print(
'-- ERROR: destructive statements generated. Use the --unsafe flag to suppress this error.',
"-- ERROR: destructive statements generated. Use the --unsafe flag to suppress this error.",
file=err,
)
return 3

2
migra/migra.py
View File

@ -78,7 +78,7 @@ class Migration(object):
creations_only=True, dependency_ordering=True
)
if v_and_f_changes:
self.add(['set check_function_bodies = off;'])
self.add(["set check_function_bodies = off;"])
self.add(v_and_f_changes)
self.add(self.changes.sequences(drops_only=True))
self.add(self.changes.enums(drops_only=True, modifications=False))

9
migra/statements.py
View File

@ -4,11 +4,10 @@ import re
def check_for_drop(s):
return not not re.search('(drop\s+)', s, re.IGNORECASE)
return not not re.search("(drop\s+)", s, re.IGNORECASE)
class Statements(list):
def __init__(self, *args, **kwargs):
self.safe = True
super(Statements, self).__init__(*args, **kwargs)
@ -18,14 +17,14 @@ class Statements(list):
if self.safe:
self.raise_if_unsafe()
if not self:
return ''
return ""
return '\n\n'.join(self) + '\n\n'
return "\n\n".join(self) + "\n\n"
def raise_if_unsafe(self):
if any(check_for_drop(s) for s in self):
raise UnsafeMigrationException(
'unsafe/destructive change being autogenerated, refusing to carry on further'
"unsafe/destructive change being autogenerated, refusing to carry on further"
)
def __add__(self, other):

57
tests/test_migra.py
View File

@ -15,19 +15,19 @@ SQL = """select 1;
select 2;
"""
DROP = 'drop table x;'
DROP = "drop table x;"
def test_statements():
s1 = Statements(['select 1;'])
s2 = Statements(['select 2;'])
s1 = Statements(["select 1;"])
s2 = Statements(["select 2;"])
s3 = s1 + s2
assert type(s1) == type(s2) == type(s3)
s3 = s3 + Statements([DROP])
with raises(UnsafeMigrationException):
assert s3.sql == SQL
s3.safe = False
SQL_WITH_DROP = SQL + DROP + '\n\n'
SQL_WITH_DROP = SQL + DROP + "\n\n"
assert s3.sql == SQL_WITH_DROP
@ -36,48 +36,53 @@ def outs():
def test_with_fixtures():
for FIXTURE_NAME in ['dependencies']:
for FIXTURE_NAME in ["dependencies"]:
do_fixture_test(FIXTURE_NAME)
for FIXTURE_NAME in ['everything']:
for FIXTURE_NAME in ["everything"]:
do_fixture_test(FIXTURE_NAME, with_privileges=True)
for FIXTURE_NAME in ['singleschema']:
do_fixture_test(FIXTURE_NAME, schema='goodschema')
for FIXTURE_NAME in ['singleschema_ext']:
for FIXTURE_NAME in ["singleschema"]:
do_fixture_test(FIXTURE_NAME, schema="goodschema")
for FIXTURE_NAME in ["singleschema_ext"]:
do_fixture_test(FIXTURE_NAME, create_extensions_only=True)
for FIXTURE_NAME in ['privileges']:
for FIXTURE_NAME in ["privileges"]:
do_fixture_test(FIXTURE_NAME, with_privileges=True)
def do_fixture_test(fixture_name, schema=None, create_extensions_only=False, with_privileges=False):
flags = ['--unsafe']
def do_fixture_test(
fixture_name, schema=None, create_extensions_only=False, with_privileges=False
):
flags = ["--unsafe"]
if schema:
flags += ['--schema', schema]
flags += ["--schema", schema]
if create_extensions_only:
flags += ['--create-extensions-only']
flags += ["--create-extensions-only"]
if with_privileges:
flags += ['--with-privileges']
fixture_path = 'tests/FIXTURES/{}/'.format(fixture_name)
EXPECTED = io.open(fixture_path + 'expected.sql').read().strip()
flags += ["--with-privileges"]
fixture_path = "tests/FIXTURES/{}/".format(fixture_name)
EXPECTED = io.open(fixture_path + "expected.sql").read().strip()
with temporary_database() as d0, temporary_database() as d1:
with S(d0) as s0, S(d1) as s1:
load_sql_from_file(s0, fixture_path + 'a.sql')
load_sql_from_file(s1, fixture_path + 'b.sql')
load_sql_from_file(s0, fixture_path + "a.sql")
load_sql_from_file(s1, fixture_path + "b.sql")
args = parse_args([d0, d1])
assert not args.unsafe
assert args.schema is None
out, err = outs()
assert run(args, out=out, err=err) == 3
assert out.getvalue() == ''
assert err.getvalue() == '-- ERROR: destructive statements generated. Use the --unsafe flag to suppress this error.\n'
assert out.getvalue() == ""
assert (
err.getvalue()
== "-- ERROR: destructive statements generated. Use the --unsafe flag to suppress this error.\n"
)
args = parse_args(flags + [d0, d1])
assert args.unsafe
assert args.schema == schema
out, err = outs()
assert run(args, out=out, err=err) == 2
assert err.getvalue() == ''
assert err.getvalue() == ""
assert out.getvalue().strip() == EXPECTED
ADDITIONS = io.open(fixture_path + 'additions.sql').read().strip()
EXPECTED2 = io.open(fixture_path + 'expected2.sql').read().strip()
ADDITIONS = io.open(fixture_path + "additions.sql").read().strip()
EXPECTED2 = io.open(fixture_path + "expected2.sql").read().strip()
if ADDITIONS:
with S(d0) as s0, S(d1) as s1:
m = Migration(s0, s1)
@ -95,7 +100,7 @@ def do_fixture_test(fixture_name, schema=None, create_extensions_only=False, wit
m.add_all_changes(privileges=with_privileges)
assert m.changes.i_from == m.changes.i_target
assert not m.statements # no further statements to apply
assert m.sql == ''
assert m.sql == ""
out, err = outs()
assert run(args, out=out, err=err) == 0
# test alternative parameters
@ -108,6 +113,6 @@ def do_fixture_test(fixture_name, schema=None, create_extensions_only=False, wit
m.s_from
with raises(AttributeError):
m.s_target
args = parse_args(flags + ['EMPTY', 'EMPTY'])
args = parse_args(flags + ["EMPTY", "EMPTY"])
out, err = outs()
assert run(args, out=out, err=err) == 0