Start implementing databases

dev/docker/Dockerfile

@@ -1,4 +1,4 @@
-FROM ghcr.io/astral-sh/uv:0.6-python3.13-bookworm-slim
+FROM ghcr.io/astral-sh/uv:0.8-python3.13-bookworm-slim
 
 WORKDIR /director5
 

manager/director/apps/sites/models.py

@@ -204,6 +204,14 @@ class DatabaseHost(models.Model):
     def __str__(self):
         return f"{self.dbms}://{self.hostname}:{self.port}"
 
+    def serialize_for_appserver(self) -> dict[str, str]:
+        return {
+            "admin_hostname": self.admin_hostname or self.hostname,
+            "admin_port": self.admin_port or self.port,
+            "admin_username": self.admin_username,
+            "admin_password": self.admin_password,
+        }
+
 
 class Database(models.Model):
     """A database for a specific site."""
@@ -224,12 +232,20 @@ def username(self) -> str:
     def redacted_db_url(self) -> str:
         return f"{self.host.dbms}://{self.username}:***@{self.host.hostname}:{self.host.port}/{self.username}"
 
-    def serialize_for_appserver(self) -> dict[str, str]:
+    @property
+    def db_url(self) -> str:
+        return f"{self.host.dbms}://{self.username}:{self.password}@{self.host.hostname}:{self.host.port}/{self.username}"
+
+    def serialize_for_appserver(self) -> dict[str, Any]:
         return {
-            "url": self.redacted_db_url,
-            "name": self.site.name,
+            "host": self.host.serialize_for_appserver(),
             "username": self.username,
             "password": self.password,
+            "db_type": self.host.dbms,
+            "db_host": self.host.hostname,
+            "db_port": self.host.port,
+            "db_name": self.username,
+            "db_url": self.db_url,
         }
 
 

orchestrator/orchestrator/api/database/sql.py

@@ -0,0 +1,233 @@
+from abc import ABC, abstractmethod
+from collections.abc import Iterator
+from contextlib import contextmanager
+from typing import TYPE_CHECKING, Literal, override
+
+import mysql.connector
+import psycopg
+from psycopg import sql as psql
+
+from ..docker.schema import DatabaseInfo
+
+if TYPE_CHECKING:
+    from mysql.connector.abstracts import MySQLCursorAbstract
+
+
+class DatabaseError(Exception):
+    def __init__(self, ty: Literal["postgres", "mysql"], ex: Exception):
+        super().__init__()
+        self.ty = ty
+        self.ex = ex
+
+
+class DatabaseHandler(ABC):
+    def __init__(self, db_info: DatabaseInfo):
+        self.db_info = db_info
+
+    @abstractmethod
+    def create_database(self) -> None:
+        """Create a database using the provided database information."""
+        raise NotImplementedError("This method should be implemented by subclasses.")
+
+    @abstractmethod
+    def delete_database(self) -> None:
+        """Delete a database using the provided database information."""
+        raise NotImplementedError("This method should be implemented by subclasses.")
+
+    @abstractmethod
+    def run_query(self, db: str, query: str) -> str:
+        """Run a query on the database using the provided database information."""
+        raise NotImplementedError("This method should be implemented by subclasses.")
+
+
+class MySqlHandler(DatabaseHandler):
+    @contextmanager
+    def open_cursor(self, db: str, *, admin: bool = False) -> Iterator[MySQLCursorAbstract]:
+        kwargs = {}
+        hostname = self.db_info.host.admin_hostname
+
+        if hostname.startswith("/"):
+            kwargs["unix_socket"] = hostname
+            hostname = "localhost"
+
+        if admin:
+            port = self.db_info.host.admin_port
+            user = self.db_info.host.admin_username
+            password = self.db_info.host.admin_password
+        else:
+            port = self.db_info.db_port
+            user = self.db_info.username
+            password = self.db_info.password
+
+        conn = mysql.connector.connect(
+            host=hostname,
+            port=port,
+            user=user,
+            passwd=password,
+            database=db,
+            **kwargs,
+        )
+        try:
+            yield conn.cursor()
+
+            conn.commit()
+        finally:
+            conn.close()
+
+    @staticmethod
+    def clean_identifier(identifier: str) -> str:
+        return "".join(c for c in identifier if c.isalnum() or c == "_")
+
+    @override
+    def create_database(self) -> None:
+        """Create a database using the provided database information."""
+        with self.open_cursor("mysql", admin=True) as cursor:
+            cursor.execute(
+                "SELECT 1 FROM mysql.user WHERE user = %s;",
+                (self.clean_identifier(self.db_info.username),),
+            )
+
+            if cursor.rowcount == 0:
+                cursor.execute(
+                    f"CREATE USER '{self.clean_identifier(self.db_info.username)}'@'%%' IDENTIFIED BY %s;",
+                    (self.db_info.password,),
+                )
+            else:
+                cursor.execute(
+                    f"SET PASSWORD FOR {self.clean_identifier(self.db_info.username)}@'%%' = PASSWORD(%s);",
+                    (self.db_info.password,),
+                )
+
+            cursor.execute(
+                f"CREATE DATABASE IF NOT EXISTS {self.clean_identifier(self.db_info.db_name)}"
+            )
+            cursor.execute(
+                f"GRANT ALL ON {self.clean_identifier(self.db_info.db_name)} . * TO {self.clean_identifier(self.db_info.username)};"
+            )
+
+            cursor.execute("FLUSH PRIVILEGES;")
+
+    @override
+    def delete_database(self) -> None:
+        """Delete a database using the provided database information."""
+        with self.open_cursor("mysql", admin=True) as cursor:
+            cursor.execute(
+                f"DROP DATABASE IF EXISTS {self.clean_identifier(self.db_info.db_name)};"
+            )
+
+            cursor.execute(
+                f"DROP USER IF EXISTS {self.clean_identifier(self.db_info.username)}@'%%';"
+            )
+
+    @override
+    def run_query(self, db: str, query: str) -> str:
+        """Run a query on the database using the provided database information."""
+        with self.open_cursor(db) as cursor:
+            try:
+                _ = cursor.execute(query)
+            except mysql.connector.Error as e:
+                raise DatabaseError("mysql", e) from e
+            result = (
+                "\t".join([desc[0] for desc in cursor.description]) if cursor.description else ""
+            )
+            for row in cursor:
+                result += f"\n{'\t'.join(str(x) for x in row)}"
+            return result
+
+
+class PostgresHandler(DatabaseHandler):
+    @contextmanager
+    def open_cursor(self, db: str, *, admin: bool = False):
+        hostname = self.db_info.host.admin_hostname
+        if admin:
+            port = self.db_info.host.admin_port
+            user = self.db_info.host.admin_username
+            password = self.db_info.host.admin_password
+        else:
+            port = self.db_info.db_port
+            user = self.db_info.username
+            password = self.db_info.password
+
+        with psycopg.connect(
+            host=hostname,
+            port=port,
+            user=user,
+            password=password,
+            dbname=db,
+        ) as conn:
+            with conn.cursor() as cursor:
+                yield cursor
+
+    @override
+    def create_database(self) -> None:
+        with self.open_cursor("postgres", admin=True) as cursor:
+            # Check if the user exists
+            _ = cursor.execute(
+                "SELECT 1 FROM pg_catalog.pg_user WHERE usename = %s",
+                (self.db_info.username,),
+            )
+            query = (
+                "CREATE USER {} WITH PASSWORD %s"
+                if cursor.rowcount == 0
+                else "ALTER USER {} WITH PASSWORD %s"
+            )
+            _ = cursor.execute(
+                psql.SQL(query).format(psql.Identifier(self.db_info.username)),
+                (self.db_info.password,),
+            )
+            _ = cursor.execute(
+                "SELECT 1 FROM pg_database WHERE datname = %s",
+                (self.db_info.db_name,),
+            )
+
+            # Create database and set privileges for the user
+            if cursor.rowcount == 0:
+                _ = cursor.execute(
+                    psql.SQL("CREATE DATABASE {} WITH OWNER = %s ENCODING 'UTF8'").format(
+                        psql.Identifier(self.db_info.db_name),
+                    ),
+                    (self.db_info.host.admin_username,),
+                )
+
+            _ = cursor.execute(
+                psql.SQL("GRANT ALL PRIVILEGES ON DATABASE {} TO {}").format(
+                    psql.Identifier(self.db_info.db_name),
+                    psql.Identifier(self.db_info.username),
+                )
+            )
+
+            _ = cursor.execute(
+                psql.SQL("GRANT ALL ON SCHEMA public TO {}").format(
+                    psql.Identifier(self.db_info.username),
+                )
+            )
+
+    @override
+    def delete_database(self) -> None:
+        with self.open_cursor("postgres", admin=True) as cursor:
+            _ = cursor.execute(
+                psql.SQL("DROP DATABASE IF EXISTS {}").format(psql.Identifier(self.db_info.db_name))
+            )
+
+            _ = cursor.execute(
+                psql.SQL("REVOKE ALL ON SCHEMA public FROM {}").format(
+                    psql.Identifier(self.db_info.username)
+                )
+            )
+            _ = cursor.execute(
+                psql.SQL("DROP USER IF EXISTS {}").format(psql.Identifier(self.db_info.username))
+            )
+
+    @override
+    def run_query(self, db: str, query: str) -> str:
+        with self.open_cursor(db) as cursor:
+            try:
+                _ = cursor.execute(query)
+            except psycopg.DatabaseError as e:
+                raise DatabaseError("postgres", e) from e
+            result = (
+                "\t".join([desc[0] for desc in cursor.description]) if cursor.description else ""
+            )
+            for row in cursor:
+                result += f"\n{'\t'.join(str(x) for x in row)}"
+            return result

orchestrator/orchestrator/api/docker/schema.py

@@ -1,15 +1,12 @@
 from pathlib import Path
-from typing import Annotated, Any, Literal, TypedDict, cast
+from typing import Annotated, Any, Literal, TypedDict
 
 from pydantic import (
     BaseModel,
     Field,
-    MySQLDsn,
-    PostgresDsn,
     UrlConstraints,
     ValidationInfo,
     ValidatorFunctionWrapHandler,
-    field_validator,
 )
 from pydantic.functional_validators import AfterValidator, WrapValidator
 
@@ -59,43 +56,27 @@ class ResourceLimits(BaseModel):
 _db_url_validator = UrlConstraints(host_required=True, default_port=5432)
 
 
+class DatabaseHost(BaseModel):
+    admin_hostname: str
+    admin_username: str
+    admin_password: str
+    admin_port: int
+
+
 class DatabaseInfo(BaseModel):
-    url: Annotated[PostgresDsn, _db_url_validator] | Annotated[MySQLDsn, _db_url_validator]
-    name: str
+    host: DatabaseHost
 
     username: str
     password: str
 
-    @property
-    def type_(self) -> Literal["postgres", "mysql"]:
-        return cast(Literal["postgres", "mysql"], self.url.scheme)
-
-    @property
-    def port(self) -> int:
-        if isinstance(self.url, PostgresDsn):
-            port = self.url.hosts()[0]["port"]
-        else:
-            port = self.url.port
-        assert port is not None
-        return port
-
-    @property
-    def host(self) -> str:
-        host = self.url.host if isinstance(self.url, MySQLDsn) else self.url.hosts()[0]["host"]
-        # UrlConstraints(host_required=True) was used
-        assert host is not None
-        return host
-
-    @field_validator("url", mode="after")
-    @classmethod
-    def check_db_url(cls, v: PostgresDsn | MySQLDsn) -> PostgresDsn | MySQLDsn:
-        if isinstance(v, PostgresDsn):
-            assert len(v.hosts()) == 1, "Only one host is allowed"
-            assert v.hosts()[0]["port"] is not None, "Port is required"
-        return v
+    db_type: Literal["postgres", "mysql"]
+    db_host: str
+    db_port: int
+    db_name: str
+    db_url: str
 
     def __str__(self) -> str:
-        return f"{type(self).__name__}({self.url}, {self.name})"
+        return f"{type(self).__name__}({self.db_url}, {self.db_name})"
 
 
 # We're not too strict, the Manager should have a more
@@ -119,12 +100,12 @@ def container_env(self) -> dict[str, Any]:
         }
         if self.db is not None:
             env |= {
-                "DATABASE_URL": str(self.db),
-                "DIRECTOR_DATABASE_URL": str(self.db),
-                "DIRECTOR_DATABASE_TYPE": self.db.type_,
-                "DIRECTOR_DATABASE_HOST": self.db.host,
-                "DIRECTOR_DATABASE_PORT": self.db.port,
-                "DIRECTOR_DATABASE_NAME": self.db.name,
+                "DATABASE_URL": self.db.db_url,
+                "DIRECTOR_DATABASE_URL": self.db.db_url,
+                "DIRECTOR_DATABASE_TYPE": self.db.db_type,
+                "DIRECTOR_DATABASE_HOST": self.db.db_host,
+                "DIRECTOR_DATABASE_PORT": self.db.db_port,
+                "DIRECTOR_DATABASE_NAME": self.db.db_name,
                 "DIRECTOR_DATABASE_USERNAME": self.db.username,
                 "DIRECTOR_DATABASE_PASSWORD": self.db.password,
             }

orchestrator/pyproject.toml

@@ -6,6 +6,7 @@ dependencies = [
   "docker",
   "fastapi[standard]",
   "jinja2",
+  "mysql-connector-python",
   "pydantic",
   "pydantic-extra-types",
   "requests",

pyproject.toml

@@ -17,6 +17,7 @@ dependencies = [
   "fastapi[standard]>=0.115.0",
   "heroicons[django]>=2.8.0",
   "jinja2>=3.1.4",
+  "mysql-connector-python>=9.3.0",
   "pillow>=11.0.0",
   "psycopg[binary]>3.1.8",
   "pydantic>=2.9.2",