forked from quay/quay
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmigrationutil.py
More file actions
79 lines (61 loc) · 2.29 KB
/
Copy pathmigrationutil.py
File metadata and controls
79 lines (61 loc) · 2.29 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
import os
from abc import ABCMeta, abstractmethod, abstractproperty
from collections import namedtuple
from six import add_metaclass
MigrationPhase = namedtuple("MigrationPhase", ["name", "alembic_revision", "flags"])
@add_metaclass(ABCMeta)
class DataMigration(object):
@abstractproperty
def alembic_migration_revision(self):
"""
Returns the alembic migration revision corresponding to the currently configured phase.
"""
@abstractmethod
def has_flag(self, flag):
"""
Returns true if the data migration's current phase has the given flag set.
"""
class NullDataMigration(DataMigration):
@property
def alembic_migration_revision(self):
return "head"
def has_flag(self, flag):
raise NotImplementedError()
class DefinedDataMigration(DataMigration):
def __init__(self, name, env_var, phases):
assert phases
self.name = name
self.phases = {phase.name: phase for phase in phases}
# Add a synthetic phase for new installations that skips the entire migration.
self.phases["new-installation"] = phases[-1]._replace(
name="new-installation", alembic_revision="head"
)
phase_name = os.getenv(env_var)
if phase_name is None:
msg = "Missing env var `%s` for data migration `%s`. %s" % (
env_var,
self.name,
self._error_suffix,
)
raise Exception(msg)
current_phase = self.phases.get(phase_name)
if current_phase is None:
msg = "Unknown phase `%s` for data migration `%s`. %s" % (
phase_name,
self.name,
self._error_suffix,
)
raise Exception(msg)
self.current_phase = current_phase
@property
def _error_suffix(self):
message = "Available values for this migration: %s. " % (list(self.phases.keys()))
message += "If this is a new installation, please use `new-installation`."
return message
@property
def alembic_migration_revision(self):
assert self.current_phase
return self.current_phase.alembic_revision
def has_flag(self, flag):
assert self.current_phase
return flag in self.current_phase.flags