mc_352_add_feature_flags

ENV.md

@@ -22,10 +22,36 @@ Please ensure these are properly defined in a `.env` file in the root directory.
 | `DISCORD_WEBHOOK_URL`           | The URL for the Discord webhook used for notifications                                                                                                                        | `abc123`                                                                                     |
 | `HUGGINGFACE_INFERENCE_API_KEY` | The API key required for accessing the Hugging Face Inference API.                                                                                                            | `abc123`                                                                                     |
 | `HUGGINGFACE_HUB_TOKEN`         | The API key required for uploading to the PDAP HuggingFace account via Hugging Face Hub API.                                                                                                                                                                     | `abc123`  |
-| `SCHEDULED_TASKS_FLAG`          | Set to `1` to enable running scheduled tasks.                                                                                                                                 | `1`                                                                                          |
+
+
 
 [^1:] The user account in question will require elevated permissions to access certain endpoints. At a minimum, the user will require the `source_collector` and `db_write` permissions.
 
+## Task Flags
+Task flags are used to enable/disable certain tasks. They are set to `1` to enable the task and `0` to disable the task. By default, all tasks are enabled.
+
+The following flags are available:
+
+| Flag                                  | Description                                           |
+|---------------------------------------|-------------------------------------------------------|
+| `SCHEDULED_TASKS_FLAG`                | All scheduled tasks.                                  |
+| `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_DUPLICATE_TASK_FLAG`             | Identifies duplicate URLs.                            |
+| `URL_MISC_METADATA_TASK_FLAG`         | Adds misc metadata to URLs.                           |
+| `URL_404_PROBE_TASK_FLAG`             | Probes URLs for 404 errors.                           |
+| `URL_AUTO_RELEVANCE_TASK_FLAG`        | Automatically assigns Relevances to URLs.             |
+| `URL_PROBE_TASK_FLAG`                 | Probes URLs for web metadata.                         |
+| `SYNC_AGENCIES_TASK_FLAG`              | Synchonize agencies from Data Sources App.            |
+| `SYNC_DATA_SOURCES_TASK_FLAG`          | Synchonize data sources from Data Sources App.        |
+| `PUSH_TO_HUGGING_FACE_TASK_FLAG`        | Pushes data to HuggingFace.                          |
+| `POPULATE_BACKLOG_SNAPSHOT_TASK_FLAG`  | Populates the backlog snapshot.                       |
+| `DELETE_OLD_LOGS_TASK_FLAG`            | Deletes old logs.                                     |
+| `RUN_URL_TASKS_TASK_FLAG`              | Runs URL tasks.                                       |
+
+
 ## Foreign Data Wrapper (FDW)
 ```
 FDW_DATA_SOURCES_HOST=127.0.0.1  # The host of the Data Sources Database, used for FDW setup

alembic/versions/2025_08_10_1032-11ece61d7ac2_add_scheduled_tasks.py

@@ -0,0 +1,63 @@
+"""Add scheduled tasks
+
+Revision ID: 11ece61d7ac2
+Revises: 8cd5aa7670ff
+Create Date: 2025-08-10 10:32:11.400714
+
+"""
+from typing import Sequence, Union
+
+from src.util.alembic_helpers import switch_enum_type
+
+# revision identifiers, used by Alembic.
+revision: str = '11ece61d7ac2'
+down_revision: Union[str, None] = '8cd5aa7670ff'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    switch_enum_type(
+        table_name='tasks',
+        column_name='task_type',
+        enum_name='task_type',
+        new_enum_values=[
+            'HTML',
+            'Relevancy',
+            'Record Type',
+            'Agency Identification',
+            'Misc Metadata',
+            'Submit Approved URLs',
+            'Duplicate Detection',
+            '404 Probe',
+            'Sync Agencies',
+            'Sync Data Sources',
+            'Push to Hugging Face',
+            'URL Probe',
+            'Populate Backlog Snapshot',
+            'Delete Old Logs',
+            'Run URL Task Cycles'
+        ]
+    )
+
+
+def downgrade() -> None:
+    switch_enum_type(
+        table_name='tasks',
+        column_name='task_type',
+        enum_name='task_type',
+        new_enum_values=[
+            'HTML',
+            'Relevancy',
+            'Record Type',
+            'Agency Identification',
+            'Misc Metadata',
+            'Submit Approved URLs',
+            'Duplicate Detection',
+            '404 Probe',
+            'Sync Agencies',
+            'Sync Data Sources'
+            'Push to Hugging Face',
+            'URL Probe'
+        ]
+    )

src/api/main.py

@@ -24,6 +24,7 @@
 from src.core.tasks.handler import TaskHandler
 from src.core.tasks.scheduled.loader import ScheduledTaskOperatorLoader
 from src.core.tasks.scheduled.manager import AsyncScheduledTaskManager
+from src.core.tasks.scheduled.registry.core import ScheduledJobRegistry
 from src.core.tasks.url.loader import URLTaskOperatorLoader
 from src.core.tasks.url.manager import TaskManager
 from src.core.tasks.url.operators.html.scraper.parser.core import HTMLResponseParser
@@ -98,15 +99,16 @@ async def lifespan(app: FastAPI):
         collector_manager=async_collector_manager
     )
     async_scheduled_task_manager = AsyncScheduledTaskManager(
-        async_core=async_core,
         handler=task_handler,
         loader=ScheduledTaskOperatorLoader(
             adb_client=adb_client,
             pdap_client=pdap_client,
             hf_client=HuggingFaceHubClient(
                 token=env_var_manager.hf_hub_token
-            )
-        )
+            ),
+            async_core=async_core,
+        ),
+        registry=ScheduledJobRegistry()
     )
     await async_scheduled_task_manager.setup()
 

src/core/core.py

@@ -48,6 +48,9 @@
 
 
 class AsyncCore:
+    task_manager: TaskManager | None = None
+    adb_client: AsyncDatabaseClient | None = None
+    collector_manager: AsyncCollectorManager | None = None
 
     def __init__(
             self,
@@ -57,7 +60,6 @@ def __init__(
     ):
         self.task_manager = task_manager
         self.adb_client = adb_client
-
         self.collector_manager = collector_manager
 
 

src/core/tasks/scheduled/impl/backlog/operator.py

@@ -0,0 +1,16 @@
+from src.core.tasks.scheduled.templates.operator import ScheduledTaskOperatorBase
+from src.db.client.async_ import AsyncDatabaseClient
+from src.db.enums import TaskType
+
+
+class PopulateBacklogSnapshotTaskOperator(ScheduledTaskOperatorBase):
+
+    def __init__(self, adb_client: AsyncDatabaseClient):
+        super().__init__(adb_client)
+
+    @property
+    def task_type(self) -> TaskType:
+        return TaskType.POPULATE_BACKLOG_SNAPSHOT
+
+    async def inner_task_logic(self) -> None:
+        await self.adb_client.populate_backlog_snapshot()

src/core/tasks/scheduled/impl/delete_logs/operator.py

@@ -0,0 +1,16 @@
+from src.core.tasks.scheduled.templates.operator import ScheduledTaskOperatorBase
+from src.db.client.async_ import AsyncDatabaseClient
+from src.db.enums import TaskType
+
+
+class DeleteOldLogsTaskOperator(ScheduledTaskOperatorBase):
+
+    def __init__(self, adb_client: AsyncDatabaseClient):
+        super().__init__(adb_client)
+
+    @property
+    def task_type(self) -> TaskType:
+        return TaskType.DELETE_OLD_LOGS
+
+    async def inner_task_logic(self) -> None:
+        await self.adb_client.delete_old_logs()

src/core/tasks/scheduled/huggingface/queries/check/core.py → src/core/tasks/scheduled/impl/huggingface/queries/check/core.py

@@ -1,6 +1,6 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from src.core.tasks.scheduled.huggingface.queries.check.requester import CheckValidURLsUpdatedRequester
+from src.core.tasks.scheduled.impl.huggingface.queries.check.requester import CheckValidURLsUpdatedRequester
 from src.db.queries.base.builder import QueryBuilderBase
 
 

src/core/tasks/scheduled/huggingface/queries/get/convert.py → src/core/tasks/scheduled/impl/huggingface/queries/get/convert.py

@@ -1,7 +1,7 @@
 from src.collectors.enums import URLStatus
 from src.core.enums import RecordType
-from src.core.tasks.scheduled.huggingface.queries.get.enums import RecordTypeCoarse
-from src.core.tasks.scheduled.huggingface.queries.get.mappings import FINE_COARSE_RECORD_TYPE_MAPPING, \
+from src.core.tasks.scheduled.impl.huggingface.queries.get.enums import RecordTypeCoarse
+from src.core.tasks.scheduled.impl.huggingface.queries.get.mappings import FINE_COARSE_RECORD_TYPE_MAPPING, \
     OUTCOME_RELEVANCY_MAPPING
 
 

src/core/tasks/scheduled/huggingface/queries/get/core.py → src/core/tasks/scheduled/impl/huggingface/queries/get/core.py

@@ -2,9 +2,9 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from src.collectors.enums import URLStatus
-from src.core.tasks.scheduled.huggingface.queries.get.convert import convert_url_status_to_relevant, \
+from src.core.tasks.scheduled.impl.huggingface.queries.get.convert import convert_url_status_to_relevant, \
     convert_fine_to_coarse_record_type
-from src.core.tasks.scheduled.huggingface.queries.get.model import GetForLoadingToHuggingFaceOutput
+from src.core.tasks.scheduled.impl.huggingface.queries.get.model import GetForLoadingToHuggingFaceOutput
 from src.db.models.instantiations.url.html.compressed.sqlalchemy import URLCompressedHTML
 from src.db.models.instantiations.url.core.sqlalchemy import URL
 from src.db.queries.base.builder import QueryBuilderBase

src/core/tasks/scheduled/huggingface/queries/get/mappings.py → src/core/tasks/scheduled/impl/huggingface/queries/get/mappings.py

@@ -1,6 +1,6 @@
 from src.collectors.enums import URLStatus
 from src.core.enums import RecordType
-from src.core.tasks.scheduled.huggingface.queries.get.enums import RecordTypeCoarse
+from src.core.tasks.scheduled.impl.huggingface.queries.get.enums import RecordTypeCoarse
 
 FINE_COARSE_RECORD_TYPE_MAPPING = {
     # Police and Public

src/core/tasks/scheduled/huggingface/queries/get/model.py → src/core/tasks/scheduled/impl/huggingface/queries/get/model.py

@@ -1,7 +1,7 @@
 from pydantic import BaseModel
 
 from src.core.enums import RecordType
-from src.core.tasks.scheduled.huggingface.queries.get.enums import RecordTypeCoarse
+from src.core.tasks.scheduled.impl.huggingface.queries.get.enums import RecordTypeCoarse
 
 
 class GetForLoadingToHuggingFaceOutput(BaseModel):

src/core/tasks/scheduled/impl/run_url_tasks/operator.py

@@ -0,0 +1,17 @@
+from src.core.core import AsyncCore
+from src.core.tasks.scheduled.templates.operator import ScheduledTaskOperatorBase
+from src.db.enums import TaskType
+
+
+class RunURLTasksTaskOperator(ScheduledTaskOperatorBase):
+
+    def __init__(self, async_core: AsyncCore):
+        super().__init__(async_core.adb_client)
+        self.async_core = async_core
+
+    @property
+    def task_type(self) -> TaskType:
+        return TaskType.RUN_URL_TASKS
+
+    async def inner_task_logic(self) -> None:
+        await self.async_core.run_tasks()

src/core/tasks/scheduled/sync/agency/operator.py → src/core/tasks/scheduled/impl/sync/agency/operator.py

@@ -1,5 +1,5 @@
-from src.core.tasks.scheduled.sync.check import check_max_sync_requests_not_exceeded
-from src.core.tasks.scheduled.sync.agency.dtos.parameters import AgencySyncParameters
+from src.core.tasks.scheduled.impl.sync.check import check_max_sync_requests_not_exceeded
+from src.core.tasks.scheduled.impl.sync.agency.dtos.parameters import AgencySyncParameters
 from src.core.tasks.scheduled.templates.operator import ScheduledTaskOperatorBase
 from src.db.client.async_ import AsyncDatabaseClient
 from src.db.enums import TaskType

src/core/tasks/scheduled/sync/agency/queries/get_sync_params.py → src/core/tasks/scheduled/impl/sync/agency/queries/get_sync_params.py

@@ -2,7 +2,7 @@
 from sqlalchemy.exc import NoResultFound
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from src.core.tasks.scheduled.sync.agency.dtos.parameters import AgencySyncParameters
+from src.core.tasks.scheduled.impl.sync.agency.dtos.parameters import AgencySyncParameters
 from src.db.models.instantiations.state.sync.agencies import AgenciesSyncState
 from src.db.queries.base.builder import QueryBuilderBase
 

src/core/tasks/scheduled/sync/check.py → src/core/tasks/scheduled/impl/sync/check.py

@@ -1,5 +1,5 @@
-from src.core.tasks.scheduled.sync.constants import MAX_SYNC_REQUESTS
-from src.core.tasks.scheduled.sync.exceptions import MaxRequestsExceededError
+from src.core.tasks.scheduled.impl.sync.constants import MAX_SYNC_REQUESTS
+from src.core.tasks.scheduled.impl.sync.exceptions import MaxRequestsExceededError
 
 
 def check_max_sync_requests_not_exceeded(request_count: int) -> None:

src/core/tasks/scheduled/sync/data_sources/operator.py → src/core/tasks/scheduled/impl/sync/data_sources/operator.py

@@ -1,6 +1,6 @@
 from src.core.tasks.scheduled.templates.operator import ScheduledTaskOperatorBase
-from src.core.tasks.scheduled.sync.check import check_max_sync_requests_not_exceeded
-from src.core.tasks.scheduled.sync.data_sources.params import DataSourcesSyncParameters
+from src.core.tasks.scheduled.impl.sync.check import check_max_sync_requests_not_exceeded
+from src.core.tasks.scheduled.impl.sync.data_sources.params import DataSourcesSyncParameters
 from src.db.client.async_ import AsyncDatabaseClient
 from src.db.enums import TaskType
 from src.external.pdap.client import PDAPClient

src/core/tasks/scheduled/sync/data_sources/queries/get_sync_params.py → src/core/tasks/scheduled/impl/sync/data_sources/queries/get_sync_params.py

@@ -2,7 +2,7 @@
 from sqlalchemy.exc import NoResultFound
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from src.core.tasks.scheduled.sync.data_sources.params import DataSourcesSyncParameters
+from src.core.tasks.scheduled.impl.sync.data_sources.params import DataSourcesSyncParameters
 from src.db.models.instantiations.state.sync.data_sources import DataSourcesSyncState
 from src.db.queries.base.builder import QueryBuilderBase
 

src/core/tasks/scheduled/sync/data_sources/queries/upsert/agency/core.py → src/core/tasks/scheduled/impl/sync/data_sources/queries/upsert/agency/core.py

@@ -1,7 +1,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from src.core.tasks.scheduled.sync.data_sources.queries.upsert.agency.query import URLAgencyLinkUpdateQueryBuilder
-from src.core.tasks.scheduled.sync.data_sources.queries.upsert.agency.params import UpdateLinkURLAgencyForDataSourcesSyncParams
+from src.core.tasks.scheduled.impl.sync.data_sources.queries.upsert.agency.query import URLAgencyLinkUpdateQueryBuilder
+from src.core.tasks.scheduled.impl.sync.data_sources.queries.upsert.agency.params import UpdateLinkURLAgencyForDataSourcesSyncParams
 
 
 async def update_agency_links(

src/core/tasks/scheduled/sync/data_sources/queries/upsert/agency/query.py → src/core/tasks/scheduled/impl/sync/data_sources/queries/upsert/agency/query.py

@@ -3,10 +3,10 @@
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from src.core.tasks.scheduled.sync.data_sources.queries.upsert.agency.convert import convert_to_link_url_agency_models
+from src.core.tasks.scheduled.impl.sync.data_sources.queries.upsert.agency.convert import convert_to_link_url_agency_models
 from src.db.helpers.session import session_helper as sh
 from src.db.models.instantiations.link.url_agency.pydantic import LinkURLAgencyPydantic
-from src.core.tasks.scheduled.sync.data_sources.queries.upsert.agency.params import UpdateLinkURLAgencyForDataSourcesSyncParams
+from src.core.tasks.scheduled.impl.sync.data_sources.queries.upsert.agency.params import UpdateLinkURLAgencyForDataSourcesSyncParams
 from src.db.models.instantiations.link.url_agency.sqlalchemy import LinkURLAgency
 from src.db.queries.base.builder import QueryBuilderBase
 

src/core/tasks/scheduled/sync/data_sources/queries/upsert/core.py → src/core/tasks/scheduled/impl/sync/data_sources/queries/upsert/core.py

@@ -3,13 +3,13 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 from typing_extensions import override
 
-from src.core.tasks.scheduled.sync.data_sources.queries.upsert.helpers.filter import filter_for_urls_with_ids, \
+from src.core.tasks.scheduled.impl.sync.data_sources.queries.upsert.helpers.filter import filter_for_urls_with_ids, \
     get_mappings_for_urls_without_data_sources
-from src.core.tasks.scheduled.sync.data_sources.queries.upsert.mapper import URLSyncInfoMapper
-from src.core.tasks.scheduled.sync.data_sources.queries.upsert.param_manager import \
+from src.core.tasks.scheduled.impl.sync.data_sources.queries.upsert.mapper import URLSyncInfoMapper
+from src.core.tasks.scheduled.impl.sync.data_sources.queries.upsert.param_manager import \
     UpsertURLsFromDataSourcesParamManager
-from src.core.tasks.scheduled.sync.data_sources.queries.upsert.requester import UpsertURLsFromDataSourcesDBRequester
-from src.core.tasks.scheduled.sync.data_sources.queries.upsert.url.lookup.response import \
+from src.core.tasks.scheduled.impl.sync.data_sources.queries.upsert.requester import UpsertURLsFromDataSourcesDBRequester
+from src.core.tasks.scheduled.impl.sync.data_sources.queries.upsert.url.lookup.response import \
     LookupURLForDataSourcesSyncResponse
 from src.db.dtos.url.mapping import URLMapping
 from src.db.queries.base.builder import QueryBuilderBase

src/core/tasks/scheduled/sync/data_sources/queries/upsert/helpers/convert.py → src/core/tasks/scheduled/impl/sync/data_sources/queries/upsert/helpers/convert.py

@@ -1,7 +1,7 @@
 from src.collectors.enums import URLStatus
-from src.core.tasks.scheduled.sync.data_sources.queries.upsert.url.insert.params import \
+from src.core.tasks.scheduled.impl.sync.data_sources.queries.upsert.url.insert.params import \
     InsertURLForDataSourcesSyncParams
-from src.core.tasks.scheduled.sync.data_sources.queries.upsert.url.update.params import \
+from src.core.tasks.scheduled.impl.sync.data_sources.queries.upsert.url.update.params import \
     UpdateURLForDataSourcesSyncParams
 from src.external.pdap.dtos.sync.data_sources import DataSourcesSyncResponseInnerInfo
 from src.external.pdap.enums import DataSourcesURLStatus, ApprovalStatus

src/core/tasks/scheduled/sync/data_sources/queries/upsert/helpers/filter.py → src/core/tasks/scheduled/impl/sync/data_sources/queries/upsert/helpers/filter.py

@@ -1,4 +1,4 @@
-from src.core.tasks.scheduled.sync.data_sources.queries.upsert.url.lookup.response import \
+from src.core.tasks.scheduled.impl.sync.data_sources.queries.upsert.url.lookup.response import \
     LookupURLForDataSourcesSyncResponse
 from src.db.dtos.url.mapping import URLMapping
 

src/core/tasks/scheduled/sync/data_sources/queries/upsert/param_manager.py → src/core/tasks/scheduled/impl/sync/data_sources/queries/upsert/param_manager.py

@@ -1,13 +1,13 @@
-from src.core.tasks.scheduled.sync.data_sources.queries.upsert.agency.params import \
+from src.core.tasks.scheduled.impl.sync.data_sources.queries.upsert.agency.params import \
     UpdateLinkURLAgencyForDataSourcesSyncParams
-from src.core.tasks.scheduled.sync.data_sources.queries.upsert.helpers.convert import convert_to_url_update_params, \
+from src.core.tasks.scheduled.impl.sync.data_sources.queries.upsert.helpers.convert import convert_to_url_update_params, \
     convert_to_url_insert_params
-from src.core.tasks.scheduled.sync.data_sources.queries.upsert.mapper import URLSyncInfoMapper
-from src.core.tasks.scheduled.sync.data_sources.queries.upsert.url.insert.params import \
+from src.core.tasks.scheduled.impl.sync.data_sources.queries.upsert.mapper import URLSyncInfoMapper
+from src.core.tasks.scheduled.impl.sync.data_sources.queries.upsert.url.insert.params import \
     InsertURLForDataSourcesSyncParams
-from src.core.tasks.scheduled.sync.data_sources.queries.upsert.url.lookup.response import \
+from src.core.tasks.scheduled.impl.sync.data_sources.queries.upsert.url.lookup.response import \
     LookupURLForDataSourcesSyncResponse
-from src.core.tasks.scheduled.sync.data_sources.queries.upsert.url.update.params import \
+from src.core.tasks.scheduled.impl.sync.data_sources.queries.upsert.url.update.params import \
     UpdateURLForDataSourcesSyncParams
 from src.db.dtos.url.mapping import URLMapping
 from src.db.models.instantiations.link.url_agency.pydantic import LinkURLAgencyPydantic

src/core/tasks/scheduled/sync/data_sources/queries/upsert/requester.py → src/core/tasks/scheduled/impl/sync/data_sources/queries/upsert/requester.py

@@ -1,16 +1,16 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from src.core.tasks.scheduled.sync.data_sources.queries.upsert.agency.params import \
+from src.core.tasks.scheduled.impl.sync.data_sources.queries.upsert.agency.params import \
     UpdateLinkURLAgencyForDataSourcesSyncParams
-from src.core.tasks.scheduled.sync.data_sources.queries.upsert.agency.query import \
+from src.core.tasks.scheduled.impl.sync.data_sources.queries.upsert.agency.query import \
     URLAgencyLinkUpdateQueryBuilder
-from src.core.tasks.scheduled.sync.data_sources.queries.upsert.url.insert.params import \
+from src.core.tasks.scheduled.impl.sync.data_sources.queries.upsert.url.insert.params import \
     InsertURLForDataSourcesSyncParams
-from src.core.tasks.scheduled.sync.data_sources.queries.upsert.url.lookup.query import \
+from src.core.tasks.scheduled.impl.sync.data_sources.queries.upsert.url.lookup.query import \
     LookupURLForDataSourcesSyncQueryBuilder
-from src.core.tasks.scheduled.sync.data_sources.queries.upsert.url.lookup.response import \
+from src.core.tasks.scheduled.impl.sync.data_sources.queries.upsert.url.lookup.response import \
     LookupURLForDataSourcesSyncResponse
-from src.core.tasks.scheduled.sync.data_sources.queries.upsert.url.update.params import \
+from src.core.tasks.scheduled.impl.sync.data_sources.queries.upsert.url.update.params import \
     UpdateURLForDataSourcesSyncParams
 from src.db.dtos.url.mapping import URLMapping
 from src.db.helpers.session import session_helper as sh