georgolm week3

week_1/project/week_1.py

@@ -50,16 +50,25 @@ def get_s3_data(context):
     return output
 
 
-@op
-def process_data():
-    pass
-
+@op(
+    out={"max_value": Out(dagster_type=Aggregation)},
+    description="Filter for largest daily high stock value"
+)
+def process_data(stocks):
+    #get max of daily highest stock value
+    stock_high = max(stocks, key = lambda x: x.high)
+    return Aggregation(date = stock_high.date, high = stock_high.high)
 
 @op
-def put_redis_data():
-    pass
+def put_redis_data(processed_data):
+    # output data in console for each run
+    string = str(processed_data)
+    f = open('output.txt', 'w')
+    f.write(string)
 
 
 @job
 def week_1_pipeline():
-    pass
+    stock_data = get_s3_data()
+    agg_data = process_data(stock_data)
+    put_redis_data(agg_data)

week_2/dagster_ucr/project/week_2.py

@@ -5,26 +5,53 @@
 from dagster_ucr.resources import mock_s3_resource, redis_resource, s3_resource
 
 
-@op
-def get_s3_data():
-    pass
+@op(
+    config_schema={"s3_key": str},
+    required_resource_keys={"s3"},
+    out={"stocks": Out(dagster_type=List[Stock])},
+    tags={"kind": "s3"},
+    description="Get a list of stocks from an S3 file",
+)
+def get_s3_data(context):
+    output = list()
+    key = context.op_config["s3_key"]
+    data = context.resources.s3.get_data(key)
+
+    for row in data:
+        stock = Stock.from_list(row)
+        output.append(stock)
+    return output
 
 
-@op
-def process_data():
-    # Use your op from week 1
-    pass
+@op(
+    out={"max_value": Out(dagster_type=Aggregation)},
+    description="Filter for largest daily high stock value"
+)
+def process_data(stocks):
+    #get max of daily highest stock value
+    stock_high = max(stocks, key = lambda x: x.high)
+    return Aggregation(date = stock_high.date, high = stock_high.high)
 
 
-@op
-def put_redis_data():
-    pass
+@op(
+     ins={"stock_high": In(dagster_type=Aggregation)},
+     required_resource_keys={"redis"},
+     tags={"kind": "Redis"}
+ )
+def put_redis_data(context, stock_high):
+    context.resources.redis.put_data(
+        name = str(stock_high.date), 
+        value = str(stock_high.high)
+        )
+
 
 
 @graph
 def week_2_pipeline():
-    # Use your graph from week 1
-    pass
+    stock_data = get_s3_data()
+    agg_data = process_data(stock_data)
+    put_redis_data(agg_data)
+
 
 
 local = {

week_2/dagster_ucr/resources.py

@@ -68,7 +68,6 @@ def put_data(self, name: str, value: str):
     description="A resource that can run Postgres",
 )
 def postgres_resource(context) -> Postgres:
-    """This resource defines a Postgres client"""
     return Postgres(
         host=context.resource_config["host"],
         user=context.resource_config["user"],
@@ -91,13 +90,33 @@ def mock_s3_resource():
     return s3_mock
 
 
-@resource
-def s3_resource():
-    """This resource defines a S3 client"""
-    pass
+@resource(
+    config_schema={
+        "bucket": Field(String),
+        "access_key": Field(String),
+        "secret_key": Field(String),
+        "endpoint_url": Field(String),
+    },
+    description="A resource connecting to S3",
+)
+def s3_resource(context) -> S3:
+    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():
-    """This resource defines a Redis client"""
-    pass
+@resource(
+    config_schema={
+        "host": Field(String),
+        "port": Field(Int)
+    },
+    description="A resource connecting to redis",
+)
+def redis_resource(context) -> Redis:
+    return Redis(
+        host=context.resource_config["host"],
+        port=context.resource_config["port"]
+    )