add s3 object name

Author: MattShirley

servicex_app/migrations/versions/1.6.2.py renamed to servicex_app/migrations/versions/1.7.0.py

@@ -1,13 +1,13 @@
 """
-Revision ID: v1_6_2
+Revision ID: v1_7_0
 Revises: v1_5_7a
 Create Date: 2025-05-13
 
 """
 from alembic import op
 import sqlalchemy as sa
 
-revision = 'v1_6_2'
+revision = 'v1_7_0'
 down_revision = 'v1_5_7a'
 branch_labels = None
 depends_on = None
@@ -17,11 +17,20 @@ def upgrade():
     op.add_column('transform_result', sa.Column('created_at', sa.DateTime(), nullable=True))
     op.execute('UPDATE transform_result SET created_at = CURRENT_TIMESTAMP')
     op.alter_column('transform_result', 'created_at', nullable=False)
-
     op.create_index('ix_transform_result_created_at', 'transform_result', ['created_at'])
 
+    op.add_column('transform_result', sa.Column('s3_object_name', sa.String(length=512), nullable=True))
+    op.create_index('ix_transform_result_s3_object_name', 'transform_result', ['s3_object_name'])
+
+    op.create_index('ix_transform_result_request_id', 'transform_result', ['request_id'])
+
 
 def downgrade():
     op.drop_index('ix_transform_result_created_at', table_name='transform_result')
-
     op.drop_column('transform_result', 'created_at')
+
+    op.drop_index('ix_transform_result_s3_object_name', table_name='transform_result')
+    op.drop_column('transform_result', 's3_object_name')
+
+    op.drop_index('ix_transform_result_request_id', table_name='transform_result')
+

servicex_app/servicex_app/models.py

@@ -357,6 +357,7 @@ class TransformationResult(db.Model):
     total_bytes = db.Column(db.BigInteger, nullable=True)
     avg_rate = db.Column(db.Float, nullable=True)
     created_at = db.Column(DateTime, default=func.now())
+    s3_object_name = db.Column(db.String(512), unique=False, nullable=True)
 
     __table_args__ = (
         db.UniqueConstraint('file_id', 'request_id', name='uix_file_request'),

servicex_app/servicex_app/resources/internal/transformer_file_complete.py

@@ -177,7 +177,8 @@ def save_transform_result(request_id: str, info: dict[str, str], session: Sessio
                 transform_time=info['total-time'],
                 total_bytes=info['total-bytes'],
                 total_events=info['total-events'],
-                avg_rate=info['avg-rate']
+                avg_rate=info['avg-rate'],
+                s3_object_name=info['s3-object-name'],
             )
             session.add(rec)
 

servicex_app/servicex_app/resources/transformation/results.py

@@ -18,7 +18,7 @@ def get(self, request_id):
 
         parser = reqparse.RequestParser()
         parser.add_argument(
-            'begin_at',
+            'later_than',
             type=str,
             required=False,
             location='args'
@@ -28,13 +28,13 @@ def get(self, request_id):
 
         transform_result_query = TransformationResult.query.filter_by(request_id=request_id)
 
-        begin_at_str = args.get('begin_at')
-        if begin_at_str:
+        later_than_str = args.get('later_than')
+        if later_than_str:
             try:
-                begin_at = datetime.datetime.fromisoformat(begin_at_str)
+                later_than = datetime.datetime.fromisoformat(later_than_str)
             except AttributeError:
-                return {"message": f"begin_at value {begin_at_str} is not an ISO 8601 compliant datetime"}, 400
-            transform_result_query = transform_result_query.filter(TransformationResult.created_at > begin_at)
+                return {"message": f"later_than value {later_than_str} is not an ISO 8601 compliant datetime"}, 400
+            transform_result_query = transform_result_query.filter(TransformationResult.created_at > later_than)
 
         results = [transformation_result.to_json(transformation_result) for transformation_result in transform_result_query]
 

transformer_sidecar/src/transformer_sidecar/servicex_adapter.py

@@ -47,14 +47,16 @@
 
 class FileCompleteRecord:
     def __init__(self, request_id: str, file_path: str, file_id: int, status: str,
-                 total_time: float, total_events: int, total_bytes: int):
+                 total_time: float, total_events: int, total_bytes: int, s3_object_name: str):
         assert request_id, "request_id is required"
         assert file_path, "file_path is required"
+        assert s3_object_name, "s3_object_name is required"
         assert file_id, "file_id is required"
         assert status, "status is required"
 
         self.request_id = request_id
         self.file_path = file_path
+        self.s3_object_name = s3_object_name
         self.file_id = file_id
         self.status = status
         self.total_time = total_time
@@ -66,6 +68,7 @@ def to_json(self) -> dict[str, Any]:
         return {
             "requestId": self.request_id,
             "file-path": self.file_path,
+            "s3-object-name": self.s3_object_name,
             "file-id": self.file_id,
             "status": self.status,
             "total-time": self.total_time,

transformer_sidecar/src/transformer_sidecar/transformer.py

@@ -163,7 +163,7 @@ def transform_file(
     transformer_stats = TransformerStats()
     try:
         # Loop through the replicas
-        for _file_path in _file_paths:
+        for path, _file_path in zip(paths, _file_paths):
             logger.info(
                 "trying to transform file",
                 extra={
@@ -215,6 +215,7 @@ def transform_file(
                 rec = FileCompleteRecord(
                     request_id=request_id,
                     file_path=_file_path,
+                    s3_object_name=path,
                     file_id=file_id,
                     status="success",
                     total_time=time.time() - total_time,
@@ -256,6 +257,7 @@ def transform_file(
             rec = FileCompleteRecord(
                 request_id=request_id,
                 file_path=_file_paths[0],
+                s3_object_name=paths[0],
                 file_id=file_id,
                 status="failure",
                 total_time=time.time() - total_time,
@@ -292,6 +294,7 @@ def transform_file(
         rec = FileCompleteRecord(
             request_id=request_id,
             file_path=_file_paths[0],
+            s3_object_name=paths[0],
             file_id=file_id,
             status="failure",
             total_time=time.time() - total_time,