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
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
|
"""
Drop 'group_id' and fix foreign key references on 'group_user_roles_on_resources' table
"""
import sqlite3
from yoyo import step
__depends__ = {'20230912_01_BxrhE-add-system-resource'}
def drop_group_id(conn):
"""Drop `group_id` from `group_user_roles_on_resources` table."""
conn.execute("PRAGMA foreign_keys = OFF")
conn.execute(
"""
ALTER TABLE group_user_roles_on_resources
RENAME TO group_user_roles_on_resources_bkp
""")
conn.execute(
"""
CREATE TABLE IF NOT EXISTS group_user_roles_on_resources (
user_id TEXT NOT NULL,
role_id TEXT NOT NULL,
resource_id TEXT NOT NULL,
PRIMARY KEY (user_id, role_id, resource_id),
FOREIGN KEY (user_id)
REFERENCES users(user_id)
ON UPDATE CASCADE ON DELETE RESTRICT,
FOREIGN KEY (role_id)
REFERENCES roles(role_id)
ON UPDATE CASCADE ON DELETE RESTRICT,
FOREIGN KEY (resource_id)
REFERENCES resources(resource_id)
ON UPDATE CASCADE ON DELETE RESTRICT
) WITHOUT ROWID
""")
conn.execute(
"INSERT INTO group_user_roles_on_resources "
"SELECT user_id, role_id, resource_id "
"FROM group_user_roles_on_resources_bkp")
conn.execute("DROP TABLE IF EXISTS group_user_roles_on_resources_bkp")
conn.execute("PRAGMA foreign_key_check")
conn.execute("PRAGMA foreign_keys = ON")
def restore_group_id(conn):
"""Restore `group_id` to `group_user_roles_on_resources` table."""
conn.execute("PRAGMA foreign_keys = OFF")
conn.execute(
"""
ALTER TABLE group_user_roles_on_resources
RENAME TO group_user_roles_on_resources_bkp
""")
conn.execute(
"""
CREATE TABLE IF NOT EXISTS group_user_roles_on_resources (
group_id TEXT NOT NULL,
user_id TEXT NOT NULL,
role_id TEXT NOT NULL,
resource_id TEXT NOT NULL,
PRIMARY KEY (group_id, user_id, role_id, resource_id),
FOREIGN KEY (user_id)
REFERENCES users(user_id)
ON UPDATE CASCADE ON DELETE RESTRICT,
FOREIGN KEY (group_id, role_id)
REFERENCES group_roles(group_id, role_id)
ON UPDATE CASCADE ON DELETE RESTRICT,
FOREIGN KEY (resource_id)
REFERENCES resources(resource_id)
ON UPDATE CASCADE ON DELETE RESTRICT
) WITHOUT ROWID
""")
cursor = conn.cursor()
cursor.execute(
"""
INSERT INTO group_user_roles_on_resources
SELECT
ro.group_id, gurorb.user_id, gurorb.role_id, gurorb.resource_id
FROM resource_ownership AS ro
INNER JOIN group_user_roles_on_resources_bkp AS gurorb
ON ro.resource_id=gurorb.resource_id
""")
conn.execute("DROP TABLE IF EXISTS group_user_roles_on_resources_bkp")
conn.execute("PRAGMA foreign_key_check")
conn.execute("PRAGMA foreign_keys = ON")
def link_sys_admin_user_roles(conn):
"""Link system-admins to the system resource."""
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute(
"SELECT ur.* FROM user_roles AS ur "
"INNER JOIN roles AS r ON ur.role_id=r.role_id "
"WHERE r.role_name='system-administrator'")
admins = cursor.fetchall()
cursor.execute(
"SELECT r.resource_id FROM resources AS r "
"INNER JOIN resource_categories AS rc "
"ON r.resource_category_id=rc.resource_category_id "
"WHERE rc.resource_category_key='system'")
system_resource_id = cursor.fetchone()["resource_id"]
cursor.executemany(
"INSERT INTO "
"group_user_roles_on_resources(user_id, role_id, resource_id) "
"VALUES (:user_id, :role_id, :resource_id)",
tuple({**admin, "resource_id": system_resource_id} for admin in admins))
def restore_sys_admin_user_roles(conn):
"""Restore fields into older `user_roles` table."""
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute(
"SELECT guror.user_id, guror.role_id "
"FROM group_user_roles_on_resources AS guror "
"INNER JOIN resources AS r "
"ON guror.resource_id=r.resource_id "
"INNER JOIN resource_categories AS rc "
"ON r.resource_category_id=rc.resource_category_id "
"WHERE rc.resource_category_key='system'")
user_roles = tuple(cursor.fetchall())
cursor.executemany(
"INSERT INTO user_roles(user_id, role_id) "
"VALUES (:user_id, :role_id)",
user_roles)
def link_group_leader_user_roles(conn):
"""Link group leaders to their resources."""
conn.execute(
"""
INSERT INTO group_user_roles_on_resources(user_id, role_id, resource_id)
SELECT gu.user_id, r.role_id, gr.resource_id
FROM group_resources AS gr INNER JOIN group_users AS gu
ON gr.group_id=gu.group_id INNER JOIN user_roles AS ur
ON gu.user_id=ur.user_id INNER JOIN roles AS r
ON ur.role_id=r.role_id
WHERE r.role_name='group-leader'
""")
def restore_group_leader_user_roles(conn):
"""Restore group admins to older `user_roles` table."""
conn.execute(
"""
INSERT INTO user_roles(user_id, role_id)
SELECT guror.user_id, guror.role_id
FROM group_user_roles_on_resources AS guror
INNER JOIN resources AS r ON guror.resource_id=r.resource_id
INNER JOIN resource_categories AS rc
ON r.resource_category_id=rc.resource_category_id
WHERE rc.resource_category_key='group'
""")
def link_group_creator_user_roles(conn):
"""Link group-creators to system."""
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute(
"SELECT ur.* FROM user_roles AS ur "
"INNER JOIN roles AS r ON ur.role_id=r.role_id "
"WHERE r.role_name='group_creator'")
creators = cursor.fetchall()
cursor.execute(
"SELECT r.resource_id FROM resources AS r "
"INNER JOIN resource_categories AS rc "
"ON r.resource_category_id=rc.resource_category_id "
"WHERE rc.resource_category_key='system'")
sys_res_id = cursor.fetchone()["resource_id"]
cursor.executemany(
"INSERT INTO "
"group_user_roles_on_resources(user_id, role_id, resource_id) "
"VALUES (:user_id, :role_id, :resource_id)",
tuple({**creator, "resource_id": sys_res_id} for creator in creators))
def restore_group_creator_user_roles(conn):
"Restore group-creator user roles."
conn.execute(
"""
INSERT INTO user_roles
SELECT guror.user_id, guror.role_id
FROM group_user_roles_on_resources AS guror
INNER JOIN roles AS r ON guror.role_id=r.role_id
WHERE r.role_name='group-creator'""")
def rename_table(conn):
"rename `group_user_roles_on_resources`, drop `user_roles`."
conn.execute("PRAGMA foreign_keys = OFF")
conn.execute("DROP TABLE IF EXISTS user_roles")
conn.execute(
"ALTER TABLE group_user_roles_on_resources RENAME TO user_roles")
conn.execute("PRAGMA foreign_key_check")
conn.execute("PRAGMA foreign_keys = ON")
def restore_tables(conn):
"rename to `group_user_roles_on_resources`, recreate original `user_roles`."
conn.execute("PRAGMA foreign_keys = OFF")
conn.execute(
"ALTER TABLE user_roles RENAME TO group_user_roles_on_resources")
conn.execute(
"""
CREATE TABLE user_roles(
user_id TEXT NOT NULL,
role_id TEXT NOT NULL,
PRIMARY KEY(user_id, role_id),
FOREIGN KEY(user_id) REFERENCES users(user_id)
ON UPDATE CASCADE ON DELETE RESTRICT,
FOREIGN KEY(role_id) REFERENCES roles(role_id)
ON UPDATE CASCADE ON DELETE RESTRICT
) WITHOUT ROWID
""")
conn.execute("PRAGMA foreign_key_check")
conn.execute("PRAGMA foreign_keys = ON")
steps = [
step(drop_group_id, restore_group_id),
step(link_sys_admin_user_roles, restore_sys_admin_user_roles),
step(link_group_leader_user_roles, restore_group_leader_user_roles),
step(link_group_creator_user_roles, restore_group_creator_user_roles),
step(rename_table, restore_tables)
]
|
