Finish week_2 dagster corise

week_2/dagster_ucr/project/week_2.py

@@ -1,30 +1,50 @@
 from typing import List
 
-from dagster import In, Nothing, Out, ResourceDefinition, graph, op
+from dagster import ResourceDefinition, graph, op
 from dagster_ucr.project.types import Aggregation, Stock
 from dagster_ucr.resources import mock_s3_resource, redis_resource, s3_resource
 
 
-@op
-def get_s3_data():
-    pass
+@op(
+    required_resource_keys={"s3"},
+    config_schema={"s3_key": str},
+    tags={"kind": "s3"},
+)
+def get_s3_data(context) -> List[Stock]:
+    s3 = context.resources.s3
+    return [
+        Stock.from_list(data)
+        for data in s3.get_data(context.op_config["s3_key"])
+    ]
 
 
 @op
-def process_data():
+def process_data(stocks: List[Stock]) -> Aggregation:
     # Use your op from week 1
-    pass
+    highest = max(stocks, key=lambda stock: stock.high)
+    return Aggregation(
+        date=highest.date,
+        high=highest.high,
+    )
 
 
-@op
-def put_redis_data():
-    pass
+@op(
+    required_resource_keys={"redis"},
+    tags={"kind": "redis"}
+)
+def put_redis_data(context, agg: Aggregation):
+    redis = context.resources.redis
+    key, val = agg.date.strftime("%Y-%m-%d"), agg.high
+    redis.put_data(key, val)
+    context.log.info(f"put key: ({key}, val: {val}) to redis")
 
 
 @graph
 def week_2_pipeline():
     # Use your graph from week 1
-    pass
+    stocks = get_s3_data()
+    agg = process_data(stocks)
+    put_redis_data(agg)
 
 
 local = {

week_2/dagster_ucr/resources.py

@@ -4,6 +4,7 @@
 
 import boto3
 import redis
+from regex import W
 import sqlalchemy
 from dagster import Field, Int, String, resource
 
@@ -91,13 +92,33 @@ def mock_s3_resource():
     return s3_mock
 
 
-@resource
-def s3_resource():
+@resource(
+    config_schema={
+        "bucket": str,
+        "access_key": str,
+        "secret_key": str,
+        "endpoint_url": str,
+    }
+)
+def s3_resource(context):
     """This resource defines a S3 client"""
-    pass
+    return S3(
+        bucket=context.resource_config["bucket"],
+        access_key=context.resource_config["access_key"],
+        secret_key=context.resource_config["secret_key"],
+        endpoint_url=context.resource_config["endpoint_url"],
+    )
 
 
-@resource
-def redis_resource():
+@resource(
+    config_schema={
+        "host": str,
+        "port": int,
+    }
+)
+def redis_resource(context):
     """This resource defines a Redis client"""
-    pass
+    return Redis(
+        host=context.resource_config["host"],
+        port=context.resource_config["port"],
+    )

week_2/tests/test_answer.py

@@ -50,7 +50,7 @@ def test_aggregation(aggregation):
 def test_get_s3_data(stock_list):
     s3_mock = MagicMock()
     s3_mock.get_data.return_value = [stock_list] * 10
-    with build_op_context(op_config={"s3_key": "data/stock.csv"}, resources={"s3": s3_mock}) as context:
+    with build_op_context(op_config={"s3_key": "week2/data/stock.csv"}, resources={"s3": s3_mock}) as context:
         get_s3_data(context)
         assert s3_mock.get_data.called