Mc 89 separate batches from urls

alembic/versions/2025_07_17_0856-9552d354ccf4_remove_batches_dependency_for_urls.py

@@ -0,0 +1,83 @@
+"""Remove batches dependency for urls
+
+Revision ID: 9552d354ccf4
+Revises: d6519a3ca5c9
+Create Date: 2025-07-17 08:56:22.919486
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+from src.util.alembic_helpers import id_column, created_at_column, updated_at_column, url_id_column, batch_id_column
+
+# revision identifiers, used by Alembic.
+revision: str = '9552d354ccf4'
+down_revision: Union[str, None] = 'd6519a3ca5c9'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+LINK_TABLE_NAME = 'link_batch_urls'
+
+BATCHES_COLUMN_NAME = 'batch_id'
+
+def _create_link_table():
+    op.create_table(
+        LINK_TABLE_NAME,
+        id_column(),
+        batch_id_column(),
+        url_id_column(),
+        created_at_column(),
+        updated_at_column(),
+        sa.UniqueConstraint('url_id', name='uq_link_table_url_id')
+    )
+
+def _drop_link_table():
+    op.drop_table(LINK_TABLE_NAME)
+
+def _migrate_batch_ids_to_link_table():
+    op.execute(f"""
+    INSERT INTO {LINK_TABLE_NAME} (batch_id, url_id)
+    SELECT batch_id, id
+    FROM urls
+    """)
+
+def _migrate_link_table_to_batch_ids():
+    op.execute(f"""
+    UPDATE urls
+    SET batch_id = (
+        SELECT batch_id
+        FROM {LINK_TABLE_NAME}
+        WHERE url_id = urls.id
+    )
+    """)
+
+def _drop_url_batches_column():
+    op.drop_column('urls', BATCHES_COLUMN_NAME)
+
+def _add_url_batches_column():
+    op.add_column(
+        'urls',
+        batch_id_column(nullable=True)
+    )
+
+def _add_not_null_constraint():
+    op.alter_column(
+        'urls',
+        BATCHES_COLUMN_NAME,
+        nullable=False
+    )
+
+
+def upgrade() -> None:
+    _create_link_table()
+    _migrate_batch_ids_to_link_table()
+    _drop_url_batches_column()
+
+
+def downgrade() -> None:
+    _add_url_batches_column()
+    _migrate_link_table_to_batch_ids()
+    _add_not_null_constraint()
+    _drop_link_table()

src/api/endpoints/annotate/_shared/queries/get_annotation_batch_info.py

@@ -5,6 +5,7 @@
 
 from src.api.endpoints.annotate.dtos.shared.batch import AnnotationBatchInfo
 from src.collectors.enums import URLStatus
+from src.db.models.instantiations.link.link_batch_urls import LinkBatchURL
 from src.db.models.instantiations.url.core import URL
 from src.db.queries.base.builder import QueryBuilderBase
 from src.db.statement_composer import StatementComposer
@@ -35,11 +36,14 @@ async def run(
             for model in self.models
         ]
 
-        select_url = select(func.count(URL.id))
+        select_url = (
+            select(func.count(URL.id))
+            .join(LinkBatchURL)
+        )
 
         common_where_clause = [
             URL.outcome == URLStatus.PENDING.value,
-            URL.batch_id == self.batch_id,
+            LinkBatchURL.batch_id == self.batch_id,
         ]
 
         annotated_query = (

src/api/endpoints/annotate/_shared/queries/get_next_url_for_user_annotation.py

@@ -5,6 +5,7 @@
 from src.collectors.enums import URLStatus
 from src.core.enums import SuggestedStatus
 from src.db.client.types import UserSuggestionModel
+from src.db.models.instantiations.link.link_batch_urls import LinkBatchURL
 from src.db.models.instantiations.url.core import URL
 from src.db.models.instantiations.url.suggestion.relevant.user import UserRelevantSuggestion
 from src.db.queries.base.builder import QueryBuilderBase
@@ -27,10 +28,21 @@
         self.auto_suggestion_relationship = auto_suggestion_relationship
 
     async def run(self, session: AsyncSession):
-        url_query = (
+        query = (
             select(
                 URL,
             )
+        )
+
+        if self.batch_id is not None:
+            query = (
+                query
+                .join(LinkBatchURL)
+                .where(LinkBatchURL.batch_id == self.batch_id)
+            )
+
+        query = (
+            query
             .where(URL.outcome == URLStatus.PENDING.value)
             # URL must not have user suggestion
             .where(
@@ -39,7 +51,7 @@
         )
 
         if self.check_if_annotated_not_relevant:
-            url_query = url_query.where(
+            query = query.where(
                 not_(
                     exists(
                         select(UserRelevantSuggestion)
@@ -51,14 +63,13 @@
                 )
             )
 
-        if self.batch_id is not None:
-            url_query = url_query.where(URL.batch_id == self.batch_id)
 
-        url_query = url_query.options(
+
+        query = query.options(
             joinedload(self.auto_suggestion_relationship),
             joinedload(URL.html_content)
         ).limit(1)
 
-        raw_result = await session.execute(url_query)
+        raw_result = await session.execute(query)
 
         return raw_result.unique().scalars().one_or_none()

src/api/endpoints/annotate/agency/get/queries/next_for_annotation.py

@@ -10,6 +10,7 @@
 from src.core.tasks.url.operators.url_html.scraper.parser.util import convert_to_response_html_info
 from src.db.dtos.url.mapping import URLMapping
 from src.db.models.instantiations.confirmed_url_agency import ConfirmedURLAgency
+from src.db.models.instantiations.link.link_batch_urls import LinkBatchURL
 from src.db.models.instantiations.url.core import URL
 from src.db.models.instantiations.url.suggestion.agency.auto import AutomatedUrlAgencySuggestion
 from src.db.models.instantiations.url.suggestion.agency.user import UserUrlAgencySuggestion
@@ -41,20 +42,19 @@ async def run(
             have extant autosuggestions
         """
         # Select statement
-        statement = (
-            select(URL.id, URL.url)
-            # Must not have confirmed agencies
-            .where(
-                URL.outcome == URLStatus.PENDING.value
-            )
+        query = select(URL.id, URL.url)
+        if self.batch_id is not None:
+            query = query.join(LinkBatchURL).where(LinkBatchURL.batch_id == self.batch_id)
+
+        # Must not have confirmed agencies
+        query = query.where(
+            URL.outcome == URLStatus.PENDING.value
         )
 
-        if self.batch_id is not None:
-            statement = statement.where(URL.batch_id == self.batch_id)
 
         # Must not have been annotated by a user
-        statement = (
-            statement.join(UserUrlAgencySuggestion, isouter=True)
+        query = (
+            query.join(UserUrlAgencySuggestion, isouter=True)
             .where(
                 ~exists(
                     select(UserUrlAgencySuggestion).
@@ -93,7 +93,7 @@ async def run(
                 )
             )
         ).limit(1)
-        raw_result = await session.execute(statement)
+        raw_result = await session.execute(query)
         results = raw_result.all()
         if len(results) == 0:
             return GetNextURLForAgencyAnnotationResponse(

src/api/endpoints/annotate/all/get/query.py

@@ -11,6 +11,7 @@
 from src.collectors.enums import URLStatus
 from src.db.dto_converter import DTOConverter
 from src.db.dtos.url.mapping import URLMapping
+from src.db.models.instantiations.link.link_batch_urls import LinkBatchURL
 from src.db.models.instantiations.url.core import URL
 from src.db.models.instantiations.url.suggestion.agency.user import UserUrlAgencySuggestion
 from src.db.models.instantiations.url.suggestion.record_type.user import UserRecordTypeSuggestion
@@ -32,8 +33,11 @@ async def run(
         self,
         session: AsyncSession
     ) -> GetNextURLForAllAnnotationResponse:
+        query = Select(URL)
+        if self.batch_id is not None:
+            query = query.join(LinkBatchURL).where(LinkBatchURL.batch_id == self.batch_id)
         query = (
-            Select(URL)
+            query
             .where(
                 and_(
                     URL.outcome == URLStatus.PENDING.value,
@@ -43,8 +47,7 @@ async def run(
                 )
             )
         )
-        if self.batch_id is not None:
-            query = query.where(URL.batch_id == self.batch_id)
+
 
         load_options = [
             URL.html_content,

src/api/endpoints/batch/duplicates/query.py

@@ -0,0 +1,60 @@
+from sqlalchemy import Select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import aliased
+
+from src.db.dtos.duplicate import DuplicateInfo
+from src.db.models.instantiations.batch import Batch
+from src.db.models.instantiations.duplicate import Duplicate
+from src.db.models.instantiations.link.link_batch_urls import LinkBatchURL
+from src.db.models.instantiations.url.core import URL
+from src.db.queries.base.builder import QueryBuilderBase
+
+
+class GetDuplicatesByBatchIDQueryBuilder(QueryBuilderBase):
+
+    def __init__(
+        self,
+        batch_id: int,
+        page: int
+    ):
+        super().__init__()
+        self.batch_id = batch_id
+        self.page = page
+
+    async def run(self, session: AsyncSession) -> list[DuplicateInfo]:
+        original_batch = aliased(Batch)
+        duplicate_batch = aliased(Batch)
+
+        query = (
+            Select(
+                URL.url.label("source_url"),
+                URL.id.label("original_url_id"),
+                duplicate_batch.id.label("duplicate_batch_id"),
+                duplicate_batch.parameters.label("duplicate_batch_parameters"),
+                original_batch.id.label("original_batch_id"),
+                original_batch.parameters.label("original_batch_parameters"),
+            )
+            .select_from(Duplicate)
+            .join(URL, Duplicate.original_url_id == URL.id)
+            .join(duplicate_batch, Duplicate.batch_id == duplicate_batch.id)
+            .join(LinkBatchURL, URL.id == LinkBatchURL.url_id)
+            .join(original_batch, LinkBatchURL.batch_id == original_batch.id)
+            .filter(duplicate_batch.id == self.batch_id)
+            .limit(100)
+            .offset((self.page - 1) * 100)
+        )
+        raw_results = await session.execute(query)
+        results = raw_results.all()
+        final_results = []
+        for result in results:
+            final_results.append(
+                DuplicateInfo(
+                    source_url=result.source_url,
+                    duplicate_batch_id=result.duplicate_batch_id,
+                    duplicate_metadata=result.duplicate_batch_parameters,
+                    original_batch_id=result.original_batch_id,
+                    original_metadata=result.original_batch_parameters,
+                    original_url_id=result.original_url_id
+                )
+            )
+        return final_results

src/api/endpoints/batch/routes.py

@@ -4,12 +4,12 @@
 from fastapi.params import Query, Depends
 
 from src.api.dependencies import get_async_core
-from src.api.endpoints.batch.dtos.get.duplicates import GetDuplicatesByBatchResponse
 from src.api.endpoints.batch.dtos.get.logs import GetBatchLogsResponse
 from src.api.endpoints.batch.dtos.get.summaries.response import GetBatchSummariesResponse
 from src.api.endpoints.batch.dtos.get.summaries.summary import BatchSummary
-from src.api.endpoints.batch.dtos.get.urls import GetURLsByBatchResponse
 from src.api.endpoints.batch.dtos.post.abort import MessageResponse
+from src.api.endpoints.batch.duplicates.dto import GetDuplicatesByBatchResponse
+from src.api.endpoints.batch.urls.dto import GetURLsByBatchResponse
 from src.collectors.enums import CollectorType
 from src.core.core import AsyncCore
 from src.core.enums import BatchStatus

src/api/endpoints/batch/urls/query.py

@@ -0,0 +1,31 @@
+from sqlalchemy import Select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.db.dtos.url.core import URLInfo
+from src.db.models.instantiations.link.link_batch_urls import LinkBatchURL
+from src.db.models.instantiations.url.core import URL
+from src.db.queries.base.builder import QueryBuilderBase
+
+
+class GetURLsByBatchQueryBuilder(QueryBuilderBase):
+
+    def __init__(
+        self,
+        batch_id: int,
+        page: int = 1
+    ):
+        super().__init__()
+        self.batch_id = batch_id
+        self.page = page
+
+    async def run(self, session: AsyncSession) -> list[URLInfo]:
+        query = (
+            Select(URL)
+            .join(LinkBatchURL)
+            .where(LinkBatchURL.batch_id == self.batch_id)
+            .order_by(URL.id)
+            .limit(100)
+            .offset((self.page - 1) * 100))
+        result = await session.execute(query)
+        urls = result.scalars().all()
+        return [URLInfo(**url.__dict__) for url in urls]