Week 1 Project Submission

week_1/project/week_1.py

@@ -2,7 +2,7 @@
 from datetime import datetime
 from typing import List
 
-from dagster import In, Nothing, Out, job, op, usable_as_dagster_type
+from dagster import In, Nothing, Out, job, op, usable_as_dagster_type, get_dagster_logger
 from pydantic import BaseModel
 
 
@@ -50,16 +50,29 @@ def get_s3_data(context):
     return output
 
 
-@op
-def process_data():
-    pass
+@op(
+    ins={"stocks": In(dagster_type=List[Stock])},
+    out={"agg_max": Out(dagster_type=Aggregation)},
+    description="Determine the stock with the highest value"
+)
+def process_data(stocks: List[Stock]) -> Aggregation:
+    # Find the stock with the highest value
+    max_stock = max(stocks, key=lambda x: x.high)
+    return Aggregation(date=max_stock.date, high=max_stock.high)
 
 
-@op
-def put_redis_data():
+@op(
+    ins={"agg_max": In(dagster_type=Aggregation)},
+    out=None,
+    description="Write to Redis"
+)
+def put_redis_data(agg_max: Aggregation) -> None:
+    # Log the output
+    logger = get_dagster_logger()
+    logger.info(f"Write {agg_max} to Redis.")
     pass
 
 
 @job
 def week_1_pipeline():
-    pass
+    put_redis_data(process_data(get_s3_data()))