Revamped destination and transfer pages

api/Dockerfile

@@ -20,17 +20,17 @@ RUN apt-get update \
     && apt-get -y install libpq-dev gcc \
     && pip install psycopg2
 
+# Copy and install the transfer library wheel
+COPY --from=transfer-builder /transfer-build/dist/*.whl /api/dist/
+RUN pip install --no-cache-dir /api/dist/pontoon-*.whl
+
 # Build the API app
 COPY ./api/pyproject.toml /api/pyproject.toml
 COPY ./api/app /api/app
 COPY ./api/db /api/db
 RUN pip install build && python -m build --wheel
 
-# Copy the transfer library wheel into dist
-COPY --from=transfer-builder /transfer-build/dist/*.whl /api/dist/
-
 RUN pip install --no-cache-dir /api/dist/app-*.whl
-RUN pip install --no-cache-dir /api/dist/pontoon-*.whl
 RUN pip install SQLAlchemy==2.0.41
 
 EXPOSE 8000

api/app/models/transfer_run.py

@@ -70,15 +70,34 @@ def get_latest_transfer_run(session, transfer_id:uuid.UUID, status:str = None) -
 
 
     @staticmethod
-    def list(session, destination_id:uuid.UUID, offset:int, limit:int) -> List[Model]: 
+    def list(session, offset:int, limit:int, destination_id:uuid.UUID, execution_id:uuid.UUID = None) -> List[Model]: 
+        """
+        List TransferRuns with optional filtering by destination_id and/or execution_id.
+
+        Args:
+            session: Database session
+            offset: Number of records to skip for pagination
+            limit: Maximum number of records to return
+            destination_id: Filter by destination ID
+            execution_id: Optional filter by execution_id from meta field
+
+        Returns:
+            List[Model]: List of TransferRun models matching the filters
+        """
         stmt = (
             select(TransferRun.Model)
             .join(Transfer.Model, TransferRun.Model.transfer_id == Transfer.Model.transfer_id)
             .where(Transfer.Model.destination_id == destination_id)
-            .order_by(TransferRun.Model.created_at.desc())
-            .offset(offset)
-            .limit(limit)
         )
+
+        # Apply execution_id filter if provided
+        if execution_id is not None:
+            stmt = stmt.where(
+                (TransferRun.Model.meta.is_not(None)) &
+                (TransferRun.Model.meta['execution_id'].as_string() == str(execution_id))
+            )
+
+        stmt = stmt.order_by(TransferRun.Model.created_at.desc()).offset(offset).limit(limit)
         return session.exec(stmt).all()
 
 

api/app/routers/destinations.py

@@ -1,8 +1,9 @@
 import uuid
 import time
-from typing import Annotated, List
+from typing import Annotated, List, Optional, Literal
 from datetime import datetime, timezone
 from fastapi import HTTPException, Depends, Query, APIRouter, Security, status
+from pydantic import BaseModel
 
 from app.models import Auth, Task, Destination, ScheduleModel, TransferRun, Transfer as TransferModel
 from app.routers.common import create_transfer_task, transfer_task_status
@@ -29,6 +30,11 @@ def get_destination_by_id(session, destination_id:uuid.UUID):
     return Destination.get(session, destination_id)
 
 
+class TransferScheduleOverride(BaseModel):
+    type: Literal["FULL_REFRESH", "INCREMENTAL"]
+    start: Optional[datetime] = None
+    end: Optional[datetime] = None
+
 #
 # Transfer() management helpers
 #
@@ -229,7 +235,12 @@ def create_destination_check(destination_id:uuid.UUID, session=Depends(get_sessi
 
 
 @router.post("/{destination_id}/run")
-def run_destination_transfer(destination_id:uuid.UUID, session=Depends(get_session), auth:Auth = Security(get_auth)):
+def run_destination_transfer(
+    destination_id: uuid.UUID, 
+    schedule_override: Optional[TransferScheduleOverride] = None,
+    session=Depends(get_session), 
+    auth:Auth = Security(get_auth)
+):
     try:
         if settings.skip_transfers:
             return {"ok": True}
@@ -248,17 +259,32 @@ def run_destination_transfer(destination_id:uuid.UUID, session=Depends(get_sessi
         run.set_organization(auth.org_uuid())
         run.set_destination(str(destination_id)) 
 
-        if schedule_model.type != 'FULL_REFRESH':
-            run.set_mode(Mode({
-                'type': schedule_model.type, 
-                'period': schedule_model.frequency, 
-                'start': now - Mode.delta(schedule_model.frequency), 
-                'end': now
-            }))
+        # Determine the mode based on request body or fall back to schedule defaults
+        if schedule_override:
+            if schedule_override.type == 'FULL_REFRESH':
+                run.set_mode(Mode({'type': 'FULL_REFRESH'}))
+            else:
+                end_time = schedule_override.end if schedule_override.end else now
+                start_time = schedule_override.start if schedule_override.start else (end_time - Mode.delta(schedule_model.frequency))
+                run.set_mode(Mode({
+                    'type': 'INCREMENTAL',
+                    'period': schedule_model.frequency,
+                    'start': start_time,
+                    'end': end_time
+                }))
+        else:
+            # Use schedule defaults
+            if schedule_model.type != 'FULL_REFRESH':
+                run.set_mode(Mode({
+                    'type': schedule_model.type, 
+                    'period': schedule_model.frequency, 
+                    'start': now - Mode.delta(schedule_model.frequency), 
+                    'end': now
+                }))
 
         run.run(expedited=False)
 
-    except (TransferException):
+    except TransferException as e:
         raise HTTPException(status_code=400, detail=str(e))
 
 

api/app/routers/transfers.py

@@ -27,11 +27,12 @@
 @router.get("", response_model=list[TransferRun.Model])
 def search_transfers(
         destination_id: uuid.UUID,
+        execution_id: uuid.UUID = None,
         session = Depends(get_session),
         offset: int = 0,
         limit: Annotated[int, Query(le=100)] = 100,
 ):
-    return TransferRun.list(session, destination_id, offset, limit)
+    return TransferRun.list(session, offset=offset, limit=limit, destination_id=destination_id, execution_id=execution_id)
 
 
 @router.get("/{transfer_run_id}", response_model=TransferRun.Model)
@@ -44,6 +45,16 @@ def get_transfer_run(
         raise HTTPException(status_code=404, detail="Transfer run not found")
     return transfer_run
 
+@router.get("/transfer/{transfer_id}", response_model=TransferModel.Model)
+def get_transfer(
+    transfer_id: uuid.UUID,
+    session = Depends(get_session),
+):
+    transfer = TransferModel.get(session, transfer_id)
+    if not transfer:
+        raise HTTPException(status_code=404, detail="Transfer not found")
+    return transfer
+
 
 @router.post("/{transfer_run_id}/rerun")
 def rerun_transfer(
@@ -79,6 +90,9 @@ def rerun_transfer(
             raise TransferRun.Exception("Transfer run does not have arguments; can't re-run")
 
         args = meta.get('arguments', {})
+        execution_id = meta.get('execution_id', None)
+        if execution_id is None:
+            raise TransferRun.Exception("Transfer run does not have an execution ID; can't re-run")
 
         # kick off a new transfer
         if settings.skip_transfers != True:
@@ -88,6 +102,7 @@ def rerun_transfer(
             new_run.set_destination(str(destination.destination_id))
             new_run.set_mode(Mode(args.get('mode', {})))
             new_run.set_models(args.get('models', []))
+            new_run.set_execution_id(execution_id)
             new_run.run(expedited=False)
 
         return {"ok": True}

data-transfer/pontoon/pontoon/celery/tasks.py

@@ -15,7 +15,9 @@ def transfer_task(self, args_json: str):
     print("Running transfer job as celery task...")
     try:
         args = json.loads(args_json).get('commandArgs', [])
-        args += ['--execution-id', str(self.request.id)]
+        if '--execution-id' not in args:
+            # Generate a new execution ID if one is not provided
+            args += ['--execution-id', str(self.request.id)]
         args += ['--retry-count', str(self.request.retries)]
         args += ['--retry-limit', str(TASK_MAX_RETRIES)]
 

data-transfer/pontoon/pontoon/orchestration/client.py

@@ -369,6 +369,10 @@ def set_models(self, model_ids:List[str]) -> 'Transfer':
     def set_organization(self, organization_id:str) -> 'Transfer':
         # set the org id for this transfer
         return self.set_argument('--organization-id', organization_id)
+
+    def set_execution_id(self, execution_id:str) -> 'Transfer':
+        # set the execution id for this transfer
+        return self.set_argument('--execution-id', execution_id)
 
 
     def set_argument(self, arg_name:str, arg_value) -> 'Transfer':

data-transfer/pontoon/pontoon/orchestration/transfer.py

@@ -219,7 +219,9 @@ def _schedule_to_replication_mode(self, schedule):
         schedule_hour = schedule.get('hour', 0)
         schedule_min = schedule.get('minute', 0)
 
-        end_time = self._now.replace(hour=schedule_hour, minute=schedule_min, microsecond=0)
+        end_time = self._now.replace(minute=schedule_min, second=0, microsecond=0)
+        if period != Mode.HOURLY:
+            end_time = end_time.replace(hour=schedule_hour)
 
         if period == Mode.WEEKLY:
             begin_time = end_time - timedelta(days=7, hours=12)

docker-compose.yml

@@ -94,6 +94,7 @@ services:
       - app-network
     depends_on:
       - redis
+    restart: unless-stopped
 
 volumes:
   postgres-data:

test-env/mock_data_generator.py

@@ -191,11 +191,12 @@ def generate_and_insert_mock_data():
     conn = get_db_connection()
     try:
         logger.info("Starting mock data generation...")
+        num_records_to_generate = 50
 
         # Generate mock data
-        leads = generate_mock_leads(100)
-        campaigns = generate_mock_campaigns(100)
-        attributions = generate_mock_attribution(100)
+        leads = generate_mock_leads(num_records_to_generate)
+        campaigns = generate_mock_campaigns(num_records_to_generate)
+        attributions = generate_mock_attribution(num_records_to_generate)
 
         # Insert data
         insert_leads(conn, leads)
@@ -217,8 +218,8 @@ def main():
     while True:
         try:
             generate_and_insert_mock_data()
-            logger.info("Waiting 5 minutes before next generation...")
-            time.sleep(300)  # 5 minutes
+            logger.info("Waiting 1 minute before next generation...")
+            time.sleep(60)  # 1 minute
         except KeyboardInterrupt:
             logger.info("Mock data generator stopped by user")
             break