Arrow IPC Streaming Cache Implementation (#35)

* Add arrow ipc streaming cache implementation and make it default

* fix stream unit test to reflect last_synced_at  bug fix

* small build and  stability fixes

Author: kmacrow

api/Dockerfile

@@ -22,7 +22,7 @@ RUN apt-get update \
 
 # 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
+RUN pip install --only-binary=:all: --no-cache-dir /api/dist/pontoon-*.whl
 
 # Build the API app
 COPY ./api/pyproject.toml /api/pyproject.toml

api/app/models/transfer_run.py

@@ -66,7 +66,11 @@ def get_latest_transfer_run(session, transfer_id:uuid.UUID, status:str = None) -
         if status != None:
             stmt = stmt.where(TransferRun.Model.status == status)
 
-        return session.exec(stmt).first()
+        result = session.exec(stmt).all()
+        if len(result) == 1:
+            return result[0]
+        else:
+            return None
 
 
     @staticmethod

api/app/routers/internal.py

@@ -1,4 +1,5 @@
 import uuid
+from typing import Optional
 from fastapi import HTTPException, Depends, Query, APIRouter
 
 from app.dependencies import get_session
@@ -36,7 +37,7 @@ def read_destination(destination_id: uuid.UUID, session=Depends(get_session)):
     return get_destination_by_id(session, destination_id)
 
 
-@router.get("/runs/{transfer_id}", response_model=TransferRun.Model)
+@router.get("/runs/{transfer_id}", response_model=Optional[TransferRun.Model])
 def get_transfer_run(transfer_id: uuid.UUID, session=Depends(get_session)):
     return TransferRun.get_latest_transfer_run(session, transfer_id)
 
@@ -62,7 +63,13 @@ def update_transfer_run(transfer_run_id:uuid.UUID, transfer_run:TransferRun.Upda
 
     # Get the destination vendor type
     transfer_id = transfer_run.transfer_id
-    destination_id = Transfer.get(session, transfer_id).destination_id
+    transfer = Transfer.get(session, transfer_id)
+
+    # Ad-hoc transfer runs don't have a parent transfer
+    if transfer is None:
+        return transfer_run
+
+    destination_id = transfer.destination_id
     destination_vendor_type = Destination.get(session, destination_id).connection_info['vendor_type']
 
     # Get the source vendor types from the models transferred

data-transfer/pontoon/Dockerfile

@@ -13,7 +13,7 @@ RUN apt-get update \
 COPY pyproject.toml ./pyproject.toml
 
 # Install the pontoon package and its dependencies
-RUN pip install --no-cache-dir .
+RUN pip install --only-binary=:all: --no-cache-dir .
 
 # Copy the pontoon source code
 COPY pontoon/ ./pontoon/

data-transfer/pontoon/pontoon/__init__.py

@@ -20,6 +20,7 @@
 from pontoon.base import Progress
 from pontoon.cache.memory_cache import MemoryCache
 from pontoon.cache.sqlite_cache import SqliteCache
+from pontoon.cache.arrow_ipc_cache import ArrowIpcCache
 from pontoon.base import Namespace, Stream, Record, Dataset, Cache, Mode, Source, Destination
 from pontoon.base import SourceConnectionFailed, SourceStreamDoesNotExist, SourceStreamInvalidSchema
 from pontoon.base import DestinationConnectionFailed, DestinationStreamInvalidSchema

data-transfer/pontoon/pontoon/base.py

@@ -136,7 +136,7 @@ def with_batch_id(self, batch_id:str, field_name:str='pontoon__batch_id') -> 'St
 
 
     def with_last_synced_at(self, sync_dt:datetime, field_name:str='pontoon__last_synced_at') -> 'Stream':
-        return self.with_field(field_name, pa.timestamp('us', tz='UTC'), sync_dt.isoformat())
+        return self.with_field(field_name, pa.timestamp('us', tz='UTC'), sync_dt)
 
 
     def with_version(self, version:str, field_name='pontoon__version') -> 'Stream':