Initial pipeline

liiatools_pipeline/assets/external_dataset.py

@@ -7,3 +7,12 @@
 def external_data_folder():
     external_data_location = env_config("EXTERNAL_DATA_LOCATION", cast=str)
     return open_fs(external_data_location)
+
+
+@asset
+def ons_data_location():
+    return env_config("ONS_DATA_URL", cast=str)
+
+@asset
+def ofsted_data_location():
+    return env_config("OFSTED_DATA_URL", cast=str)

liiatools_pipeline/jobs/external_dataset.py

@@ -1,8 +1,9 @@
 from dagster import job
-
+from decouple import config as env_config
 from liiatools_pipeline.ops import external_dataset as ed
-
+from liiatools_pipeline.assets import external_dataset as ed_assets
 
 @job
 def external_incoming():
-    ed.request_dataset()
+    ed.request_dataset(ed_assets.ons_data_location())
+    ed.request_dataset(ed_assets.ofsted_data_location())

liiatools_pipeline/jobs/sufficiency_903.py

@@ -0,0 +1,9 @@
+from sufficiency_data_transform.all_dim_and_fact import create_sufficiency_data
+from dagster import job
+from liiatools_pipeline.ops import ssda903
+
+
+@job
+def ssda903_sufficiency():
+    print("Hi")
+    create_sufficiency_data()

liiatools_pipeline/ops/external_dataset.py

@@ -6,6 +6,6 @@
 from liiatools_pipeline.assets.external_dataset import external_data_folder
 
 
-@op()
-def request_dataset():
-    retrieve_dataset("http://localhost:8080/sample.csv", external_data_folder)
+@op
+def request_dataset(url):
+    retrieve_dataset(url, external_data_folder())

liiatools_pipeline/repository.py

@@ -3,7 +3,9 @@
 
 from liiatools_pipeline.jobs.ssda903 import ssda903_incoming
 from liiatools_pipeline.jobs.external_dataset import external_incoming
+from liiatools_pipeline.jobs.sufficiency_903 import ssda903_sufficiency
 from liiatools_pipeline.sensors.location_sensor import location_sensor
+#from liiatools_pipeline.sensors.sufficiency_sensor import sufficiency_sensor
 
 register()
 
@@ -16,8 +18,10 @@ def sync():
     For hints on building your Dagster repository, see our documentation overview on Repositories:
     https://docs.dagster.io/overview/repositories-workspaces/repositories
     """
-    jobs = [ssda903_incoming, external_incoming]
+    jobs = [ssda903_incoming, external_incoming, ssda903_sufficiency]
     schedules = []
     sensors = [location_sensor]
 
     return jobs + schedules + sensors
+
+#sufficiency_sensor

liiatools_pipeline/sensors/sufficiency_sensor.py

@@ -0,0 +1,21 @@
+from dagster import RunRequest, run_status_sensor, sensor, DagsterRunStatus, RunConfig, DefaultSensorStatus
+
+from liiatools_pipeline.jobs.ssda903 import ssda903_incoming
+from liiatools_pipeline.jobs.sufficiency_903 import ssda903_sufficiency
+
+'''
+@run_status_sensor(
+    run_status=DagsterRunStatus.SUCCESS,
+    description="Adds in ONS and census data once 903 pipeline completes",
+    request_job=ssda903_incoming,
+)
+@sensor(
+    job=ssda903_sufficiency,
+    minimum_interval_seconds=300,
+    default_status=DefaultSensorStatus.RUNNING,
+)
+def sufficiency_sensor(context):
+    yield RunRequest(
+        run_key=None,
+        run_config=RunConfig(),
+    )'''