airflow/dags/example_dags/__init__.py

Empty file.

airflow/dags/example_dags/print/__init__.py

Empty file.

airflow/dags/example_dags/print/dag.py

-Original file line number
+Diff line change
@@ -0,0 +1,25 @@
+    """Example DAG demonstrating the usage of the BashOperator."""
+    from __future__ import annotations
+    import logging
+    from datetime import datetime
+    from airflow.decorators import dag
+    from example_dags.print.tasks import print_context, print_finish, print_params
+    logger = logging.getLogger(__name__)
+    @dag(
+        schedule="@daily",
+        start_date=datetime(2021, 12, 1),
+        catchup=False,
+        dag_id="print",
+    )
+    def taskflow():
+        r1 = print_context()
+        r2 = print_params(r1)
+        print_finish(r2)
+    taskflow()

airflow/dags/example_dags/print/tasks/__init__.py

-Original file line number
+Diff line change
@@ -0,0 +1,3 @@
+    from .print_context import print_context
+    from .print_finish import print_finish
+    from .print_params import print_params

airflow/dags/example_dags/print/tasks/print_context.py

-Original file line number
+Diff line change
@@ -0,0 +1,12 @@
+    import logging
+    from pprint import pformat
+    from airflow.decorators import task
+    logger = logging.getLogger(__name__)
+    @task
+    def print_context(**context) -> dict:
+        logger.info("PARAMS: " + pformat(context))
+        return context.get("params", {}) or {"empty": True}

airflow/dags/example_dags/print/tasks/print_finish.py

-Original file line number
+Diff line change
@@ -0,0 +1,10 @@
+    import logging
+    from airflow.decorators import task
+    logger = logging.getLogger(__name__)
+    @task
+    def print_finish(data: str):
+        logger.info(data.upper())

airflow/dags/example_dags/print/tasks/print_params.py

-Original file line number
+Diff line change
@@ -0,0 +1,16 @@
+    import logging
+    from pprint import pformat
+    from airflow.decorators import task
+    logger = logging.getLogger(__name__)
+    @task
+    def print_params(data: dict, **kwargs) -> str:
+        logger.info("PARAMS: " + pformat(data))
+        logger.info("KWARGS: " + pformat(kwargs))
+        return "finish"
+    ""

airflow/dags/print_operator.py

This file was deleted.

feat: splited print dag tasks #866

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open

SunnyCapt wants to merge 2 commits into main from feat/print-dag-example

-Original file line number
+Diff line change
@@ -0,0 +1,25 @@
+    """Example DAG demonstrating the usage of the BashOperator."""
+    from __future__ import annotations
+    import logging
+    from datetime import datetime
+    from airflow.decorators import dag
+    from example_dags.print.tasks import print_context, print_finish, print_params
+    logger = logging.getLogger(__name__)
+    @dag(
+        schedule="@daily",
+        start_date=datetime(2021, 12, 1),
+        catchup=False,
+        dag_id="print",
+    )
+    def taskflow():
+        r1 = print_context()
+        r2 = print_params(r1)
+        print_finish(r2)
+    taskflow()

-Original file line number
+Diff line change
@@ -0,0 +1,3 @@
+    from .print_context import print_context
+    from .print_finish import print_finish
+    from .print_params import print_params

-Original file line number
+Diff line change
@@ -0,0 +1,12 @@
+    import logging
+    from pprint import pformat
+    from airflow.decorators import task
+    logger = logging.getLogger(__name__)
+    @task
+    def print_context(**context) -> dict:
+        logger.info("PARAMS: " + pformat(context))
+        return context.get("params", {}) or {"empty": True}

-Original file line number
+Diff line change
@@ -0,0 +1,10 @@
+    import logging
+    from airflow.decorators import task
+    logger = logging.getLogger(__name__)
+    @task
+    def print_finish(data: str):
+        logger.info(data.upper())

-Original file line number
+Diff line change
@@ -0,0 +1,16 @@
+    import logging
+    from pprint import pformat
+    from airflow.decorators import task
+    logger = logging.getLogger(__name__)
+    @task
+    def print_params(data: dict, **kwargs) -> str:
+        logger.info("PARAMS: " + pformat(data))
+        logger.info("KWARGS: " + pformat(kwargs))
+        return "finish"
+    ""

Provide feedback