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

initial stage

This commit is contained in:
Giambattista Bloisi 2024-08-06 12:09:57 +02:00
parent e072311240
commit dac3849ced
1 changed files with 11 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
airflow/dags/remove_old_indexes.py
View File

@ -18,6 +18,13 @@ default_args = {
'retry_delay': timedelta(minutes=5), 'retry_delay': timedelta(minutes=5),
} }
managed_indexes = {'catalog_datasources', 'catalog_interoperability-records', 'catalog_providers',
'catalog_resource-interoperability-records', 'catalog_services', 'catalog_training-resources',
'datasource', 'grants', 'interoperability',
'organizations', 'persons', 'products',
'services', 'topics', 'training', 'venues'
}
@dag( @dag(
dag_id="remove_old_indexes", dag_id="remove_old_indexes",
@ -43,7 +50,9 @@ def remove_old_indexes():
print(json.dumps(aliases)) print(json.dumps(aliases))
print(json.dumps(indexes)) print(json.dumps(indexes))
# indexes referred by aliases
alias_index_names = {alias['index'] for alias in aliases} alias_index_names = {alias['index'] for alias in aliases}
# indexes ordered by timestamp
index_dict = {} index_dict = {}
for index in indexes: for index in indexes:
@ -51,6 +60,8 @@ def remove_old_indexes():
if '_' in index_name: if '_' in index_name:
base_name = '_'.join(index_name.split('_')[:-1]) base_name = '_'.join(index_name.split('_')[:-1])
timestamp = index_name.split('_')[-1] timestamp = index_name.split('_')[-1]
if not (base_name in managed_indexes and timestamp.isdigit()):
continue
if base_name not in index_dict: if base_name not in index_dict:
index_dict[base_name] = [] index_dict[base_name] = []
index_dict[base_name].append((index_name, timestamp)) index_dict[base_name].append((index_name, timestamp))