[DOP-31839] Add job tags

Author: dolfinus

data_rentgen/VERSION

@@ -1 +1 @@
-0.4.8
+0.5.0

data_rentgen/db/migrations/versions/2026-01-21_675ed36b7807_add_job_tag.py

@@ -0,0 +1,37 @@
+# SPDX-FileCopyrightText: 2024-present MTS PJSC
+# SPDX-License-Identifier: Apache-2.0
+"""Add job_tag
+
+Revision ID: 675ed36b7807
+Revises: f1c76c422433
+Create Date: 2026-01-21 11:30:01.406676
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "675ed36b7807"
+down_revision = "f1c76c422433"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "job_tag",
+        sa.Column("job_id", sa.BigInteger(), nullable=False),
+        sa.Column("tag_value_id", sa.BigInteger(), nullable=False),
+        sa.ForeignKeyConstraint(["job_id"], ["job.id"], name=op.f("fk__job_tag__job_id__job"), ondelete="CASCADE"),
+        sa.ForeignKeyConstraint(
+            ["tag_value_id"], ["tag_value.id"], name=op.f("fk__job_tag__tag_value_id__tag_value"), ondelete="CASCADE"
+        ),
+        sa.PrimaryKeyConstraint("job_id", "tag_value_id", name=op.f("pk__job_tag")),
+    )
+    op.create_index("ix__job_tag__tag_value_id", "job_tag", ["tag_value_id"], unique=False)
+
+
+def downgrade() -> None:
+    op.drop_index("ix__job_tag__tag_value_id", table_name="job_tag")
+    op.drop_table("job_tag")

data_rentgen/db/migrations/versions/2026-01-21_f1c76c422433_rename_table_dataset_tags_to_dataset_tag.py

@@ -0,0 +1,27 @@
+# SPDX-FileCopyrightText: 2024-present MTS PJSC
+# SPDX-License-Identifier: Apache-2.0
+"""Rename table dataset_tags to dataset_tag
+
+Revision ID: f1c76c422433
+Revises: 8e8891273099
+Create Date: 2026-01-21 11:27:20.255770
+
+"""
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "f1c76c422433"
+down_revision = "8e8891273099"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.rename_table("dataset_tags", "dataset_tag")
+    op.execute("ALTER INDEX ix__dataset_tags__tag_value_id RENAME TO ix__dataset_tag__tag_value_id")
+
+
+def downgrade() -> None:
+    op.rename_table("dataset_tag", "dataset_tags")
+    op.execute("ALTER INDEX ix__dataset_tag__tag_value_id RENAME TO ix__dataset_tags__tag_value_id")

data_rentgen/db/models/dataset.py

@@ -39,7 +39,10 @@ class Dataset(Base):
         doc="Dataset name, e.g. table name or filesystem path",
     )
 
-    tag_values: Mapped[set[TagValue]] = relationship(secondary=lambda: dataset_tags_table)
+    tag_values: Mapped[set[TagValue]] = relationship(
+        secondary=lambda: dataset_tags_table,
+        doc="Dataset tag values",
+    )
 
     search_vector: Mapped[str] = mapped_column(
         TSVECTOR,
@@ -63,9 +66,9 @@ class Dataset(Base):
 
 
 dataset_tags_table: Table = Table(
-    "dataset_tags",
+    "dataset_tag",
     Base.metadata,
     Column("dataset_id", ForeignKey("dataset.id", ondelete="CASCADE"), primary_key=True),
     Column("tag_value_id", ForeignKey("tag_value.id", ondelete="CASCADE"), primary_key=True),
-    Index("ix__dataset_tags__tag_value_id", "tag_value_id"),
+    Index("ix__dataset_tag__tag_value_id", "tag_value_id"),
 )

data_rentgen/db/models/job.py

@@ -3,13 +3,14 @@
 
 from __future__ import annotations
 
-from sqlalchemy import BigInteger, Column, Computed, ForeignKey, Index, String, column, func, select
+from sqlalchemy import BigInteger, Column, Computed, ForeignKey, Index, String, Table, column, func, select
 from sqlalchemy.dialects.postgresql import TSVECTOR
 from sqlalchemy.orm import Mapped, column_property, mapped_column, relationship
 
 from data_rentgen.db.models.base import Base
 from data_rentgen.db.models.job_type import JobType
 from data_rentgen.db.models.location import Location
+from data_rentgen.db.models.tag_value import TagValue
 
 
 class Job(Base):
@@ -47,6 +48,11 @@ class Job(Base):
         select(JobType.type).where(Column("type_id") == JobType.id).scalar_subquery(),
     )
 
+    tag_values: Mapped[set[TagValue]] = relationship(
+        secondary=lambda: job_tag_table,
+        doc="Job tag values",
+    )
+
     search_vector: Mapped[str] = mapped_column(
         TSVECTOR,
         Computed(
@@ -66,3 +72,12 @@ class Job(Base):
         deferred=True,
         doc="Full-text search vector",
     )
+
+
+job_tag_table: Table = Table(
+    "job_tag",
+    Base.metadata,
+    Column("job_id", ForeignKey("job.id", ondelete="CASCADE"), primary_key=True),
+    Column("tag_value_id", ForeignKey("tag_value.id", ondelete="CASCADE"), primary_key=True),
+    Index("ix__job_tag__tag_value_id", "tag_value_id"),
+)

data_rentgen/db/repositories/job.py

@@ -16,14 +16,15 @@
     bindparam,
     cast,
     desc,
+    distinct,
     func,
     select,
     tuple_,
     union,
 )
 from sqlalchemy.orm import selectinload
 
-from data_rentgen.db.models import Address, Job, Location
+from data_rentgen.db.models import Address, Job, Location, TagValue
 from data_rentgen.db.repositories.base import Repository
 from data_rentgen.db.utils.search import make_tsquery, ts_match, ts_rank
 from data_rentgen.dto import JobDTO, PaginationDTO
@@ -76,6 +77,7 @@ async def paginate(
         page_size: int,
         job_ids: Collection[int],
         job_types: Collection[str],
+        tag_value_ids: Collection[int],
         location_ids: Collection[int],
         location_types: Collection[str],
         search_query: str | None,
@@ -92,6 +94,18 @@ async def paginate(
             location_type_lower = [location_type.lower() for location_type in location_types]
             where.append(Location.type == any_(location_type_lower))  # type: ignore[arg-type]
 
+        if tag_value_ids:
+            tv_ids = list(tag_value_ids)
+            job_ids_subq = (
+                select(Job.id)
+                .join(Job.tag_values)
+                .where(TagValue.id.in_(tv_ids))
+                .group_by(Job.id)
+                # If multiple tag values are passed, job should have both of them (AND, not OR)
+                .having(func.count(distinct(TagValue.id)) == len(tv_ids))
+            )
+            where.append(Job.id.in_(job_ids_subq))
+
         query: Select | CompoundSelect
         order_by: list[ColumnElement | SQLColumnExpression]
         if search_query:
@@ -128,7 +142,10 @@ async def paginate(
             query = select(Job).join(Location, location_join_clause).where(*where)
             order_by = [Job.name]
 
-        options = [selectinload(Job.location).selectinload(Location.addresses)]
+        options = [
+            selectinload(Job.location).selectinload(Location.addresses),
+            selectinload(Job.tag_values).selectinload(TagValue.tag),
+        ]
         return await self._paginate_by_query(
             query=query,
             order_by=order_by,

data_rentgen/server/api/v1/router/job.py

@@ -43,6 +43,7 @@ async def paginate_jobs(
         page_size=query_args.page_size,
         job_ids=query_args.job_id,
         job_types=query_args.job_type,
+        tag_value_ids=query_args.tag_value_id,
         location_ids=query_args.location_id,
         location_types=query_args.location_type,
         search_query=query_args.search_query,

data_rentgen/server/schemas/v1/job.py

@@ -6,6 +6,7 @@
 
 from data_rentgen.server.schemas.v1.location import LocationResponseV1
 from data_rentgen.server.schemas.v1.pagination import PaginateQueryV1
+from data_rentgen.server.schemas.v1.tag import TagResponseV1
 
 
 class JobResponseV1(BaseModel):
@@ -22,6 +23,7 @@ class JobResponseV1(BaseModel):
 class JobDetailedResponseV1(BaseModel):
     id: str = Field(description="Job id", coerce_numbers_to_str=True)
     data: JobResponseV1 = Field(description="Job data")
+    tags: list[TagResponseV1] = Field(default_factory=list, description="Job tags")
 
     model_config = ConfigDict(from_attributes=True)
 
@@ -55,6 +57,14 @@ class JobPaginateQueryV1(PaginateQueryV1):
         description="Types of jobs",
         examples=[["SPARK_APPLICATION", "AIRFLOW_DAG"]],
     )
+    tag_value_id: list[int] = Field(
+        default_factory=list,
+        description=(
+            "Get jobs having specific tag values assigned. "
+            "If multiple values are passed, job should have all of them (AND, not OR)"
+        ),
+        examples=[[123]],
+    )
     location_id: list[int] = Field(
         default_factory=list,
         description="Ids of locations the job started at",

data_rentgen/server/services/dataset.py

@@ -7,7 +7,7 @@
 
 from fastapi import Depends
 
-from data_rentgen.db.models.location import Location
+from data_rentgen.db.models import Location
 from data_rentgen.dto.pagination import PaginationDTO
 from data_rentgen.server.services.tag import TagData, TagValueData
 from data_rentgen.services.uow import UnitOfWork
@@ -18,7 +18,6 @@ class DatasetData:
     id: int
     name: str
     location: Location
-    schema = None
 
 
 @dataclass

data_rentgen/server/services/job.py

@@ -2,19 +2,30 @@
 # SPDX-License-Identifier: Apache-2.0
 from collections.abc import Collection, Sequence
 from dataclasses import dataclass
+from itertools import groupby
 from typing import Annotated
 
 from fastapi import Depends
 
-from data_rentgen.db.models.job import Job
+from data_rentgen.db.models import Location
 from data_rentgen.dto.pagination import PaginationDTO
+from data_rentgen.server.services.tag import TagData, TagValueData
 from data_rentgen.services.uow import UnitOfWork
 
 
+@dataclass
+class JobData:
+    id: int
+    name: str
+    type: str
+    location: Location
+
+
 @dataclass
 class JobServiceResult:
     id: int
-    data: Job
+    data: JobData
+    tags: list[TagData]
 
 
 class JobServicePaginatedResult(PaginationDTO[JobServiceResult]):
@@ -31,6 +42,7 @@ async def paginate(
         page_size: int,
         job_ids: Collection[int],
         job_types: Collection[str],
+        tag_value_ids: Collection[int],
         location_ids: Collection[int],
         location_types: Collection[str],
         search_query: str | None,
@@ -40,6 +52,7 @@ async def paginate(
             page_size=page_size,
             job_ids=job_ids,
             job_types=job_types,
+            tag_value_ids=tag_value_ids,
             location_ids=location_ids,
             location_types=location_types,
             search_query=search_query,
@@ -49,7 +62,31 @@ async def paginate(
             page=pagination.page,
             page_size=pagination.page_size,
             total_count=pagination.total_count,
-            items=[JobServiceResult(id=job.id, data=job) for job in pagination.items],
+            items=[
+                JobServiceResult(
+                    id=job.id,
+                    data=JobData(
+                        id=job.id,
+                        name=job.name,
+                        type=job.type,
+                        location=job.location,
+                    ),
+                    tags=[
+                        TagData(
+                            id=tag.id,
+                            name=tag.name,
+                            values=[
+                                TagValueData(id=tv.id, value=tv.value) for tv in sorted(group, key=lambda tv: tv.value)
+                            ],
+                        )
+                        for tag, group in groupby(
+                            sorted(job.tag_values, key=lambda tv: tv.tag.name),
+                            key=lambda tv: tv.tag,
+                        )
+                    ],
+                )
+                for job in pagination.items
+            ],
         )
 
     async def get_job_types(self) -> Sequence[str]: