Update draft

Author: maxachis

ENV.md

@@ -70,6 +70,15 @@ Note that some tasks/subtasks are themselves enabled by other tasks.
 | `DELETE_STALE_SCREENSHOTS_TASK_FLAG` | Deletes stale screenshots for URLs already validated.                         |
 | `TASK_CLEANUP_TASK_FLAG` | Cleans up tasks that are no longer needed. |
 | `REFRESH_MATERIALIZED_VIEWS_TASK_FLAG` | Refreshes materialized views. |
+| `DS_APP_SYNC_AGENCY_ADD_FLAG` | Adds new agencies to the Data Sources App|
+| `DS_APP_SYNC_AGENCY_UPDATE_FLAG` | Updates existing agencies in the Data Sources App|
+| `DS_APP_SYNC_AGENCY_DELETE_FLAG` | Deletes agencies in the Data Sources App|
+| `DS_APP_SYNC_DATA_SOURCE_ADD_FLAG` | Adds new data sources to the Data Sources App|
+| `DS_APP_SYNC_DATA_SOURCE_UPDATE_FLAG` | Updates existing data sources in the Data Sources App|
+| `DS_APP_SYNC_DATA_SOURCE_DELETE_FLAG` | Deletes data sources in the Data Sources App|
+| `DS_APP_SYNC_META_URL_ADD_FLAG` | Adds new meta URLs to the Data Sources App|
+| `DS_APP_SYNC_META_URL_UPDATE_FLAG` | Updates existing meta URLs in the Data Sources App|
+| `DS_APP_SYNC_META_URL_DELETE_FLAG` | Deletes meta URLs in the Data Sources App|
 
 ### URL Task Flags
 
@@ -81,7 +90,6 @@ URL Task Flags are collectively controlled by the `RUN_URL_TASKS_TASK_FLAG` flag
 | `URL_HTML_TASK_FLAG`                | URL HTML scraping task.                               |
 | `URL_RECORD_TYPE_TASK_FLAG`         | Automatically assigns Record Types to URLs.           |
 | `URL_AGENCY_IDENTIFICATION_TASK_FLAG` | Automatically assigns and suggests Agencies for URLs. |
-| `URL_SUBMIT_APPROVED_TASK_FLAG`     | Submits approved URLs to the Data Sources App.        |
 | `URL_MISC_METADATA_TASK_FLAG`       | Adds misc metadata to URLs.                           |
 | `URL_AUTO_RELEVANCE_TASK_FLAG`      | Automatically assigns Relevances to URLs.             |
 | `URL_PROBE_TASK_FLAG`               | Probes URLs for web metadata.                         |
@@ -90,7 +98,6 @@ URL Task Flags are collectively controlled by the `RUN_URL_TASKS_TASK_FLAG` flag
 | `URL_AUTO_VALIDATE_TASK_FLAG`       | Automatically validates URLs.                         |
 | `URL_AUTO_NAME_TASK_FLAG`           | Automatically names URLs.                             |
 | `URL_SUSPEND_TASK_FLAG`             | Suspends URLs meeting suspension criteria.            |
-| `URL_SUBMIT_META_URLS_TASK_FLAG`    | Submits meta URLs to the Data Sources App.            |
 
 ### Agency ID Subtasks
 

alembic/versions/2025_10_28_1539-a57c3b5b6e93_add_sync_log_table.py

@@ -10,7 +10,7 @@
 from alembic import op
 import sqlalchemy as sa
 
-from src.util.alembic_helpers import created_at_column
+from src.util.alembic_helpers import created_at_column, updated_at_column, create_updated_at_trigger
 
 # revision identifiers, used by Alembic.
 revision: str = 'a57c3b5b6e93'
@@ -90,6 +90,9 @@ def _add_link_table_modification_triggers():
     )
 
 
+
+
+
 def upgrade() -> None:
     _create_sync_log()
     _create_ds_agency_link()
@@ -102,6 +105,16 @@ def upgrade() -> None:
     _add_flag_deletion_tables()
     _add_last_synced_at_columns()
     _add_link_table_modification_triggers()
+    _add_updated_at_to_optional_data_source_metadata_table()
+
+def _add_updated_at_to_optional_data_source_metadata_table():
+    op.add_column(
+        "url_optional_data_source_metadata",
+        updated_at_column()
+    )
+    create_updated_at_trigger(
+        "url_optional_data_source_metadata"
+    )
 
 def _add_last_synced_at_columns():
     op.add_column(

src/core/tasks/scheduled/impl/sync_to_ds/impl/agencies/add/queries/add_links.py

@@ -0,0 +1,17 @@
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.db.queries.base.builder import QueryBuilderBase
+from src.external.pdap.impl.sync.shared.models.mapping import DSSyncIDMapping
+
+
+class DSAppSyncAgenciesAddInsertLinksQueryBuilder(QueryBuilderBase):
+
+    def __init__(
+        self,
+        mappings: list[DSSyncIDMapping]
+    ):
+        super().__init__()
+        self._mappings = mappings
+
+    async def run(self, session: AsyncSession) -> None:
+        raise NotImplementedError

src/core/tasks/scheduled/impl/sync_to_ds/impl/agencies/add/queries/cte.py

@@ -1,4 +1,32 @@
 """
 Agencies to be added to the DS database must not have a
 ds app link entry
-"""
+"""
+from sqlalchemy import Column, select, exists, CTE
+
+from src.db.models.impl.agency.ds_link.sqlalchemy import DSAppLinkAgency
+from src.db.models.impl.agency.sqlalchemy import Agency
+
+
+class DSAppLinkSyncAgencyAddPrerequisitesCTEContainer:
+
+    def __init__(self):
+        self._cte = (
+            select(
+                Agency.id
+            )
+            .where(
+                ~exists(
+                    select(DSAppLinkAgency.agency_id)
+                    .where(DSAppLinkAgency.agency_id == Agency.id)
+                )
+            ).cte()
+        )
+
+    @property
+    def agency_id(self) -> Column[int]:
+        return self._cte.columns.id
+
+    @property
+    def cte(self) -> CTE:
+        return self._cte

src/core/tasks/scheduled/impl/sync_to_ds/impl/agencies/add/queries/get.py

@@ -0,0 +1,10 @@
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.db.queries.base.builder import QueryBuilderBase
+from src.external.pdap.impl.sync.agencies.add.request import AddAgenciesOuterRequest
+
+
+class DSAppSyncAgenciesAddGetQueryBuilder(QueryBuilderBase):
+
+    async def run(self, session: AsyncSession) -> AddAgenciesOuterRequest:
+        raise NotImplementedError

src/core/tasks/scheduled/impl/sync_to_ds/impl/agencies/add/queries/prereq.py

@@ -0,0 +1,9 @@
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.db.queries.base.builder import QueryBuilderBase
+
+
+class DSAppSyncAgenciesAddPrerequisitesQueryBuilder(QueryBuilderBase):
+
+    async def run(self, session: AsyncSession) -> bool:
+        raise NotImplementedError

src/core/tasks/scheduled/impl/sync_to_ds/impl/agencies/delete/queries/cte.py

@@ -1,3 +1,29 @@
 """
 Agencies to be deleted from the DS database must be flagged for deletion
-"""
+"""
+from sqlalchemy import select, Column, CTE
+
+from src.db.models.impl.agency.ds_link.sqlalchemy import DSAppLinkAgency
+from src.db.models.impl.flag.ds_delete.agency import FlagDSDeleteAgency
+
+
+class DSAppLinkSyncAgencyDeletePrerequisitesCTEContainer:
+
+    def __init__(self):
+        self._cte = (
+            select(
+                DSAppLinkAgency.ds_agency_id
+            )
+            .join(
+                FlagDSDeleteAgency,
+                FlagDSDeleteAgency.ds_agency_id == DSAppLinkAgency.ds_agency_id
+            ).cte()
+        )
+
+    @property
+    def ds_agency_id(self) -> Column[int]:
+        return self._cte.columns.ds_agency_id
+
+    @property
+    def cte(self) -> CTE:
+        return self._cte

src/core/tasks/scheduled/impl/sync_to_ds/impl/agencies/delete/queries/delete_flags.py

@@ -0,0 +1,16 @@
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.db.queries.base.builder import QueryBuilderBase
+
+
+class DSAppSyncAgenciesDeleteRemoveFlagsQueryBuilder(QueryBuilderBase):
+
+    def __init__(
+        self,
+        ds_agency_ids: list[int]
+    ):
+        super().__init__()
+        self._ds_agency_ids = ds_agency_ids
+
+    async def run(self, session: AsyncSession) -> None:
+        raise NotImplementedError

src/core/tasks/scheduled/impl/sync_to_ds/impl/agencies/delete/queries/delete_links.py

@@ -0,0 +1,16 @@
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.db.queries.base.builder import QueryBuilderBase
+
+
+class DSAppSyncAgenciesDeleteRemoveLinksQueryBuilder(QueryBuilderBase):
+
+    def __init__(
+        self,
+        ds_agency_ids: list[int]
+    ):
+        super().__init__()
+        self._ds_agency_ids = ds_agency_ids
+
+    async def run(self, session: AsyncSession) -> None:
+        raise NotImplementedError

src/core/tasks/scheduled/impl/sync_to_ds/impl/agencies/delete/queries/get.py

@@ -0,0 +1,10 @@
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.db.queries.base.builder import QueryBuilderBase
+
+
+class DSAppSyncAgenciesDeleteGetQueryBuilder(QueryBuilderBase):
+
+    async def run(self, session: AsyncSession) -> list[int]:
+        """Get DS App links to delete."""
+        raise NotImplementedError