WIP: Dug data dictionary compatibility (and package refactor)

.github/workflows/build-push-dev-image.yml

@@ -6,6 +6,7 @@ on:
   push:
     branches:
       - develop
+      - package-refactor
     paths-ignore:
       - README.md
       - .old_cicd/*
@@ -83,4 +84,4 @@ jobs:
           containers.renci.org/${{ github.repository }}:develop
           containers.renci.org/${{ github.repository }}:${{ steps.vars.outputs.short_sha }}
         cache-from: type=registry,ref=${{ github.repository }}:buildcache-dev
-        cache-to: type=registry,ref=${{ github.repository }}:buildcache-dev,mode=max
+        cache-to: type=registry,ref=${{ github.repository }}:buildcache-dev,mode=max

Dockerfile

@@ -1,8 +1,8 @@
 # Use a Debian-based image for better compatibility
-FROM python:3.11.14-slim-trixie
+FROM python:3.12-slim-trixie
 
 # Set Airflow version and home directory
-ARG AIRFLOW_VERSION=3.1.5
+ARG AIRFLOW_VERSION=3.1.7
 ARG AIRFLOW_HOME=/opt/airflow
 
 # Environment variables
@@ -37,6 +37,10 @@ RUN pip install --no-cache-dir \
     "apache-airflow-providers-cncf-kubernetes" \
     --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-3.11.txt"
 
+# Fix auth rollback bug.
+RUN pip install --no-cache-dir \
+    "apache-airflow-providers-fab==3.3.0rc1"
+
 # Optional: install extra packages
 RUN pip install --no-cache-dir psycopg2-binary redis
 
@@ -46,7 +50,8 @@ RUN pip install -r /tmp/requirements.txt
 
 RUN rm /tmp/requirements.txt
 
-
+# COPY . /opt/roger
+# RUN pip install /opt/roger
 
 RUN apt-get purge -y --auto-remove \
     build-essential \
@@ -57,13 +62,17 @@ RUN apt-get purge -y --auto-remove \
     git && \
     apt-get clean
 
+RUN if [ -n "$ROGER_SOURCE" ]; then pip install -e $ROGER_SOURCE; fi
+
 # Set ownership
 RUN chown -R airflow:airflow ${AIRFLOW_HOME}
 
 # Switch to airflow user
 USER airflow
 WORKDIR ${AIRFLOW_HOME}
 
+ENV PYTHONPATH=/opt/airflow/dags/repo/src/
+
 # Expose Airflow webserver port
 EXPOSE 8080
 

bin/dug_annotate/Makefile

@@ -33,12 +33,12 @@ clean:
 	$(RM) -rf ${KGX_DIR}
 
 get_input_files:
-	$(TIME) python ${CLI_WRAPPER} -gd
+	$(TIME) roger -gd
 
 annotate_and_normalize:
-	$(TIME) python ${CLI_WRAPPER} -l
+	$(TIME) roger -l
 
 create_kgx_files:
-	$(TIME) python ${CLI_WRAPPER} -t
+	$(TIME) roger -t
 
 all: get_input_files annotate_and_normalize create_kgx_files

bin/dug_indexing/Makefile

@@ -19,7 +19,6 @@ THIS_MAKEFILE_PATH:=$(word $(words $(MAKEFILE_LIST)),$(MAKEFILE_LIST))
 THIS_DIR:=$(shell cd $(dir $(THIS_MAKEFILE_PATH));pwd)
 
 ROGER_HOME=${THIS_DIR}/../..
-CLI_WRAPPER=${ROGER_HOME}/cli.py
 
 # Override Roger data dir ENV
 INDEXING_DIR=${ROGERENV_DATA__ROOT}/dug/expanded_concepts
@@ -34,19 +33,19 @@ clean:
 	$(RM) -rf ${CRAWL_DIR}
 
 crawl_concepts:
-	$(TIME) python ${CLI_WRAPPER} -C
+	$(TIME) roger -C
 
 index_concepts: crawl_concepts
-	$(TIME) python ${CLI_WRAPPER} -ic
+	$(TIME) roger -ic
 
 index_variables:
-	$(TIME) python ${CLI_WRAPPER} -iv
+	$(TIME) roger -iv
 
 validate_indexed_concepts: index_concepts
-	$(TIME) python ${CLI_WRAPPER} -vc
+	$(TIME) roger -vc
 
 validate_indexed_variables: index_variables
-	$(TIME) python ${CLI_WRAPPER} -vv
+	$(TIME) roger -vv
 
 all: validate_indexed_concepts validate_indexed_variables
 

dags/annotate_and_index.py

@@ -9,8 +9,8 @@
 import os
 
 from airflow.models import DAG
-from airflow.operators.empty import EmptyOperator
-from airflow.operators.python import PythonOperator
+from airflow.providers.standard.operators.empty import EmptyOperator
+from airflow.providers.standard.operators.python import PythonOperator
 from roger.tasks import default_args, create_pipeline_taskgroup, logger, create_python_task
 
 env_enabled_datasets = os.getenv(
@@ -52,35 +52,5 @@
 
         init >> create_pipeline_taskgroup(dag, pipeline_class, config) >> finish
 
-
-
-
-with DAG(
-        dag_id='dag_test',
-        default_args=default_args,
-        params=
-            {
-                "repository_id": None,
-                "branch_name": None,
-                "commitid_from": None,
-                "commitid_to": None
-            },
-        # schedule_interval=None
-) as dag:
-
-    init = EmptyOperator(task_id="init", dag=dag)
-    finish = EmptyOperator(task_id="finish", dag=dag)
-
-    def print_context(ds=None, **kwargs):
-        print(">>>All kwargs")
-        print(kwargs)
-        print(">>>All ds")
-        print(ds)
-
-
-    init >> create_python_task(dag, "get_from_lakefs", print_context) >> finish
-
-    #run_this = PythonOperator(task_id="print_the_context", python_callable=print_context)
-
 if __name__ == "__main__":
-    dag.test()
+    dag.test()