|
21 | 21 | if typing.TYPE_CHECKING:
|
22 | 22 | import relations.database_requires
|
23 | 23 |
|
24 |
| -ROLE_DML = "charmed_dml" |
25 |
| -ROLE_READ = "charmed_read" |
| 24 | +_ROLE_DML = "charmed_dml" |
| 25 | +_ROLE_READ = "charmed_read" |
26 | 26 |
|
27 | 27 | logger = logging.getLogger(__name__)
|
28 | 28 |
|
@@ -128,17 +128,34 @@ def _get_attributes(self, additional_attributes: dict = None) -> str:
|
128 | 128 | attributes.update(additional_attributes)
|
129 | 129 | return json.dumps(attributes)
|
130 | 130 |
|
| 131 | + # TODO python3.10 min version: Use `set` instead of `typing.Set` |
| 132 | + def _get_mysql_roles(self, name_pattern: str) -> typing.Set[str]: |
| 133 | + """Returns a set with the MySQL roles.""" |
| 134 | + logger.debug(f"Getting MySQL roles with {name_pattern=}") |
| 135 | + output_file = self._container.path("/tmp/mysqlsh_output.json") |
| 136 | + self._run_code( |
| 137 | + _jinja_env.get_template("get_mysql_roles_with_pattern.py.jinja").render( |
| 138 | + name_pattern=name_pattern, |
| 139 | + output_filepath=output_file.relative_to_container, |
| 140 | + ) |
| 141 | + ) |
| 142 | + with output_file.open("r") as file: |
| 143 | + rows = json.load(file) |
| 144 | + output_file.unlink() |
| 145 | + logger.debug(f"MySQL roles found for {name_pattern=}: {len(rows)}") |
| 146 | + return set(rows) |
| 147 | + |
131 | 148 | def create_application_database(self, *, database: str) -> str:
|
132 | 149 | """Create database for related database_provides application."""
|
133 |
| - mysql_roles = self.get_mysql_roles("charmed_%") |
| 150 | + mysql_roles = self._get_mysql_roles("charmed_%") |
134 | 151 | statements = [f"CREATE DATABASE IF NOT EXISTS `{database}`"]
|
135 |
| - if ROLE_READ in mysql_roles: |
| 152 | + if _ROLE_READ in mysql_roles: |
136 | 153 | statements.append(
|
137 |
| - f"GRANT SELECT ON `{database}`.* TO {ROLE_READ}", |
| 154 | + f"GRANT SELECT ON `{database}`.* TO {_ROLE_READ}", |
138 | 155 | )
|
139 |
| - if ROLE_DML in mysql_roles: |
| 156 | + if _ROLE_DML in mysql_roles: |
140 | 157 | statements.append(
|
141 |
| - f"GRANT SELECT, INSERT, DELETE, UPDATE ON `{database}`.* TO {ROLE_DML}", |
| 158 | + f"GRANT SELECT, INSERT, DELETE, UPDATE ON `{database}`.* TO {_ROLE_DML}", |
142 | 159 | )
|
143 | 160 |
|
144 | 161 | logger.debug(f"Creating {database=}")
|
@@ -170,23 +187,6 @@ def add_attributes_to_mysql_router_user(
|
170 | 187 | self._run_sql([f"ALTER USER `{username}` ATTRIBUTE '{attributes}'"])
|
171 | 188 | logger.debug(f"Added {attributes=} to {username=}")
|
172 | 189 |
|
173 |
| - # TODO python3.10 min version: Use `set` instead of `typing.Set` |
174 |
| - def get_mysql_roles(self, name_pattern: str) -> typing.Set[str]: |
175 |
| - """Returns a set with the MySQL roles.""" |
176 |
| - logger.debug(f"Getting MySQL roles with {name_pattern=}") |
177 |
| - output_file = self._container.path("/tmp/mysqlsh_output.json") |
178 |
| - self._run_code( |
179 |
| - _jinja_env.get_template("get_mysql_roles_with_pattern.py.jinja").render( |
180 |
| - name_pattern=name_pattern, |
181 |
| - output_filepath=output_file.relative_to_container, |
182 |
| - ) |
183 |
| - ) |
184 |
| - with output_file.open("r") as file: |
185 |
| - rows = json.load(file) |
186 |
| - output_file.unlink() |
187 |
| - logger.debug(f"MySQL roles found for {name_pattern=}: {len(rows)}") |
188 |
| - return set(rows) |
189 |
| - |
190 | 190 | def get_mysql_router_user_for_unit(
|
191 | 191 | self, unit_name: str
|
192 | 192 | ) -> typing.Optional[RouterUserInformation]:
|
|
0 commit comments