add kns_query + misc. minor improvements

Author: OriHoch

airflow/.gitignore

@@ -4,3 +4,4 @@ venv
 data
 *.egg-info
 .airflow
+.venv

airflow/README.md

@@ -12,28 +12,25 @@ install Airflow unless you want to check some Airflow specific detail.
 
 Prerequisites:
 
-* Python 3.8
+* Python 3.8 + [uv](https://pypi.org/project/uv/)
 * Docker Compose
 
 Create virtualenv and install dependencies
 
 ```
-python3.8 -m venv venv &&\
-. venv/bin/activate &&\
-pip install --upgrade pip setuptools wheel &&\
-pip install -e .
+uv sync
 ```
 
 Start a Database:
 
 ```
-docker-compose up -d db
+docker compose up -d db
 ```
 
 Run commands from the CLI:
 
 ```
-knesset-data-pipelines --help
+uv run knesset-data-pipelines --help
 ```
 
 Depending on the specific command, you will probably need to run dependant pipelines or download some packages or

airflow/docker-compose.yaml renamed to airflow/compose.yaml

@@ -1,5 +1,3 @@
-version: '2.4'
-
 services:
 
   db:

airflow/knesset_data_pipelines/run_pipeline.py

@@ -39,15 +39,26 @@ class RequestThrottledException(Exception):
     pass
 
 
+def post_process_airflow_pipeline(pipeline_path, pipeline_name, pipeline):
+    if 'schemas-bucket' not in pipeline:
+        pipeline['schemas-bucket'] = pipeline_path
+    return pipeline
+
+
 def get_pipeline_spec(pipeline_id):
     *pipeline_path, pipeline_name = pipeline_id.split("/")
     pipeline_path = '/'.join(pipeline_path)
     source_spec_yaml = os.path.join(config.KNESSET_DATA_PIPELINES_ROOT_DIR, pipeline_path, 'knesset.source-spec.yaml')
-    assert os.path.exists(source_spec_yaml)
-    with open(source_spec_yaml) as f:
-        source_spec = yaml.safe_load(f)
-    pipeline = source_spec[pipeline_name]
-    return pipeline_name, pipeline
+    if os.path.exists(source_spec_yaml):
+        with open(source_spec_yaml) as f:
+            source_spec = yaml.safe_load(f)
+        if pipeline_name in source_spec:
+            return pipeline_name, source_spec[pipeline_name]
+    source_spec_yaml = os.path.join(config.KNESSET_DATA_PIPELINES_ROOT_DIR, 'airflow/pipelines', pipeline_path, f'{pipeline_name}.yaml')
+    if os.path.exists(source_spec_yaml):
+        with open(source_spec_yaml) as f:
+            return pipeline_name, post_process_airflow_pipeline(pipeline_path, pipeline_name, yaml.safe_load(f))
+    raise Exception(f'pipeline {pipeline_id} not found')
 
 
 def get_pipeline_params(pipeline_name, pipeline):
@@ -410,6 +421,13 @@ def _iterate_pipelines(filter_pipeline_ids=None):
                 continue
             pipeline_name, pipeline = get_pipeline_spec(pipeline_id)
             yield pipeline_id, pipeline_name, pipeline
+    for pipeline_yaml in glob(os.path.join(config.KNESSET_DATA_PIPELINES_ROOT_DIR, 'airflow/pipelines/**/*.yaml'), recursive=True):
+        pipeline_path, pipeline_name = pipeline_yaml.split('/')[-2:]
+        pipeline_name = pipeline_name.replace('.yaml', '')
+        pipeline_id = f'{pipeline_path}/{pipeline_name}'
+        with open(pipeline_yaml) as f:
+            pipeline = post_process_airflow_pipeline(pipeline_path, pipeline_name, yaml.safe_load(f))
+        yield pipeline_id, pipeline_name, pipeline
 
 
 def list_pipelines(full=False, filter_pipeline_ids=None, all_=False, with_dependencies=False):

airflow/pipelines/knesset/kns_query.yaml

@@ -0,0 +1,44 @@
+pipeline-type: knesset dataservice
+dataservice-parameters:
+  service-name: api
+  method-name: "KNS_Query"
+  fields:
+    QueryID:
+      source: "{name}"
+      type: integer
+    Number:
+      source: "{name}"
+      type: integer
+    KnessetNum:
+      source: "{name}"
+      type: string
+    Name:
+      source: "{name}"
+      type: string
+    TypeID:
+      source: "{name}"
+      type: integer
+    TypeDesc:
+      source: "{name}"
+      type: string
+    StatusID:
+      source: "{name}"
+      type: integer
+    PersonID:
+      source: "{name}"
+      type: integer
+    GovMinistryID:
+      source: "{name}"
+      type: integer
+    SubmitDate:
+      source: "{name}"
+      type: datetime
+    ReplyMinisterDate:
+      source: "{name}"
+      type: datetime
+    ReplyDatePlanned:
+      source: "{name}"
+      type: datetime
+    LastUpdatedDate:
+      source: "{name}"
+      type: datetime

airflow/pyproject.toml

@@ -0,0 +1,29 @@
+[project]
+name = "knesset-data-pipelines"
+version = "0.1.0"
+requires-python = "==3.8.*"
+dependencies = [
+    "SQLAlchemy==1.4.46",
+    "click==8.1.3",
+    "ruamel.yaml==0.17.21",
+    "python-dotenv==0.21.1",
+    "dataflows==0.3.16",
+    "requests[socks]==2.28.2",
+    "beautifulsoup4==4.11.2",
+    "google-cloud-storage==2.7.0",
+    "google-api-python-client==2.90.0",
+    "apache-airflow-providers-cncf-kubernetes==7.2.0",
+    "pyquery==1.4.0",
+]
+
+#[dependency-groups]
+#dev = [
+#    "pytest>=8.3.5",
+#]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project.scripts]
+knesset-data-pipelines = "knesset_data_pipelines.cli:main"