feature/video3-dagster-job-and-schedule

Coder2j · Coder2j · commit 38e863550717 · 2023-11-26T21:53:03.000+01:00
* added dagster job
* added dagster schedule
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
@@ -1,10 +1,34 @@
 # See https://pre-commit.com for more information
 # See https://pre-commit.com/hooks.html for more hooks
 repos:
--   repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v3.2.0
+  - repo: https://github.com/pre-commit/pre-commit-hooks
+    rev: v4.5.0
     hooks:
-    -   id: trailing-whitespace
-    -   id: end-of-file-fixer
-    -   id: check-yaml
-    -   id: check-added-large-files
+      - id: check-added-large-files # prevents giant files from being committed.
+      - id: check-case-conflict # checks for files that would conflict in case-insensitive filesystems.
+      - id: check-merge-conflict # checks for files that contain merge conflict strings.
+      - id: check-yaml # checks yaml files for parseable syntax.
+      - id: detect-private-key # detects the presence of private keys.
+      - id: end-of-file-fixer # ensures that a file is either empty, or ends with one newline.
+      - id: fix-byte-order-marker # removes utf-8 byte order marker.
+      - id: mixed-line-ending # replaces or checks mixed line ending.
+      - id: requirements-txt-fixer # sorts entries in requirements.txt.
+      - id: trailing-whitespace # trims trailing whitespace.
+  - repo: https://github.com/pre-commit/mirrors-mypy
+    rev: "v1.7.0" # Use the sha / tag you want to point at
+    hooks:
+      - id: mypy
+  - repo: https://github.com/pre-commit/mirrors-prettier
+    rev: "v3.1.0" # Use the sha / tag you want to point at
+    hooks:
+      - id: prettier
+  - repo: https://github.com/psf/black
+    rev: 23.11.0
+    hooks:
+      - id: black
+        args: [--line-length=79]
+  - repo: https://github.com/asottile/reorder-python-imports
+    rev: v3.12.0
+    hooks:
+      - id: reorder-python-imports
+        args: [--py38-plus]
diff --git a/hello-dagster.py b/hello-dagster.py
@@ -1,5 +1,12 @@
 from typing import List
-from dagster import asset, AssetExecutionContext, AssetIn
+
+from dagster import asset
+from dagster import AssetExecutionContext
+from dagster import AssetIn
+from dagster import AssetSelection
+from dagster import define_asset_job
+from dagster import Definitions
+from dagster import ScheduleDefinition
 
 
 @asset(key="my_awesome_first_asset", group_name="get_started")
@@ -12,7 +19,10 @@ def my_first_asset(context: AssetExecutionContext):
     return [1, 2, 3]
 
 
-@asset(ins={"upstream": AssetIn(key="my_awesome_first_asset")}, group_name="get_started")
+@asset(
+    ins={"upstream": AssetIn(key="my_awesome_first_asset")},
+    group_name="get_started",
+)
 def my_second_asset(context: AssetExecutionContext, upstream: List):
     """
     This is our second asset
@@ -22,21 +32,41 @@ def my_second_asset(context: AssetExecutionContext, upstream: List):
     return data
 
 
-@asset(ins={
-    "first_upstream": AssetIn("my_awesome_first_asset"),
-    "second_upstream": AssetIn("my_second_asset")
-}, group_name="get_started")
+@asset(
+    ins={
+        "first_upstream": AssetIn("my_awesome_first_asset"),
+        "second_upstream": AssetIn("my_second_asset"),
+    },
+    group_name="get_started",
+)
 def my_third_asset(
-    context: AssetExecutionContext,
-    first_upstream: List,
-    second_upstream: List):
+    context: AssetExecutionContext, first_upstream: List, second_upstream: List
+):
     """
     This is our third asset
     """
     data = {
         "first_asset": first_upstream,
         "second_asset": second_upstream,
-        "third_asset": second_upstream + [7, 8]
+        "third_asset": second_upstream + [7, 8],
     }
     context.log.info(f"Output data is: {data}")
     return data
+
+
+defs = Definitions(
+    assets=[my_first_asset, my_second_asset, my_third_asset],
+    jobs=[
+        define_asset_job(
+            name="hello_dagster_job",
+            selection=AssetSelection.groups("get_started"),
+        )
+    ],
+    schedules=[
+        ScheduleDefinition(
+            name="hello_dagster_schedule",
+            job_name="hello_dagster_job",
+            cron_schedule="* * * * *",
+        )
+    ],
+)
