[DRAFT]

Author: sinclert-canonical

src/mysql_shell/__init__.py

@@ -21,6 +21,9 @@
 if typing.TYPE_CHECKING:
     import relations.database_requires
 
+ROLE_DML = "charmed_dml"
+ROLE_READ = "charmed_read"
+
 logger = logging.getLogger(__name__)
 
 
@@ -125,17 +128,30 @@ def _get_attributes(self, additional_attributes: dict = None) -> str:
             attributes.update(additional_attributes)
         return json.dumps(attributes)
 
-    def create_application_database_and_user(self, *, username: str, database: str) -> str:
-        """Create database and user for related database_provides application."""
+    def create_application_database(self, *, database: str) -> str:
+        """Create database for related database_provides application."""
+        mysql_roles = self.get_mysql_roles("charmed_%")
+        statements = [f"CREATE DATABASE IF NOT EXISTS `{database}`"]
+        if ROLE_READ in mysql_roles:
+            statements.append(f"GRANT SELECT ON `{database}`.* TO {ROLE_READ}")
+        if ROLE_DML in mysql_roles:
+            statements.append(f"GRANT INSERT, DELETE, UPDATE ON `{database}`.* TO {ROLE_DML}")
+
+        logger.debug(f"Creating {database=}")
+        self._run_sql(statements)
+        logger.debug(f"Created {database=}")
+        return database
+
+    def create_application_user(self, *, database: str, username: str) -> str:
+        """Create database user for related database_provides application."""
         attributes = self._get_attributes()
-        logger.debug(f"Creating {database=} and {username=} with {attributes=}")
         password = utils.generate_password()
+        logger.debug(f"Creating {username=} with {attributes=}")
         self._run_sql([
-            f"CREATE DATABASE IF NOT EXISTS `{database}`",
             f"CREATE USER `{username}` IDENTIFIED BY '{password}' ATTRIBUTE '{attributes}'",
             f"GRANT ALL PRIVILEGES ON `{database}`.* TO `{username}`",
         ])
-        logger.debug(f"Created {database=} and {username=} with {attributes=}")
+        logger.debug(f"Created {username=} with {attributes=}")
         return password
 
     def add_attributes_to_mysql_router_user(
@@ -150,6 +166,23 @@ def add_attributes_to_mysql_router_user(
         self._run_sql([f"ALTER USER `{username}` ATTRIBUTE '{attributes}'"])
         logger.debug(f"Added {attributes=} to {username=}")
 
+    # TODO python3.10 min version: Use `set` instead of `typing.Set`
+    def get_mysql_roles(self, name_pattern: str) -> typing.Set[str]:
+        """Returns a set with the MySQL roles."""
+        logger.debug(f"Getting MySQL roles with {name_pattern=}")
+        output_file = self._container.path("/tmp/mysqlsh_output.json")
+        self._run_code(
+            _jinja_env.get_template("get_mysql_roles_with_pattern.py.jinja").render(
+                name_pattern=name_pattern,
+                output_filepath=output_file.relative_to_container,
+            )
+        )
+        with output_file.open("r") as file:
+            rows = json.load(file)
+        output_file.unlink()
+        logger.debug(f"MySQL roles found for {name_pattern=}: {len(rows)}")
+        return set(rows)
+
     def get_mysql_router_user_for_unit(
         self, unit_name: str
     ) -> typing.Optional[RouterUserInformation]:

src/mysql_shell/templates/get_mysql_roles_with_pattern.py.jinja

@@ -0,0 +1,12 @@
+import json
+
+result = session.run_sql(
+    "SELECT user FROM mysql.user WHERE user LIKE '{{ name_pattern }}'"
+)
+rows = result.fetch_all()
+# mysqlsh objects are weird—they quack (i.e. duck typing) like standard Python objects (e.g. list,
+# dict), but do not serialize to JSON correctly.
+# Cast to str & load from JSON str before serializing
+rows = json.loads(str(rows))
+with open("{{ output_filepath }}", "w") as file:
+    json.dump(rows, file)

src/relations/database_provides.py

@@ -116,9 +116,8 @@ def create_database_and_user(
         shell.delete_user(username, must_exist=False)
         logger.debug("Deleted user if exists before creating user")
 
-        password = shell.create_application_database_and_user(
-            username=username, database=self._database
-        )
+        ________ = shell.create_application_database(database=self._database)
+        password = shell.create_application_user(database=self._database, username=username)
 
         self._set_databag(
             username=username,

tests/unit/conftest.py

@@ -33,6 +33,7 @@ def patch(monkeypatch):
     )
     monkeypatch.setattr("workload.AuthenticatedWorkload._router_username", "")
     monkeypatch.setattr("mysql_shell.Shell._run_code", lambda *args, **kwargs: None)
+    monkeypatch.setattr("mysql_shell.Shell.get_mysql_roles", lambda *args, **kwargs: set())
     monkeypatch.setattr(
         "mysql_shell.Shell.get_mysql_router_user_for_unit", lambda *args, **kwargs: None
     )