Add tag/untag endpoints for tasks, flows, and runs

src/core/tagging.py

@@ -0,0 +1,51 @@
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+from sqlalchemy import Row
+from sqlalchemy.ext.asyncio import AsyncConnection
+
+from core.errors import TagAlreadyExistsError, TagNotFoundError, TagNotOwnedError
+from database.users import User, UserGroup
+
+
+async def tag_entity(
+    entity_id: int,
+    tag: str,
+    user: User,
+    expdb: AsyncConnection,
+    *,
+    get_tags_fn: Callable[[int, AsyncConnection], Awaitable[list[str]]],
+    tag_fn: Callable[..., Awaitable[None]],
+    response_key: str,
+) -> dict[str, dict[str, Any]]:
+    tags = await get_tags_fn(entity_id, expdb)
+    if tag.casefold() in (t.casefold() for t in tags):
+        msg = f"Entity {entity_id} already tagged with {tag!r}."
+        raise TagAlreadyExistsError(msg)
+    await tag_fn(entity_id, tag, user_id=user.user_id, expdb=expdb)
+    tags = await get_tags_fn(entity_id, expdb)
+    return {response_key: {"id": str(entity_id), "tag": tags}}
+
+
+async def untag_entity(
+    entity_id: int,
+    tag: str,
+    user: User,
+    expdb: AsyncConnection,
+    *,
+    get_tag_fn: Callable[[int, str, AsyncConnection], Awaitable[Row | None]],
+    delete_tag_fn: Callable[[int, str, AsyncConnection], Awaitable[None]],
+    get_tags_fn: Callable[[int, AsyncConnection], Awaitable[list[str]]],
+    response_key: str,
+) -> dict[str, dict[str, Any]]:
+    existing = await get_tag_fn(entity_id, tag, expdb)
+    if existing is None:
+        msg = f"Tag {tag!r} not found on entity {entity_id}."
+        raise TagNotFoundError(msg)
+    groups = await user.get_groups()
+    if existing.uploader != user.user_id and UserGroup.ADMIN not in groups:
+        msg = f"Tag {tag!r} on entity {entity_id} is not owned by you."
+        raise TagNotOwnedError(msg)
+    await delete_tag_fn(entity_id, tag, expdb)
+    tags = await get_tags_fn(entity_id, expdb)
+    return {response_key: {"id": str(entity_id), "tag": tags}}

src/database/flows.py

@@ -4,6 +4,11 @@
 from sqlalchemy import Row, text
 from sqlalchemy.ext.asyncio import AsyncConnection
 
+from database.tagging import insert_tag, remove_tag, select_tag, select_tags
+
+_TABLE = "implementation_tag"
+_ID_COLUMN = "id"
+
 
 async def get_subflows(for_flow: int, expdb: AsyncConnection) -> Sequence[Row]:
     rows = await expdb.execute(
@@ -23,18 +28,7 @@ async def get_subflows(for_flow: int, expdb: AsyncConnection) -> Sequence[Row]:
 
 
 async def get_tags(flow_id: int, expdb: AsyncConnection) -> list[str]:
-    rows = await expdb.execute(
-        text(
-            """
-            SELECT tag
-            FROM implementation_tag
-            WHERE id = :flow_id
-            """,
-        ),
-        parameters={"flow_id": flow_id},
-    )
-    tag_rows = rows.all()
-    return [tag.tag for tag in tag_rows]
+    return await select_tags(table=_TABLE, id_column=_ID_COLUMN, id_=flow_id, expdb=expdb)
 
 
 async def get_parameters(flow_id: int, expdb: AsyncConnection) -> Sequence[Row]:
@@ -54,6 +48,25 @@ async def get_parameters(flow_id: int, expdb: AsyncConnection) -> Sequence[Row]:
     )
 
 
+async def tag(id_: int, tag_: str, *, user_id: int, expdb: AsyncConnection) -> None:
+    await insert_tag(
+        table=_TABLE,
+        id_column=_ID_COLUMN,
+        id_=id_,
+        tag_=tag_,
+        user_id=user_id,
+        expdb=expdb,
+    )
+
+
+async def get_tag(id_: int, tag_: str, expdb: AsyncConnection) -> Row | None:
+    return await select_tag(table=_TABLE, id_column=_ID_COLUMN, id_=id_, tag_=tag_, expdb=expdb)
+
+
+async def delete_tag(id_: int, tag_: str, expdb: AsyncConnection) -> None:
+    await remove_tag(table=_TABLE, id_column=_ID_COLUMN, id_=id_, tag_=tag_, expdb=expdb)
+
+
 async def get_by_name(name: str, external_version: str, expdb: AsyncConnection) -> Row | None:
     """Get flow by name and external version."""
     row = await expdb.execute(

src/database/runs.py

@@ -0,0 +1,44 @@
+from sqlalchemy import Row, text
+from sqlalchemy.ext.asyncio import AsyncConnection
+
+from database.tagging import insert_tag, remove_tag, select_tag, select_tags
+
+_TABLE = "run_tag"
+_ID_COLUMN = "id"
+
+
+async def get(id_: int, expdb: AsyncConnection) -> Row | None:
+    row = await expdb.execute(
+        text(
+            """
+            SELECT *
+            FROM run
+            WHERE `id` = :run_id
+            """,
+        ),
+        parameters={"run_id": id_},
+    )
+    return row.one_or_none()
+
+
+async def get_tags(id_: int, expdb: AsyncConnection) -> list[str]:
+    return await select_tags(table=_TABLE, id_column=_ID_COLUMN, id_=id_, expdb=expdb)
+
+
+async def tag(id_: int, tag_: str, *, user_id: int, expdb: AsyncConnection) -> None:
+    await insert_tag(
+        table=_TABLE,
+        id_column=_ID_COLUMN,
+        id_=id_,
+        tag_=tag_,
+        user_id=user_id,
+        expdb=expdb,
+    )
+
+
+async def get_tag(id_: int, tag_: str, expdb: AsyncConnection) -> Row | None:
+    return await select_tag(table=_TABLE, id_column=_ID_COLUMN, id_=id_, tag_=tag_, expdb=expdb)
+
+
+async def delete_tag(id_: int, tag_: str, expdb: AsyncConnection) -> None:
+    await remove_tag(table=_TABLE, id_column=_ID_COLUMN, id_=id_, tag_=tag_, expdb=expdb)

src/database/tagging.py

@@ -0,0 +1,82 @@
+from sqlalchemy import Row, text
+from sqlalchemy.ext.asyncio import AsyncConnection
+
+
+async def insert_tag(
+    *,
+    table: str,
+    id_column: str,
+    id_: int,
+    tag_: str,
+    user_id: int,
+    expdb: AsyncConnection,
+) -> None:
+    await expdb.execute(
+        text(
+            f"""
+            INSERT INTO {table}(`{id_column}`, `tag`, `uploader`)
+            VALUES (:id, :tag, :user_id)
+            """,
+        ),
+        parameters={"id": id_, "tag": tag_, "user_id": user_id},
+    )
+
+
+async def select_tag(
+    *,
+    table: str,
+    id_column: str,
+    id_: int,
+    tag_: str,
+    expdb: AsyncConnection,
+) -> Row | None:
+    result = await expdb.execute(
+        text(
+            f"""
+            SELECT `{id_column}` as id, `tag`, `uploader`
+            FROM {table}
+            WHERE `{id_column}` = :id AND `tag` = :tag
+            """,
+        ),
+        parameters={"id": id_, "tag": tag_},
+    )
+    return result.one_or_none()
+
+
+async def remove_tag(
+    *,
+    table: str,
+    id_column: str,
+    id_: int,
+    tag_: str,
+    expdb: AsyncConnection,
+) -> None:
+    await expdb.execute(
+        text(
+            f"""
+            DELETE FROM {table}
+            WHERE `{id_column}` = :id AND `tag` = :tag
+            """,
+        ),
+        parameters={"id": id_, "tag": tag_},
+    )
+
+
+async def select_tags(
+    *,
+    table: str,
+    id_column: str,
+    id_: int,
+    expdb: AsyncConnection,
+) -> list[str]:
+    result = await expdb.execute(
+        text(
+            f"""
+            SELECT `tag`
+            FROM {table}
+            WHERE `{id_column}` = :id
+            """,
+        ),
+        parameters={"id": id_},
+    )
+    return [row.tag for row in result.all()]

src/database/tasks.py

@@ -4,6 +4,11 @@
 from sqlalchemy import Row, text
 from sqlalchemy.ext.asyncio import AsyncConnection
 
+from database.tagging import insert_tag, remove_tag, select_tag, select_tags
+
+_TABLE = "task_tag"
+_ID_COLUMN = "id"
+
 
 async def get(id_: int, expdb: AsyncConnection) -> Row | None:
     row = await expdb.execute(
@@ -103,15 +108,23 @@ async def get_task_type_inout_with_template(
 
 
 async def get_tags(id_: int, expdb: AsyncConnection) -> list[str]:
-    rows = await expdb.execute(
-        text(
-            """
-            SELECT `tag`
-            FROM task_tag
-            WHERE `id` = :task_id
-            """,
-        ),
-        parameters={"task_id": id_},
+    return await select_tags(table=_TABLE, id_column=_ID_COLUMN, id_=id_, expdb=expdb)
+
+
+async def tag(id_: int, tag_: str, *, user_id: int, expdb: AsyncConnection) -> None:
+    await insert_tag(
+        table=_TABLE,
+        id_column=_ID_COLUMN,
+        id_=id_,
+        tag_=tag_,
+        user_id=user_id,
+        expdb=expdb,
     )
-    tag_rows = rows.all()
-    return [row.tag for row in tag_rows]
+
+
+async def get_tag(id_: int, tag_: str, expdb: AsyncConnection) -> Row | None:
+    return await select_tag(table=_TABLE, id_column=_ID_COLUMN, id_=id_, tag_=tag_, expdb=expdb)
+
+
+async def delete_tag(id_: int, tag_: str, expdb: AsyncConnection) -> None:
+    await remove_tag(table=_TABLE, id_column=_ID_COLUMN, id_=id_, tag_=tag_, expdb=expdb)

src/main.py

@@ -15,6 +15,7 @@
 from routers.openml.evaluations import router as evaluationmeasures_router
 from routers.openml.flows import router as flows_router
 from routers.openml.qualities import router as qualities_router
+from routers.openml.runs import router as runs_router
 from routers.openml.setups import router as setup_router
 from routers.openml.study import router as study_router
 from routers.openml.tasks import router as task_router
@@ -68,6 +69,7 @@ def create_api() -> FastAPI:
     app.include_router(estimationprocedure_router)
     app.include_router(task_router)
     app.include_router(flows_router)
+    app.include_router(runs_router)
     app.include_router(study_router)
     app.include_router(setup_router)
     return app

src/routers/openml/flows.py

@@ -1,17 +1,57 @@
-from typing import Annotated, Literal
+from typing import Annotated, Any, Literal
 
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Body, Depends
 from sqlalchemy.ext.asyncio import AsyncConnection
 
 import database.flows
 from core.conversions import _str_to_num
 from core.errors import FlowNotFoundError
-from routers.dependencies import expdb_connection
+from core.tagging import tag_entity, untag_entity
+from database.users import User
+from routers.dependencies import expdb_connection, fetch_user_or_raise
+from routers.types import SystemString64
 from schemas.flows import Flow, Parameter, Subflow
 
 router = APIRouter(prefix="/flows", tags=["flows"])
 
 
+@router.post(path="/tag")
+async def tag_flow(
+    flow_id: Annotated[int, Body()],
+    tag: Annotated[str, SystemString64],
+    user: Annotated[User, Depends(fetch_user_or_raise)],
+    expdb: Annotated[AsyncConnection, Depends(expdb_connection)],
+) -> dict[str, dict[str, Any]]:
+    return await tag_entity(
+        flow_id,
+        tag,
+        user,
+        expdb,
+        get_tags_fn=database.flows.get_tags,
+        tag_fn=database.flows.tag,
+        response_key="flow_tag",
+    )
+
+
+@router.post(path="/untag")
+async def untag_flow(
+    flow_id: Annotated[int, Body()],
+    tag: Annotated[str, SystemString64],
+    user: Annotated[User, Depends(fetch_user_or_raise)],
+    expdb: Annotated[AsyncConnection, Depends(expdb_connection)],
+) -> dict[str, dict[str, Any]]:
+    return await untag_entity(
+        flow_id,
+        tag,
+        user,
+        expdb,
+        get_tag_fn=database.flows.get_tag,
+        delete_tag_fn=database.flows.delete_tag,
+        get_tags_fn=database.flows.get_tags,
+        response_key="flow_tag",
+    )
+
+
 @router.get("/exists/{name}/{external_version}")
 async def flow_exists(
     name: str,