convert-examples-repo-to-defs cloud nux quickstart_gcp

examples/quickstart_aws/pyproject.toml

@@ -1,3 +1,6 @@
 [build-system]
 requires = ["setuptools"]
 build-backend = "setuptools.build_meta"
+
+[tool.dagster]
+module_name = "quickstart_aws"

examples/quickstart_aws/quickstart_aws/__init__.py

@@ -1 +1,32 @@
-from .repository import quickstart_aws
+from dagster_aws.s3.io_manager import s3_pickle_io_manager
+from dagster_aws.s3.resources import s3_resource
+
+from dagster import (
+    Definitions,
+    ScheduleDefinition,
+    define_asset_job,
+    load_assets_from_package_module,
+)
+
+from . import assets
+
+daily_refresh_schedule = ScheduleDefinition(
+    job=define_asset_job(name="all_assets_job"), cron_schedule="0 0 * * *"
+)
+
+
+defs = Definitions(
+    assets=load_assets_from_package_module(assets),
+    # The AWS resources use boto under the hood, so if you are accessing your private
+    # buckets, you will need to provide the AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY
+    # environment variables or follow one of the other boto authentication methods.
+    # Read about using environment variables and secrets in Dagster:
+    #   https://docs.dagster.io/guides/dagster/using-environment-variables-and-secrets
+    resources={
+        # With this I/O manager in place, your job runs will store data passed between assets
+        # on S3 in the location s3://<bucket>/dagster/storage/<asset key>.
+        "io_manager": s3_pickle_io_manager.configured({"s3_bucket": {"env": "S3_BUCKET"}}),
+        "s3": s3_resource,
+    },
+    schedules=[daily_refresh_schedule],
+)

examples/quickstart_aws/quickstart_aws_tests/test_defs.py

@@ -0,0 +1,5 @@
+from quickstart_aws import defs
+
+
+def test_def_can_load():
+    assert defs.get_job_def("all_assets_job")

examples/quickstart_gcp/pyproject.toml

@@ -1,3 +1,6 @@
 [build-system]
 requires = ["setuptools"]
 build-backend = "setuptools.build_meta"
+
+[tool.dagster]
+module_name = "quickstart_gcp"

examples/quickstart_gcp/quickstart_gcp/__init__.py

@@ -1 +1,29 @@
-from .repository import quickstart_gcp
+from dagster import (
+    Definitions,
+    ScheduleDefinition,
+    define_asset_job,
+    load_assets_from_package_module,
+)
+
+from . import assets
+from .io_managers import bigquery_pandas_io_manager
+
+daily_refresh_schedule = ScheduleDefinition(
+    job=define_asset_job(name="all_assets_job"), cron_schedule="0 0 * * *"
+)
+
+
+defs = Definitions(
+    assets=load_assets_from_package_module(assets),
+    resources={
+        # Read about using environment variables and secrets in Dagster:
+        #   https://docs.dagster.io/guides/dagster/using-environment-variables-and-secrets
+        "io_manager": bigquery_pandas_io_manager.configured(
+            {
+                "credentials": {"env": "BIGQUERY_SERVICE_ACCOUNT_CREDENTIALS"},
+                "project_id": {"env": "BIGQUERY_PROJECT_ID"},
+            }
+        ),
+    },
+    schedules=[daily_refresh_schedule],
+)

examples/quickstart_gcp/quickstart_gcp_tests/test_defs.py

@@ -0,0 +1,5 @@
+from quickstart_gcp import defs
+
+
+def test_def_can_load():
+    assert defs.get_job_def("all_assets_job")