Chore: Implement suggestions by CodeRabbit (periodic tasks)

Author: amotl

operation/periodic-task/README.md

@@ -54,7 +54,7 @@ python {cron.py,worker.py}
 ```
 
 
-[Apache Airflow]: https://airflow.apache.org/
+[Airflow]: https://airflow.apache.org/
 [APScheduler]: https://pypi.org/project/APScheduler/
 [cron]: https://en.wikipedia.org/wiki/Cron
 [crontab guru]: https://crontab.guru/

operation/periodic-task/airflow/README.md

@@ -30,28 +30,28 @@ Navigate to http://localhost:8080/.
 
 If your DAGs don't start running, check for any hard errors.
 ```shell
-docker exec -it airflow-scheduler-1 sh -c 'airflow dags list-import-errors'
+docker compose exec scheduler sh -c 'airflow dags list-import-errors'
 ```
 
 List available DAGs.
 ```shell
-docker exec -it airflow-scheduler-1 sh -c 'airflow dags list'
+docker compose exec scheduler sh -c 'airflow dags list'
 ```
 
 List runs of specific DAG.
 ```shell
-docker exec -it airflow-scheduler-1 sh -c 'airflow dags list-runs sys_summits'
-docker exec -it airflow-scheduler-1 sh -c 'airflow dags list-runs import_export'
+docker compose exec scheduler sh -c 'airflow dags list-runs sys_summits'
+docker compose exec scheduler sh -c 'airflow dags list-runs import_export'
 ```
 
 Invoke DAG workflow manually.
 ```shell
-docker exec -it airflow-scheduler-1 sh -c 'airflow dags trigger import_export'
+docker compose exec scheduler sh -c 'airflow dags trigger import_export'
 ```
 
 Check state of DAG.
 ```shell
-docker exec -it airflow-scheduler-1 sh -c 'airflow tasks state export_job export_to_s3 scheduled__2021-11-11T00:00:11+00:00'
+docker compose exec scheduler sh -c 'airflow tasks state export_job export_to_s3 scheduled__2021-11-11T00:00:11+00:00'
 ```
 
 

operation/periodic-task/airflow/compose.yaml

@@ -2,7 +2,7 @@ services:
 
   # CrateDB database.
   cratedb:
-    image: docker.io/crate/crate:${CRATEDB_VERSION:-latest}
+    image: docker.io/crate:${CRATEDB_VERSION:-latest}
     command: >
       crate
         '-Cdiscovery.type=single-node'

operation/periodic-task/airflow/workflows.py

@@ -27,7 +27,7 @@
     dag_id="sys_summits",
     description="Example DAG for plain-text SQL task.",
     default_args={"conn_id": "cratedb_connection"},
-    schedule="*/10 * * * * * *",  # @continuous, @daily
+    schedule="*/1 * * * *",  # @continuous, @daily
     catchup=False,
     max_active_runs=1,
 ):
@@ -44,7 +44,7 @@ def execute_query_taskflow():
     dag_id="import_export",
     description="Example DAG for SQLExecuteQueryOperator",
     default_args={"conn_id": "cratedb_connection"},
-    schedule="*/30 * * * * * *",  # @continuous, @daily
+    schedule="*/5 * * * *",  # @continuous, @daily
     catchup=False,
     max_active_runs=1,
 ):
@@ -61,7 +61,7 @@ def execute_query_taskflow():
     }
 
     ddl = SQLExecuteQueryOperator(
-        task_id=f"submit_ddl",
+        task_id="submit_ddl",
         conn_id="cratedb_connection",
         sql="""
         CREATE TABLE IF NOT EXISTS {{params.table}} (
@@ -76,7 +76,7 @@ def execute_query_taskflow():
     )
 
     insert = SQLExecuteQueryOperator(
-        task_id=f"import_from_http",
+        task_id="import_from_http",
         conn_id="cratedb_connection",
         sql="""
         COPY {{params.table}}
@@ -88,7 +88,7 @@ def execute_query_taskflow():
     )
 
     export = SQLExecuteQueryOperator(
-        task_id=f"export_to_s3",
+        task_id="export_to_s3",
         conn_id="cratedb_connection",
         sql="""
         COPY {{params.table}}
@@ -114,7 +114,7 @@ def execute_query_taskflow():
     dag_id="export_deltalake",
     description="Example DAG for CrateDB Toolkit I/O",
     default_args={"conn_id": "cratedb_connection"},
-    schedule="*/30 * * * * * *",  # @continuous, @daily
+    schedule="*/5 * * * *",  # @continuous, @daily
     catchup=False,
     max_active_runs=1,
 ):

operation/periodic-task/plombery/Dockerfile

@@ -17,7 +17,6 @@ COPY requirements.txt .
 RUN --mount=type=cache,id=uv-${TARGETARCH}${TARGETVARIANT},target=/root/.cache/uv \
     uv pip install --requirements=requirements.txt
 
-# Add programs to be scheduled periodically to the `dags` folder.
-# Their schedule is defined inline with the program source.
+# Add the workflow program to be scheduled periodically.
 COPY workflow.py .
 CMD ["python", "workflow.py"]

operation/periodic-task/plombery/compose.yaml

@@ -1,7 +1,7 @@
 services:
 
   cratedb:
-    image: docker.io/crate/crate:${CRATEDB_VERSION:-latest}
+    image: docker.io/crate:${CRATEDB_VERSION:-latest}
     command: >
       crate
         '-Cdiscovery.type=single-node'

operation/periodic-task/plombery/requirements.txt

@@ -1,2 +1,2 @@
-plombery
-sqlalchemy-cratedb
+plombery>=0.4.0,<1.0
+sqlalchemy-cratedb>=0.40.0,<1.0

operation/periodic-task/prefect/Dockerfile

@@ -17,7 +17,6 @@ COPY requirements.txt .
 RUN --mount=type=cache,id=uv-${TARGETARCH}${TARGETVARIANT},target=/root/.cache/uv \
     uv pip install --requirements=requirements.txt
 
-# Add programs to be scheduled periodically to the `dags` folder.
-# Their schedule is defined inline with the program source.
+# Add the workflow program to be scheduled periodically.
 COPY workflow.py .
 CMD ["python", "workflow.py"]

operation/periodic-task/prefect/compose.yaml

@@ -1,7 +1,7 @@
 services:
 
   cratedb:
-    image: docker.io/crate/crate:${CRATEDB_VERSION:-latest}
+    image: docker.io/crate:${CRATEDB_VERSION:-latest}
     command: >
       crate
         '-Cdiscovery.type=single-node'

operation/periodic-task/prefect/requirements.txt

@@ -1,2 +1,2 @@
-prefect
-sqlalchemy-cratedb
+prefect>=3.0.0,<4.0
+sqlalchemy-cratedb>=0.40.0,<1.0