workflow orchestration

Author: KayvanShah1

dag/.prefectignore

@@ -0,0 +1,38 @@
+# prefect artifacts
+.prefectignore
+
+# python artifacts
+__pycache__/
+*.py[cod]
+*$py.class
+*.egg-info/
+*.egg
+
+# Type checking artifacts
+.mypy_cache/
+.dmypy.json
+dmypy.json
+.pyre/
+
+# IPython
+profile_default/
+ipython_config.py
+*.ipynb_checkpoints/*
+
+# Environments
+.python-version
+.env
+.venv
+env/
+venv/
+
+# MacOS
+.DS_Store
+
+# Dask
+dask-worker-space/
+
+# Editors
+.idea/
+.vscode/
+*.yaml

dag/Dockerfile

@@ -0,0 +1,9 @@
+# We're using the latest version of Prefect with Python 3.10
+FROM prefecthq/prefect:2-python3.10
+
+# Add our requirements.txt file to the image and install dependencies
+COPY requirements.txt .
+RUN pip install -r requirements.txt --trusted-host pypi.python.org --no-cache-dir
+
+# Add our flow code to the image
+ADD flows /opt/prefect/flows

dag/flows/healthcheck.py

@@ -0,0 +1,23 @@
+import platform
+import prefect
+from prefect import task, flow, get_run_logger
+import sys
+
+
+@task
+def log_platform_info():
+    logger = get_run_logger()
+    logger.info("Host's network name = %s", platform.node())
+    logger.info("Python version = %s", platform.python_version())
+    logger.info("Platform information (instance type) = %s ", platform.platform())
+    logger.info("OS/Arch = %s/%s", sys.platform, platform.machine())
+    logger.info("Prefect Version = %s 🚀", prefect.__version__)
+
+
+@flow
+def healthcheck():
+    log_platform_info()
+
+
+if __name__ == "__main__":
+    healthcheck()

dag/flows/sample.py

@@ -0,0 +1,17 @@
+import httpx
+from prefect import flow
+
+
+@flow(log_prints=True)
+def get_repo_info(repo_name: str = "PrefectHQ/prefect"):
+    url = f"https://api.github.com/repos/{repo_name}"
+    response = httpx.get(url)
+    response.raise_for_status()
+    repo = response.json()
+    print(f"{repo_name} repository statistics 🤓:")
+    print(f"Stars 🌠 : {repo['stargazers_count']}")
+    print(f"Forks 🍴 : {repo['forks_count']}")
+
+
+if __name__ == "__main__":
+    get_repo_info.serve(name="prefect-docker-guide")

dag/flows/ss_etl.py

@@ -0,0 +1,40 @@
+from prefect import flow, task
+
+from steam_sales.steam_etl import SteamDataClean, SteamSpyFetcher, SteamSpyMetadataFetcher, SteamStoreFetcher
+
+
+@task(name="Fetch & Update Metadata")
+def update_metadata():
+    fetcher = SteamSpyMetadataFetcher()
+    fetcher.run()
+    return 0
+
+
+@task(name="Fetch & Update SteamSpy data")
+def update_steamspy_data():
+    fetcher = SteamSpyFetcher()
+    fetcher.run()
+
+
+@task(name="Fetch & Update SteamStore data")
+def update_steamstore_data():
+    fetcher = SteamStoreFetcher()
+    fetcher.run()
+
+
+@task(name="Clean New Steamstore database")
+def clean_steamstore_data():
+    cleaner = SteamDataClean()
+    cleaner.ingest()
+
+
+@flow(name="Steam ETL Pipeline", log_prints=True)
+def steam_etl():
+    update_metadata()
+    update_steamspy_data()
+    update_steamstore_data()
+    clean_steamstore_data()
+
+
+if __name__ == "__main__":
+    steam_etl.serve(name="steam_etl")

dag/requirements.txt

@@ -0,0 +1,3 @@
+prefect>=2.12.0
+steamstore_etl
+httpx