Add Completed Transfer Table

alembic/versions/71df5b41ae41_initial_schema.py

@@ -14,7 +14,6 @@
 branch_labels = None
 depends_on = None
 
-import enum
 
 from sqlalchemy import (
     BigInteger,

alembic/versions/84785333a677_add_completed_transfers_table.py

@@ -0,0 +1,43 @@
+# Copyright 2017 the HERA Collaboration
+# Licensed under the 2-clause BSD License.
+
+"""Add completed_transfers table
+
+Revision ID: 84785333a677
+Revises: 38a604ac628b
+Create Date: 2025-07-25 00:51:04.843683
+
+"""
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "84785333a677"
+down_revision = "38a604ac628b"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        "completed_transfers",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("task_id", sa.String(), nullable=False),
+        sa.Column("source_endpoint_id", sa.String(), nullable=False),
+        sa.Column("destination_endpoint_id", sa.String(), nullable=False),
+        sa.Column("start_time", sa.DateTime(), nullable=False),
+        sa.Column("end_time", sa.DateTime(), nullable=False),
+        sa.Column("duration_seconds", sa.Float(), nullable=False),
+        sa.Column("bytes_transferred", sa.BigInteger(), nullable=False),
+        sa.Column("effective_bandwidth_bps", sa.Float(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["id"],
+            ["send_queue.id"],
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("task_id"),
+    )
+
+
+def downgrade():
+    op.drop_table("completed_transfers")

alembic/versions_1.0/0e0e6d02a01a_add_index_on_fileevent_name.py

@@ -9,7 +9,6 @@
 Create Date: 2016-12-13 11:01:34.767814
 
 """
-import sqlalchemy as sa
 
 from alembic import op
 

alembic/versions_1.0/b02eb31402b1_add_index_on_fileinstance_name.py

@@ -9,7 +9,6 @@
 Create Date: 2017-02-09 17:32:01.979450
 
 """
-import sqlalchemy as sa
 
 from alembic import op
 

hera_librarian/async_transfers/globus.py

@@ -3,10 +3,13 @@
 """
 
 import os
+from datetime import datetime
 from pathlib import Path
 
 import globus_sdk
+from loguru import logger
 
+from hera_librarian.models.transfer import CompletedTransferCore
 from hera_librarian.transfer import TransferStatus
 from hera_librarian.utils import GLOBUS_ERROR_EVENTS
 
@@ -91,7 +94,7 @@ def authorize(self, settings: "ServerSettings"):
                 authorizer = globus_sdk.RefreshTokenAuthorizer(
                     refresh_token=settings.globus_client_secret, auth_client=client
                 )
-            except globus_sdk.AuthAPIError as e:
+            except globus_sdk.AuthAPIError:
                 return None
         else:
             try:
@@ -107,7 +110,7 @@ def authorize(self, settings: "ServerSettings"):
                 authorizer = globus_sdk.AccessTokenAuthorizer(
                     access_token=transfer_token
                 )
-            except globus_sdk.AuthAPIError as e:
+            except globus_sdk.AuthAPIError:
                 return None
 
         return authorizer
@@ -213,10 +216,11 @@ def transfer(
         # try to submit the task
         try:
             task_doc = transfer_client.submit_transfer(transfer_data)
-        except globus_sdk.TransferAPIError as e:
+        except globus_sdk.TransferAPIError:
             return False
 
         self.task_id = task_doc["task_id"]
+
         return True
 
     def batch_transfer(
@@ -284,7 +288,7 @@ def batch_transfer(
         # submit the transfer
         try:
             task_doc = transfer_client.submit_transfer(transfer_data)
-        except globus_sdk.TransferAPIError as e:
+        except globus_sdk.TransferAPIError:
             return False
 
         self.task_id = task_doc["task_id"]
@@ -376,6 +380,63 @@ def fail_transfer(self, settings: "ServerSettings") -> bool:
 
         try:
             _ = transfer_client.cancel_task(self.task_id)
-        except globus_sdk.TransferAPIError as e:
+        except globus_sdk.TransferAPIError:
             return False
         return True
+
+    def gather_transfer_details(
+        self, settings: "ServerSettings"
+    ) -> CompletedTransferCore | None:
+        """
+        Gathers details about a completed transfer from Globus and
+        returns them in a Pydantic object.
+        """
+
+        if self.authorizer is None:
+            logger.debug("Authorizer not provided, attempting internal authorization")
+            self.authorizer = self.authorize(settings=settings)
+
+        if not self.authorizer:
+            logger.error("Authorization failed")
+            return None
+
+        logger.debug("Authorization successful")
+
+        transfer_client = globus_sdk.TransferClient(authorizer=self.authorizer)
+
+        try:
+            logger.debug(f"Fetching task details for ID: {self.task_id}")
+            task_doc = transfer_client.get_task(self.task_id)
+            logger.debug(f"Task data fetched. Status is: {task_doc['status']}")
+        except globus_sdk.TransferAPIError as e:
+            logger.error(f"Globus API Error when fetching task: {e}")
+            return None
+
+        if task_doc["status"] != "SUCCEEDED":
+            logger.warning("Task status is not SUCCEEDED.")
+            return None
+
+        start_time = datetime.fromisoformat(task_doc["request_time"])
+        end_time = datetime.fromisoformat(task_doc["completion_time"])
+        bytes_transferred = task_doc["bytes_transferred"]
+        bandwidth_bps = task_doc["effective_bytes_per_second"]
+
+        duration = end_time - start_time
+        duration_seconds = duration.total_seconds()
+
+        try:
+            transfer_record = CompletedTransferCore(
+                task_id=task_doc["task_id"],
+                source_endpoint_id=task_doc["source_endpoint_id"],
+                destination_endpoint_id=task_doc["destination_endpoint_id"],
+                start_time=start_time,
+                end_time=end_time,
+                duration_seconds=duration_seconds,
+                bytes_transferred=bytes_transferred,
+                effective_bandwidth_bps=bandwidth_bps,
+            )
+            return transfer_record
+        except (KeyError, ValueError) as e:
+            logger.error(
+                f"Missing key or malformed value: {e} related to task {self.task_id}"
+            )

hera_librarian/async_transfers/local.py

@@ -5,10 +5,14 @@
 import copy
 import os
 import shutil
+from datetime import datetime, timezone
 from pathlib import Path
 from socket import gethostname
+from typing import Optional
 
+from hera_librarian.models.transfer import CompletedTransferCore
 from hera_librarian.transfer import TransferStatus
+from hera_librarian.utils import get_size_from_path
 
 from .core import CoreAsyncTransferManager
 
@@ -19,7 +23,18 @@ class LocalAsyncTransferManager(CoreAsyncTransferManager):
     transfer_attempted: bool = False
     transfer_complete: bool = False
 
+    # Attributes to store performance metrics for the completion report
+    start_time_transfer: Optional[datetime] = None
+    bytes_transfer: Optional[int] = None
+
     def batch_transfer(self, paths: list[tuple[Path]], settings: "ServerSettings"):
+        # Initialization of time and total bytes
+        self.start_time_transfer = datetime.now(timezone.utc)
+        # Use the existing utility function to calculate the total size
+        total_bytes = sum(get_size_from_path(local_path) for local_path, _ in paths)
+
+        self.bytes_transfer = total_bytes
+
         copy_success = True
 
         for local_path, remote_path in paths:
@@ -126,3 +141,40 @@ def transfer_status(self, settings: "ServerSettings") -> TransferStatus:
 
     def fail_transfer(self, settings: "ServerSettings") -> bool:
         return True
+
+    def gather_transfer_details(self) -> CompletedTransferCore | None:
+        """
+        Gathers details about a locall completed transfer and
+        returns them in a CompletedTransferCore Pydantic Object
+        """
+        # Check if the transfer metrics were recorded
+        if self.start_time_transfer is None or self.bytes_transfer is None:
+            logger.error("Transfer metrics were not recorded")
+            return None
+
+        # Calculate performance metrics from the recorded data
+        end_time = datetime.now(timezone.utc)
+        duration_seconds = (end_time - self.start_time_transfer).total_seconds()
+        total_bytes = self.bytes_transfer
+
+        if duration_seconds > 0:
+            duration_seconds = 1
+
+        bandwidth_bps = total_bytes / duration_seconds
+
+        try:
+            transfer_record = CompletedTransferCore(
+                task_id=f"local_{end_time}",
+                source_endpoint_id=gethostname(),
+                destination_endpoint_id=gethostname(),
+                start_time=self.start_time_transfer,
+                end_time=end_time,
+                duration_seconds=duration_seconds,
+                bytes_transferred=total_bytes,
+                effective_bandwidth_bps=bandwidth_bps,
+            )
+            return transfer_record
+        except (KeyError, ValueError) as e:
+            logger.error(
+                f"Failed to create transfer report object due to a data validation error: {e}"
+            )

hera_librarian/async_transfers/rsync.py

@@ -43,7 +43,7 @@ def transfer(self, local_path: Path, remote_path: Path, settings: "ServerSetting
             )
 
             return True
-        except sysrsync.RsyncError as e:
+        except sysrsync.RsyncError:
             return False
 
     def batch_transfer(self, paths: list[tuple[Path]], settings: "ServerSettings"):

hera_librarian/cli.py

@@ -8,10 +8,8 @@
 
 import argparse
 import datetime
-import json
 import os
 import sys
-import time
 from pathlib import Path
 from typing import Optional
 
@@ -630,8 +628,8 @@ def get_librarian_list(args):
         librarian_list = client.get_librarian_list().librarians
     except LibrarianHTTPError as e:
         die(f"Unexpected error communicating with the librarian server: {e.reason}")
-    except LibrarianError as e:
-        die(f"You are not authorized to perform this action.")
+    except LibrarianError:
+        die("You are not authorized to perform this action.")
 
     if len(librarian_list) == 0:
         print("No librarians found.")

hera_librarian/client.py

@@ -49,7 +49,6 @@
 from .models.errors import (
     ErrorClearRequest,
     ErrorClearResponse,
-    ErrorSearchFailedResponse,
     ErrorSearchRequest,
     ErrorSearchResponse,
     ErrorSearchResponses,
@@ -76,11 +75,7 @@
     FileValidationResponseItem,
 )
 from .settings import ClientInfo
-from .utils import (
-    get_checksum_from_path,
-    get_hash_function_from_hash,
-    get_size_from_path,
-)
+from .utils import get_checksum_from_path, get_size_from_path
 
 if TYPE_CHECKING:
     from .transfers import CoreTransferManager

hera_librarian/models/clone.py

@@ -5,7 +5,7 @@
 from pathlib import Path
 from typing import Union
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel
 
 from hera_librarian.async_transfers import (
     GlobusAsyncTransferManager,

hera_librarian/models/search.py

@@ -3,12 +3,10 @@
 """
 
 from datetime import datetime
-from pathlib import Path
 from typing import Optional
 
 from pydantic import BaseModel, Field, RootModel
 
-from hera_librarian.deletion import DeletionPolicy
 from hera_librarian.models.instances import (
     InstanceSearchResponse,
     RemoteInstanceSearchResponse,

hera_librarian/models/transfer.py

@@ -0,0 +1,16 @@
+from pydantic import BaseModel
+from datetime import datetime
+
+class CompletedTransferCore(BaseModel):
+    """
+    A Pydantic model representing the data for a completed transfer.
+    """
+
+    task_id: str
+    source_endpoint_id: str
+    destination_endpoint_id: str
+    start_time: datetime
+    end_time: datetime
+    duration_seconds: float 
+    bytes_transferred: int 
+    effective_bandwidth_bps: float

hera_librarian/models/uploads.py

@@ -5,9 +5,9 @@
 from pathlib import Path
 from typing import Union
 
-from pydantic import BaseModel, SerializeAsAny, field_validator
+from pydantic import BaseModel
 
-from ..transfers import CoreTransferManager, LocalTransferManager
+from ..transfers import LocalTransferManager
 
 
 class UploadInitiationRequest(BaseModel):

librarian_background/check_integrity.py

@@ -12,7 +12,7 @@
 from hera_librarian.utils import compare_checksums
 from librarian_server.database import get_session
 from librarian_server.orm import Instance, StoreMetadata
-from librarian_server.orm.file import CorruptFile, File
+from librarian_server.orm.file import CorruptFile
 
 from .task import Task
 

librarian_background/core.py

@@ -10,8 +10,6 @@
 
 from schedule import Scheduler
 
-from .task import Task
-
 logger = logging.getLogger("schedule")
 
 

librarian_background/corruption_fixer.py

@@ -2,7 +2,7 @@
 A background task that queries the corrupt files table and remedies them.
 """
 
-from datetime import datetime, timedelta, timezone
+from datetime import datetime, timezone
 from time import perf_counter
 
 from loguru import logger