@@ -27,16 +27,16 @@ def __init__(self, name, node, tables=None, dbname=None, username=None):
27
27
# create publication in database
28
28
t = "table " + ", " .join (tables ) if tables else "all tables"
29
29
query = "create publication {} for {}"
30
- node .safe_psql (query .format (name , t ),
31
- dbname = dbname ,
32
- username = username )
30
+ node .safe_psql (query .format (name , t ), dbname = dbname , username = username )
33
31
34
32
def drop (self , dbname = None , username = None ):
35
33
"""
36
34
Drop publication
37
35
"""
38
- self .node .safe_psql ("drop publication {}" .format (self .name ),
39
- dbname = dbname , username = username )
36
+ self .node .safe_psql (
37
+ "drop publication {}" .format (self .name ),
38
+ dbname = dbname ,
39
+ username = username )
40
40
41
41
def add_tables (self , tables , dbname = None , username = None ):
42
42
"""
@@ -49,9 +49,10 @@ def add_tables(self, tables, dbname=None, username=None):
49
49
raise ValueError ("Tables list is empty" )
50
50
51
51
query = "alter publication {} add table {}"
52
- self .node .safe_psql (query .format (self .name , ", " .join (tables )),
53
- dbname = dbname or self .dbname ,
54
- username = username or self .username )
52
+ self .node .safe_psql (
53
+ query .format (self .name , ", " .join (tables )),
54
+ dbname = dbname or self .dbname ,
55
+ username = username or self .username )
55
56
56
57
57
58
class Subscription (object ):
@@ -87,8 +88,8 @@ def __init__(self,
87
88
}
88
89
89
90
query = (
90
- "create subscription {} connection '{}' publication {}"
91
- ). format ( name , options_string (** conninfo ), self .pub .name )
91
+ "create subscription {} connection '{}' publication {}" ). format (
92
+ name , options_string (** conninfo ), self .pub .name )
92
93
93
94
# additional parameters
94
95
if kwargs :
@@ -101,34 +102,33 @@ def disable(self, dbname=None, username=None):
101
102
Disables the running subscription.
102
103
"""
103
104
query = "alter subscription {} disable"
104
- self .node .safe_psql (query .format (self .name ),
105
- dbname = None ,
106
- username = None )
105
+ self .node .safe_psql (query .format (self .name ), dbname = None , username = None )
107
106
108
107
def enable (self , dbname = None , username = None ):
109
108
"""
110
109
Enables the previously disabled subscription.
111
110
"""
112
111
query = "alter subscription {} enable"
113
- self .node .safe_psql (query .format (self .name ),
114
- dbname = None ,
115
- username = None )
112
+ self .node .safe_psql (query .format (self .name ), dbname = None , username = None )
116
113
117
114
def refresh (self , copy_data = True , dbname = None , username = None ):
118
115
"""
119
116
Disables the running subscription.
120
117
"""
121
118
query = "alter subscription {} refresh publication with (copy_data={})"
122
- self .node .safe_psql (query .format (self .name , copy_data ),
123
- dbname = dbname ,
124
- username = username )
119
+ self .node .safe_psql (
120
+ query .format (self .name , copy_data ),
121
+ dbname = dbname ,
122
+ username = username )
125
123
126
124
def drop (self , dbname = None , username = None ):
127
125
"""
128
126
Drops subscription
129
127
"""
130
- self .node .safe_psql ("drop subscription {}" .format (self .name ),
131
- dbname = dbname , username = username )
128
+ self .node .safe_psql (
129
+ "drop subscription {}" .format (self .name ),
130
+ dbname = dbname ,
131
+ username = username )
132
132
133
133
def catchup (self , username = None ):
134
134
"""
@@ -139,8 +139,8 @@ def catchup(self, username=None):
139
139
"""
140
140
query = (
141
141
"select pg_current_wal_lsn() - replay_lsn = 0 "
142
- "from pg_stat_replication where application_name = '{}'"
143
- ). format ( self .name )
142
+ "from pg_stat_replication where application_name = '{}'" ). format (
143
+ self .name )
144
144
145
145
try :
146
146
# wait until this LSN reaches subscriber
0 commit comments