Add anonymous annotation endpoint

alembic/versions/2025_10_13_2007-7aace6587d1a_add_anonymous_annotation_tables.py

@@ -0,0 +1,60 @@
+"""Add anonymous annotation tables
+
+Revision ID: 7aace6587d1a
+Revises: 43077d7e08c5
+Create Date: 2025-10-13 20:07:18.388899
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+from src.util.alembic_helpers import url_id_column, agency_id_column, created_at_column, location_id_column, enum_column
+
+# revision identifiers, used by Alembic.
+revision: str = '7aace6587d1a'
+down_revision: Union[str, None] = '43077d7e08c5'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "anonymous_annotation_agency",
+        url_id_column(),
+        agency_id_column(),
+        created_at_column(),
+        sa.PrimaryKeyConstraint('url_id', 'agency_id')
+    )
+    op.create_table(
+        "anonymous_annotation_location",
+        url_id_column(),
+        location_id_column(),
+        created_at_column(),
+        sa.PrimaryKeyConstraint('url_id', 'location_id')
+    )
+    op.create_table(
+        "anonymous_annotation_record_type",
+        url_id_column(),
+        enum_column(
+            column_name="record_type",
+            enum_name="record_type"
+        ),
+        created_at_column(),
+        sa.PrimaryKeyConstraint('url_id', 'record_type')
+    )
+    op.create_table(
+        "anonymous_annotation_url_type",
+        url_id_column(),
+        enum_column(
+            column_name="url_type",
+            enum_name="url_type"
+        ),
+        created_at_column(),
+        sa.PrimaryKeyConstraint('url_id', 'url_type')
+    )
+
+
+def downgrade() -> None:
+    pass

src/api/endpoints/annotate/_shared/extract.py

@@ -0,0 +1,64 @@
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.api.endpoints.annotate._shared.queries.get_annotation_batch_info import GetAnnotationBatchInfoQueryBuilder
+from src.api.endpoints.annotate.all.get.models.agency import AgencyAnnotationResponseOuterInfo
+from src.api.endpoints.annotate.all.get.models.location import LocationAnnotationResponseOuterInfo
+from src.api.endpoints.annotate.all.get.models.name import NameAnnotationSuggestion
+from src.api.endpoints.annotate.all.get.models.record_type import RecordTypeAnnotationSuggestion
+from src.api.endpoints.annotate.all.get.models.response import GetNextURLForAllAnnotationResponse, \
+    GetNextURLForAllAnnotationInnerResponse
+from src.api.endpoints.annotate.all.get.models.url_type import URLTypeAnnotationSuggestion
+from src.api.endpoints.annotate.all.get.queries.agency.core import GetAgencySuggestionsQueryBuilder
+from src.api.endpoints.annotate.all.get.queries.convert import \
+    convert_user_url_type_suggestion_to_url_type_annotation_suggestion, \
+    convert_user_record_type_suggestion_to_record_type_annotation_suggestion
+from src.api.endpoints.annotate.all.get.queries.location_.core import GetLocationSuggestionsQueryBuilder
+from src.api.endpoints.annotate.all.get.queries.name.core import GetNameSuggestionsQueryBuilder
+from src.db.dto_converter import DTOConverter
+from src.db.dtos.url.mapping import URLMapping
+from src.db.models.impl.url.core.sqlalchemy import URL
+from src.db.models.impl.url.suggestion.agency.user import UserUrlAgencySuggestion
+
+
+async def extract_and_format_get_annotation_result(
+    session: AsyncSession,
+    url: URL,
+    batch_id: int | None = None
+):
+    html_response_info = DTOConverter.html_content_list_to_html_response_info(
+        url.html_content
+    )
+    url_type_suggestions: list[URLTypeAnnotationSuggestion] = \
+        convert_user_url_type_suggestion_to_url_type_annotation_suggestion(
+            url.user_relevant_suggestions
+        )
+    record_type_suggestions: list[RecordTypeAnnotationSuggestion] = \
+        convert_user_record_type_suggestion_to_record_type_annotation_suggestion(
+            url.user_record_type_suggestions
+        )
+    agency_suggestions: AgencyAnnotationResponseOuterInfo = \
+        await GetAgencySuggestionsQueryBuilder(url_id=url.id).run(session)
+    location_suggestions: LocationAnnotationResponseOuterInfo = \
+        await GetLocationSuggestionsQueryBuilder(url_id=url.id).run(session)
+    name_suggestions: list[NameAnnotationSuggestion] = \
+        await GetNameSuggestionsQueryBuilder(url_id=url.id).run(session)
+    return GetNextURLForAllAnnotationResponse(
+        next_annotation=GetNextURLForAllAnnotationInnerResponse(
+            url_info=URLMapping(
+                url_id=url.id,
+                url=url.url
+            ),
+            html_info=html_response_info,
+            url_type_suggestions=url_type_suggestions,
+            record_type_suggestions=record_type_suggestions,
+            agency_suggestions=agency_suggestions,
+            batch_info=await GetAnnotationBatchInfoQueryBuilder(
+                batch_id=batch_id,
+                models=[
+                    UserUrlAgencySuggestion,
+                ]
+            ).run(session),
+            location_suggestions=location_suggestions,
+            name_suggestions=name_suggestions
+        )
+    )

src/api/endpoints/annotate/all/get/queries/core.py

@@ -2,23 +2,9 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import joinedload
 
-from src.api.endpoints.annotate._shared.queries.get_annotation_batch_info import GetAnnotationBatchInfoQueryBuilder
-from src.api.endpoints.annotate.all.get.models.agency import AgencyAnnotationResponseOuterInfo
-from src.api.endpoints.annotate.all.get.models.location import LocationAnnotationResponseOuterInfo
-from src.api.endpoints.annotate.all.get.models.name import NameAnnotationSuggestion
-from src.api.endpoints.annotate.all.get.models.record_type import RecordTypeAnnotationSuggestion
-from src.api.endpoints.annotate.all.get.models.response import GetNextURLForAllAnnotationResponse, \
-    GetNextURLForAllAnnotationInnerResponse
-from src.api.endpoints.annotate.all.get.models.url_type import URLTypeAnnotationSuggestion
-from src.api.endpoints.annotate.all.get.queries.agency.core import GetAgencySuggestionsQueryBuilder
-from src.api.endpoints.annotate.all.get.queries.convert import \
-    convert_user_url_type_suggestion_to_url_type_annotation_suggestion, \
-    convert_user_record_type_suggestion_to_record_type_annotation_suggestion
-from src.api.endpoints.annotate.all.get.queries.location_.core import GetLocationSuggestionsQueryBuilder
-from src.api.endpoints.annotate.all.get.queries.name.core import GetNameSuggestionsQueryBuilder
+from src.api.endpoints.annotate._shared.extract import extract_and_format_get_annotation_result
+from src.api.endpoints.annotate.all.get.models.response import GetNextURLForAllAnnotationResponse
 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.impl.flag.url_suspended.sqlalchemy import FlagURLSuspended
 from src.db.models.impl.link.batch_url.sqlalchemy import LinkBatchURL
 from src.db.models.impl.url.core.sqlalchemy import URL
@@ -135,43 +121,5 @@ async def run(
                 next_annotation=None
             )
 
-        html_response_info = DTOConverter.html_content_list_to_html_response_info(
-            url.html_content
-        )
-
-        url_type_suggestions: list[URLTypeAnnotationSuggestion] = \
-            convert_user_url_type_suggestion_to_url_type_annotation_suggestion(
-                url.user_relevant_suggestions
-            )
-        record_type_suggestions: list[RecordTypeAnnotationSuggestion] = \
-            convert_user_record_type_suggestion_to_record_type_annotation_suggestion(
-                url.user_record_type_suggestions
-            )
-        agency_suggestions: AgencyAnnotationResponseOuterInfo = \
-            await GetAgencySuggestionsQueryBuilder(url_id=url.id).run(session)
-        location_suggestions: LocationAnnotationResponseOuterInfo = \
-            await GetLocationSuggestionsQueryBuilder(url_id=url.id).run(session)
-        name_suggestions: list[NameAnnotationSuggestion] = \
-            await GetNameSuggestionsQueryBuilder(url_id=url.id).run(session)
-
+        return await extract_and_format_get_annotation_result(session, url=url, batch_id=self.batch_id)
 
-        return GetNextURLForAllAnnotationResponse(
-            next_annotation=GetNextURLForAllAnnotationInnerResponse(
-                url_info=URLMapping(
-                    url_id=url.id,
-                    url=url.url
-                ),
-                html_info=html_response_info,
-                url_type_suggestions=url_type_suggestions,
-                record_type_suggestions=record_type_suggestions,
-                agency_suggestions=agency_suggestions,
-                batch_info=await GetAnnotationBatchInfoQueryBuilder(
-                    batch_id=self.batch_id,
-                    models=[
-                        UserUrlAgencySuggestion,
-                    ]
-                ).run(session),
-                location_suggestions=location_suggestions,
-                name_suggestions=name_suggestions
-            )
-        )

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

@@ -3,10 +3,6 @@
 from src.api.endpoints.annotate.all.post.models.request import AllAnnotationPostInfo
 from src.api.endpoints.annotate.all.post.requester import AddAllAnnotationsToURLRequester
 from src.db.models.impl.flag.url_validated.enums import URLType
-from src.db.models.impl.url.suggestion.agency.user import UserUrlAgencySuggestion
-from src.db.models.impl.url.suggestion.location.user.sqlalchemy import UserLocationSuggestion
-from src.db.models.impl.url.suggestion.record_type.user import UserRecordTypeSuggestion
-from src.db.models.impl.url.suggestion.relevant.user import UserURLTypeSuggestion
 from src.db.queries.base.builder import QueryBuilderBase
 
 

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

@@ -0,0 +1,61 @@
+from typing import Any
+
+from sqlalchemy import Select, func
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import joinedload
+
+from src.api.endpoints.annotate._shared.extract import extract_and_format_get_annotation_result
+from src.api.endpoints.annotate.all.get.models.response import GetNextURLForAllAnnotationResponse
+from src.collectors.enums import URLStatus
+from src.db.helpers.query import not_exists_url
+from src.db.models.impl.url.core.sqlalchemy import URL
+from src.db.models.impl.url.suggestion.anonymous.url_type.sqlalchemy import AnonymousAnnotationURLType
+from src.db.models.views.unvalidated_url import UnvalidatedURL
+from src.db.models.views.url_anno_count import URLAnnotationCount
+from src.db.models.views.url_annotations_flags import URLAnnotationFlagsView
+from src.db.queries.base.builder import QueryBuilderBase
+
+
+class GetNextURLForAnonymousAnnotationQueryBuilder(QueryBuilderBase):
+
+    async def run(self, session: AsyncSession) -> GetNextURLForAllAnnotationResponse:
+
+        query = (
+            Select(URL)
+            # URL Must be unvalidated
+            .join(
+                UnvalidatedURL,
+                UnvalidatedURL.url_id == URL.id
+            )
+            .join(
+                URLAnnotationFlagsView,
+                URLAnnotationFlagsView.url_id == URL.id
+            )
+            .join(
+                URLAnnotationCount,
+                URLAnnotationCount.url_id == URL.id
+            )
+            .where(
+                URL.status == URLStatus.OK.value,
+                not_exists_url(AnonymousAnnotationURLType)
+            )
+            .options(
+                joinedload(URL.html_content),
+                joinedload(URL.user_relevant_suggestions),
+                joinedload(URL.user_record_type_suggestions),
+                joinedload(URL.name_suggestions),
+            )
+            .order_by(
+                func.random()
+            )
+            .limit(1)
+        )
+
+        raw_results = (await session.execute(query)).unique()
+        url: URL | None = raw_results.scalars().one_or_none()
+        if url is None:
+            return GetNextURLForAllAnnotationResponse(
+                next_annotation=None
+            )
+
+        return await extract_and_format_get_annotation_result(session, url=url)

src/api/endpoints/annotate/anonymous/post/query.py

@@ -0,0 +1,56 @@
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.api.endpoints.annotate.all.post.models.request import AllAnnotationPostInfo
+from src.db.models.impl.url.suggestion.anonymous.agency.sqlalchemy import AnonymousAnnotationAgency
+from src.db.models.impl.url.suggestion.anonymous.location.sqlalchemy import AnonymousAnnotationLocation
+from src.db.models.impl.url.suggestion.anonymous.record_type.sqlalchemy import AnonymousAnnotationRecordType
+from src.db.models.impl.url.suggestion.anonymous.url_type.sqlalchemy import AnonymousAnnotationURLType
+from src.db.queries.base.builder import QueryBuilderBase
+
+
+class AddAnonymousAnnotationsToURLQueryBuilder(QueryBuilderBase):
+    def __init__(
+        self,
+        url_id: int,
+        post_info: AllAnnotationPostInfo
+    ):
+        super().__init__()
+        self.url_id = url_id
+        self.post_info = post_info
+
+    async def run(self, session: AsyncSession) -> None:
+
+        url_type_suggestion = AnonymousAnnotationURLType(
+            url_id=self.url_id,
+            url_type=self.post_info.suggested_status
+        )
+        session.add(url_type_suggestion)
+
+        if self.post_info.record_type is not None:
+            record_type_suggestion = AnonymousAnnotationRecordType(
+                url_id=self.url_id,
+                record_type=self.post_info.record_type
+            )
+            session.add(record_type_suggestion)
+
+        if len(self.post_info.location_info.location_ids) != 0:
+            location_suggestions = [
+                AnonymousAnnotationLocation(
+                    url_id=self.url_id,
+                    location_id=location_id
+                )
+                for location_id in self.post_info.location_info.location_ids
+            ]
+            session.add_all(location_suggestions)
+
+        if len(self.post_info.agency_info.agency_ids) != 0:
+            agency_suggestions = [
+                AnonymousAnnotationAgency(
+                    url_id=self.url_id,
+                    agency_id=agency_id
+                )
+                for agency_id in self.post_info.agency_info.agency_ids
+            ]
+            session.add_all(agency_suggestions)
+
+        # Ignore Name suggestions