From 2c2404e975ed4765e4594f4d08cb5ddb46a5cbe8 Mon Sep 17 00:00:00 2001 From: schepa Date: Fri, 24 Apr 2026 16:48:35 +0300 Subject: [PATCH 1/3] fix: update outlets and schedule datasets trigger to ODS_USERDATA --- dags/ods/user/info.py | 2 +- dags/ods/user/user_to_back.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/dags/ods/user/info.py b/dags/ods/user/info.py index f64578b..a98b28e 100644 --- a/dags/ods/user/info.py +++ b/dags/ods/user/info.py @@ -31,7 +31,7 @@ # порядок датасетов здесь важен! # см. info.sql и https://airflow.apache.org/docs/apache-airflow/stable/templates-ref.html#variables inlets=[Dataset("STG_USERDATA.info"), Dataset("STG_USERDATA.param"), Dataset("STG_UNION_MEMBER.union_member")], - outlets=[Dataset("ODS_USER_INFO.info")], + outlets=[Dataset("ODS_USERDATA")], params={"tablename": "info"}, ) # run_sql_encrypted = PostgresOperator( diff --git a/dags/ods/user/user_to_back.py b/dags/ods/user/user_to_back.py index 56b4ca7..98061bc 100644 --- a/dags/ods/user/user_to_back.py +++ b/dags/ods/user/user_to_back.py @@ -334,7 +334,7 @@ def remove_non_union_members_from_union_group(union_members_ids: list): with DAG( dag_id="union_member_to_backend", - schedule=[Dataset("ODS_USERDATA.student_id")], + schedule=[Dataset("ODS_USERDATA")], start_date=datetime.datetime(start_year, start_month, start_day), catchup=False, tags=["ods", "userdata", "union_member", "backend"], From c731854a5af4ded56e5e18717f89e4ac525c2d3a Mon Sep 17 00:00:00 2001 From: schepa Date: Sat, 25 Apr 2026 20:28:37 +0300 Subject: [PATCH 2/3] add inlets and fix logging in ods.user.user_to_back, change DAG id in ods.user.info.py --- dags/ods/user/info.py | 4 ++-- dags/ods/user/user_to_back.py | 5 +++-- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/dags/ods/user/info.py b/dags/ods/user/info.py index a98b28e..efe354c 100644 --- a/dags/ods/user/info.py +++ b/dags/ods/user/info.py @@ -12,7 +12,7 @@ with DAG( - dag_id="ODS_USER_INFO.info", + dag_id="ODS_USERDATA", start_date=datetime(2024, 10, 1), schedule=[Dataset("STG_USERDATA.info"), Dataset("STG_USERDATA.param"), Dataset("STG_UNION_MEMBER.union_member")], catchup=False, @@ -46,6 +46,6 @@ # Dataset("STG_USERDATA.param"), # Dataset("STG_UNION_MEMBER.union_member"), # ], - # outlets=[Dataset("ODS_USER_INFO.encrypted_info")], + # outlets=[Dataset("ODS_USERDATA.encrypted_info")], # params={"tablename": "encrypted_info"}, # ) TODO сделать мерж юсердаты шифрованной diff --git a/dags/ods/user/user_to_back.py b/dags/ods/user/user_to_back.py index 98061bc..ab305d9 100644 --- a/dags/ods/user/user_to_back.py +++ b/dags/ods/user/user_to_back.py @@ -273,10 +273,10 @@ def post_members_to_union_group_to_backend(union_members_ids: list): json=data, ) if response.status_code == 200: - logging.info(f"Updated group with union members {union_members_ids} from dwh database") + logging.info(f"Updated group with union member {union_member} from dwh database") else: logging.error( - f"Update group with union members {union_members_ids} copy to backend failed with code: {response.status_code}\n Response text: {response.text}" + f"Update group with union member {union_member} copy to backend failed with code: {response.status_code}\n Response text: {response.text}" ) except Exception as e: logging.error(f"Error sending data to backend: {str(e)}") @@ -335,6 +335,7 @@ def remove_non_union_members_from_union_group(union_members_ids: list): with DAG( dag_id="union_member_to_backend", schedule=[Dataset("ODS_USERDATA")], + inlets = [Dataset("ODS_USERDATA")], start_date=datetime.datetime(start_year, start_month, start_day), catchup=False, tags=["ods", "userdata", "union_member", "backend"], From 20c4198b53035f6d8844b238222b93aef22be11c Mon Sep 17 00:00:00 2001 From: schepa Date: Sat, 25 Apr 2026 20:48:15 +0300 Subject: [PATCH 3/3] fix lint --- dags/ods/user/user_to_back.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dags/ods/user/user_to_back.py b/dags/ods/user/user_to_back.py index ab305d9..5230add 100644 --- a/dags/ods/user/user_to_back.py +++ b/dags/ods/user/user_to_back.py @@ -335,7 +335,7 @@ def remove_non_union_members_from_union_group(union_members_ids: list): with DAG( dag_id="union_member_to_backend", schedule=[Dataset("ODS_USERDATA")], - inlets = [Dataset("ODS_USERDATA")], + inlets=[Dataset("ODS_USERDATA")], start_date=datetime.datetime(start_year, start_month, start_day), catchup=False, tags=["ods", "userdata", "union_member", "backend"],