giambattista.bloisi
/
lot1-kickoff
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

simple test DAG

This commit is contained in:
Giambattista Bloisi 2024-03-14 22:27:51 +01:00
parent c513072be9
commit 602fedc6cb
1 changed files with 4 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
airflow/dags/skg_if_pipeline.py
View File

@ -73,11 +73,11 @@ def skg_if_pipeline():
session = requests.Session()
session.auth = (OPENSEARCH_USER, OPENSEARCH_PASSWD)
response = session.delete(f'${OPENSEARCH_URL}/{entity}_index',
response = session.delete(f'{OPENSEARCH_URL}/{entity}_index',
verify=False)
response.raise_for_status()
response = session.put(f'${OPENSEARCH_URL}/{entity}_index',
response = session.put(f'{OPENSEARCH_URL}/{entity}_index',
data=json.dumps({
"settings": {
"index": {
@ -107,14 +107,14 @@ def skg_if_pipeline():
for line in buff:
data = json.loads(line)
response = session.post(
f'${OPENSEARCH_URL}/{entity}_index/_doc/' + requests.utils.quote(
f'{OPENSEARCH_URL}/{entity}_index/_doc/' + requests.utils.quote(
data['local_identifier'], safe='') + "?refresh=false",
data=json.dumps(data),
headers={"Content-Type": "application/json"},
verify=False)
response.raise_for_status()
response = session.post(
f'${OPENSEARCH_URL}/{entity}_index/_refresh',
f'{OPENSEARCH_URL}/{entity}_index/_refresh',
verify=False)
response.raise_for_status()