refactor: add custom ULID type for sqlalchemy

Author: Panaetius

components/renku_data_services/project/db.py

@@ -171,7 +171,7 @@ async def insert_project(
             creation_date=datetime.now(UTC).replace(microsecond=0),
             keywords=project.keywords,
         )
-        project_slug = schemas.ProjectSlug(slug, project_id=project_orm.id, namespace_id=ns.id)
+        project_slug = schemas.ProjectSlug(slug, project_id=str(project_orm.id), namespace_id=ns.id)
 
         session.add(project_slug)
         session.add(project_orm)

components/renku_data_services/project/orm.py

@@ -13,6 +13,7 @@
 from renku_data_services.namespace.orm import NamespaceORM
 from renku_data_services.project import models
 from renku_data_services.project.apispec import Visibility
+from renku_data_services.utils.sqlalchemy import ULIDType
 
 metadata_obj = MetaData(schema="projects")  # Has to match alembic ini section name
 
@@ -27,7 +28,7 @@ class ProjectORM(BaseORM):
     """A Renku native project."""
 
     __tablename__ = "projects"
-    id: Mapped[str] = mapped_column("id", String(26), primary_key=True, default_factory=lambda: str(ULID()), init=False)
+    id: Mapped[ULID] = mapped_column("id", ULIDType, primary_key=True, default_factory=lambda: str(ULID()), init=False)
     name: Mapped[str] = mapped_column("name", String(99))
     visibility: Mapped[Visibility]
     created_by_id: Mapped[str] = mapped_column("created_by_id", String())

components/renku_data_services/utils/py.typed

@@ -0,0 +1,25 @@
+"""Utilities for SQLAlchemy."""
+
+from typing import cast
+
+from sqlalchemy import Dialect, types
+from ulid import ULID
+
+
+class ULIDType(types.TypeDecorator):
+    """Wrapper type for ULID <--> str conversion."""
+
+    impl = types.String
+    cache_ok = True
+
+    def process_bind_param(self, value: ULID | None, dialect: Dialect) -> str | None:
+        """Transform value for storing in the database."""
+        if value is None:
+            return None
+        return str(value)
+
+    def process_result_value(self, value: str | None, dialect: Dialect) -> ULID | None:
+        """Transform string from database into ULID."""
+        if value is None:
+            return None
+        return cast(ULID, ULID.from_str(value))