Skip to content

Commit b6dd76f

Browse files
OriHochOriHoch
committed
add missing tables/fields
1 parent 6a22015 commit b6dd76f

23 files changed

Lines changed: 823 additions & 3 deletions

‎airflow/dags/kns_odata.py‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,22 @@
1+
from airflow import DAG
2+
from airflow.utils.dates import days_ago
3+
from airflow.operators.python import PythonOperator
4+
5+
from knesset_data_pipelines import kns_odata
6+
7+
8+
dag_kwargs = dict(
9+
default_args={
10+
'owner': 'airflow',
11+
},
12+
schedule_interval='10 0 * * *',
13+
start_date=days_ago(1),
14+
catchup=False,
15+
)
16+
17+
18+
with DAG('kns_odata', **dag_kwargs) as dag:
19+
PythonOperator(
20+
python_callable=kns_odata.compare_parliamentinfo_tables_pipelines,
21+
task_id='compare_parliamentinfo_tables_pipelines'
22+
)

‎airflow/knesset_data_pipelines/cli.py‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
import importlib
2+
import json
23

34
import click
45
import dotenv
@@ -35,11 +36,20 @@ def run(**kwargs):
3536

3637
@main.command('list')
3738
@click.option('--filter-pipeline-ids', help='comma separated list of pipeline ids to filter')
39+
@click.option('--full', is_flag=True)
3840
def list_(**kwargs):
3941
"""List all pipelines"""
4042
from .run_pipeline import list_pipelines
41-
for error, pipeline_id, pipeline_dependencies, pipeline_schedule in list_pipelines(**kwargs, all_=True, with_dependencies=True):
42-
print(f'- {pipeline_id}{" (e)" if error else ""} (dependencies: {pipeline_dependencies}){" (scheduled)" if pipeline_schedule else ""}')
43+
if kwargs.get('full'):
44+
print('[')
45+
for i, data in enumerate(list_pipelines(**kwargs)):
46+
if i > 0:
47+
print(',')
48+
print(json.dumps(data, indent=2))
49+
print(']')
50+
else:
51+
for error, pipeline_id, pipeline_dependencies, pipeline_schedule in list_pipelines(**kwargs, all_=True, with_dependencies=True):
52+
print(f'- {pipeline_id}{" (e)" if error else ""} (dependencies: {pipeline_dependencies}){" (scheduled)" if pipeline_schedule else ""}')
4353

4454

4555
@main.command()

‎airflow/knesset_data_pipelines/kns_odata.py‎

Lines changed: 150 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,150 @@
1+
import os
2+
from textwrap import dedent
3+
4+
import requests
5+
6+
from .run_pipeline import list_pipelines
7+
8+
9+
IGNORE_FIELD_TYPE_ERRORS = {
10+
'KNS_BillHistoryInitiator': [
11+
'StartDate', # field type is string instead of datetime
12+
]
13+
}
14+
15+
16+
def get_table_res_field(field):
17+
field_types = {
18+
'Edm.Int64': 'integer', # 123
19+
'Edm.Int32': 'integer', # 123
20+
'Edm.Int16': 'integer', # 123
21+
'Edm.Byte': 'integer', # 123
22+
'Edm.DateTimeOffset': 'datetime', # "2016-02-28T10:22:10.843+02:00"
23+
None: 'string', # "string",
24+
'Edm.Boolean': 'boolean',
25+
}
26+
assert field.get("$Type") in field_types, f"unknown field type {field.get('$Type')}"
27+
return {
28+
'type': field_types[field.get("$Type")]
29+
}
30+
31+
32+
def get_parliamentinfo_tables():
33+
data = requests.get('https://knesset.gov.il/OdataV4/ParliamentInfo/$metadata?$format=json').json()
34+
res = {}
35+
for table_name, fields in data['OdataService.DAL.ParliamentInfo'].items():
36+
assert fields.pop("$Kind") == 'EntityType'
37+
keys = fields.pop("$Key") or []
38+
assert len(keys) == 1, f"unexpected number of keys in table {table_name}: {keys}"
39+
res[table_name] = {
40+
'primary_key': keys[0],
41+
'fields': {
42+
field_name: get_table_res_field(field) for field_name, field in fields.items() if field.get('$Kind') is None
43+
}
44+
}
45+
return res
46+
47+
48+
def get_pipelines_res_field(field):
49+
assert field['source'] == '{name}', f'unexpected source {field["source"]}'
50+
return {
51+
'type': field['type'],
52+
'primary_key': bool(field.get('primaryKey')),
53+
}
54+
55+
56+
def get_parliamentinfo_pipelines():
57+
res = {}
58+
for pipeline in list_pipelines(full=True):
59+
if pipeline.get('dataservice_params') and pipeline['dataservice_params'].get('service-name') == 'api':
60+
res[pipeline['dataservice_params']['method-name']] = {}
61+
res[pipeline['dataservice_params']['method-name']]['fields'] = {
62+
name: get_pipelines_res_field(field) for name, field in pipeline['dataservice_params']['fields'].items()
63+
}
64+
primary_keys = [name for name, field in res[pipeline['dataservice_params']['method-name']]['fields'].items() if field.get('primary_key')]
65+
assert len(primary_keys) <= 1, f"unexpected number of primary keys in pipeline {pipeline['pipeline_id']}: {primary_keys}"
66+
res[pipeline['dataservice_params']['method-name']]['primary_key'] = primary_keys[0] if primary_keys else None
67+
return res
68+
69+
70+
class MissingPipeline:
71+
72+
def __init__(self, table_name, table):
73+
self.table_name = table_name
74+
self.table = table
75+
76+
def __str__(self):
77+
res = f'table {self.table_name} is missing from pipelines\n'
78+
filename = f'knesset/{self.table_name.lower()}.yaml'
79+
filecontent = dedent(f'''
80+
pipeline-type: knesset dataservice
81+
dataservice-parameters:
82+
service-name: api
83+
method-name: "{self.table_name}"
84+
fields:
85+
''')
86+
for field_name, field in self.table['fields'].items():
87+
filecontent += f' {field_name}:\n'
88+
filecontent += f' source: "{{name}}"\n'
89+
filecontent += f' type: "{field["type"]}"\n'
90+
if field_name == self.table['primary_key']:
91+
filecontent += f' primaryKey: true\n'
92+
with open(os.path.join(os.path.dirname(__file__), '..', 'pipelines', filename), 'w') as f:
93+
f.write(filecontent)
94+
return res
95+
96+
97+
class IncompletePipeline:
98+
99+
def __init__(self, table_name, table, pipeline):
100+
self.table_name = table_name
101+
self.table = table
102+
self.pipeline = pipeline
103+
self.missing_fields = []
104+
self.wrong_type_fields = []
105+
106+
def __str__(self):
107+
res = f'pipeline for table {self.table_name} is incomplete:\n'
108+
if len(self.missing_fields) > 0:
109+
res += ' - missing fields:\n'
110+
for field_name in self.missing_fields:
111+
res += f'{field_name}: {{source: "{{name}}", type: "{self.table["fields"][field_name]["type"]}"}}\n'
112+
if len(self.wrong_type_fields) > 0:
113+
res += ' - wrong type fields:\n'
114+
for field_name in self.wrong_type_fields:
115+
res += f'{field_name}: {{source: "{{name}}", type: "{self.table["fields"][field_name]["type"]}"}}\n'
116+
return res
117+
118+
119+
def compare_parliamentinfo_tables_pipelines():
120+
fix_objects = []
121+
tables = get_parliamentinfo_tables()
122+
pipelines = get_parliamentinfo_pipelines()
123+
for table_name, table in tables.items():
124+
if table_name not in pipelines:
125+
fix_objects.append(MissingPipeline(table_name, table))
126+
continue
127+
pipeline = pipelines[table_name]
128+
incomplete_pipeline = IncompletePipeline(table_name, table, pipeline)
129+
for field_name, field in table['fields'].items():
130+
if field_name not in pipeline['fields']:
131+
if table['primary_key'] == field_name and pipeline['primary_key']:
132+
field_name = pipeline['primary_key']
133+
else:
134+
incomplete_pipeline.missing_fields.append(field_name)
135+
continue
136+
pipeline_field = pipeline['fields'][field_name]
137+
if field['type'] != pipeline_field['type'] and field_name not in IGNORE_FIELD_TYPE_ERRORS.get(table_name, []):
138+
incomplete_pipeline.wrong_type_fields.append(field_name)
139+
if len(incomplete_pipeline.missing_fields) > 0 or len(incomplete_pipeline.wrong_type_fields) > 0:
140+
fix_objects.append(incomplete_pipeline)
141+
if len(fix_objects) > 0:
142+
for fix_object in fix_objects:
143+
print(fix_object)
144+
raise Exception('need to fix the pipelines')
145+
else:
146+
print('all pipelines are complete and correct')
147+
148+
149+
if __name__ == '__main__':
150+
compare_parliamentinfo_tables_pipelines()

‎airflow/pipelines/knesset/kns_agenda.yaml‎

Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,67 @@
1+
2+
pipeline-type: knesset dataservice
3+
dataservice-parameters:
4+
service-name: api
5+
method-name: "KNS_Agenda"
6+
fields:
7+
Id:
8+
source: "{name}"
9+
type: "integer"
10+
primaryKey: true
11+
Number:
12+
source: "{name}"
13+
type: "integer"
14+
ClassificationID:
15+
source: "{name}"
16+
type: "integer"
17+
ClassificationDesc:
18+
source: "{name}"
19+
type: "string"
20+
LeadingAgendaID:
21+
source: "{name}"
22+
type: "integer"
23+
KnessetNum:
24+
source: "{name}"
25+
type: "integer"
26+
Name:
27+
source: "{name}"
28+
type: "string"
29+
SubTypeID:
30+
source: "{name}"
31+
type: "integer"
32+
SubTypeDesc:
33+
source: "{name}"
34+
type: "string"
35+
StatusID:
36+
source: "{name}"
37+
type: "integer"
38+
InitiatorPersonID:
39+
source: "{name}"
40+
type: "integer"
41+
GovRecommendationID:
42+
source: "{name}"
43+
type: "integer"
44+
GovRecommendationDesc:
45+
source: "{name}"
46+
type: "string"
47+
PresidentDecisionDate:
48+
source: "{name}"
49+
type: "datetime"
50+
PostopenmentReasonID:
51+
source: "{name}"
52+
type: "integer"
53+
PostopenmentReasonDesc:
54+
source: "{name}"
55+
type: "string"
56+
CommitteeID:
57+
source: "{name}"
58+
type: "integer"
59+
RecommendCommitteeID:
60+
source: "{name}"
61+
type: "integer"
62+
MinisterPersonID:
63+
source: "{name}"
64+
type: "integer"
65+
LastUpdatedDate:
66+
source: "{name}"
67+
type: "datetime"

‎airflow/pipelines/knesset/kns_broadcastcommittesession.yaml‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
2+
pipeline-type: knesset dataservice
3+
dataservice-parameters:
4+
service-name: api
5+
method-name: "KNS_BroadcastCommitteSession"
6+
fields:
7+
Id:
8+
source: "{name}"
9+
type: "integer"
10+
primaryKey: true
11+
BroadcastId:
12+
source: "{name}"
13+
type: "integer"
14+
BroadcastUrl:
15+
source: "{name}"
16+
type: "string"

‎airflow/pipelines/knesset/kns_documentagenda.yaml‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
2+
pipeline-type: knesset dataservice
3+
dataservice-parameters:
4+
service-name: api
5+
method-name: "KNS_DocumentAgenda"
6+
fields:
7+
Id:
8+
source: "{name}"
9+
type: "integer"
10+
primaryKey: true
11+
AgendaID:
12+
source: "{name}"
13+
type: "integer"
14+
GroupTypeID:
15+
source: "{name}"
16+
type: "integer"
17+
GroupTypeDesc:
18+
source: "{name}"
19+
type: "string"
20+
ApplicationID:
21+
source: "{name}"
22+
type: "integer"
23+
ApplicationDesc:
24+
source: "{name}"
25+
type: "string"
26+
FilePath:
27+
source: "{name}"
28+
type: "string"
29+
LastUpdatedDate:
30+
source: "{name}"
31+
type: "datetime"

‎airflow/pipelines/knesset/kns_documentisraellaw.yaml‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
2+
pipeline-type: knesset dataservice
3+
dataservice-parameters:
4+
service-name: api
5+
method-name: "KNS_DocumentIsraelLaw"
6+
fields:
7+
Id:
8+
source: "{name}"
9+
type: "integer"
10+
primaryKey: true
11+
IsraelLawID:
12+
source: "{name}"
13+
type: "integer"
14+
GroupTypeID:
15+
source: "{name}"
16+
type: "integer"
17+
GroupTypeDesc:
18+
source: "{name}"
19+
type: "string"
20+
ApplicationID:
21+
source: "{name}"
22+
type: "integer"
23+
ApplicationDesc:
24+
source: "{name}"
25+
type: "string"
26+
FilePath:
27+
source: "{name}"
28+
type: "string"
29+
LastUpdatedDate:
30+
source: "{name}"
31+
type: "datetime"

‎airflow/pipelines/knesset/kns_documentquery.yaml‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
2+
pipeline-type: knesset dataservice
3+
dataservice-parameters:
4+
service-name: api
5+
method-name: "KNS_DocumentQuery"
6+
fields:
7+
Id:
8+
source: "{name}"
9+
type: "integer"
10+
primaryKey: true
11+
QueryID:
12+
source: "{name}"
13+
type: "integer"
14+
GroupTypeID:
15+
source: "{name}"
16+
type: "integer"
17+
GroupTypeDesc:
18+
source: "{name}"
19+
type: "string"
20+
ApplicationID:
21+
source: "{name}"
22+
type: "integer"
23+
ApplicationDesc:
24+
source: "{name}"
25+
type: "string"
26+
FilePath:
27+
source: "{name}"
28+
type: "string"
29+
LastUpdatedDate:
30+
source: "{name}"
31+
type: "datetime"

0 commit comments

Comments
 (0)