diff --git a/dags/ods/user/info.py b/dags/ods/user/info.py index f64578b..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, @@ -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( @@ -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 56b4ca7..5230add 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)}") @@ -334,7 +334,8 @@ 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")], + inlets=[Dataset("ODS_USERDATA")], start_date=datetime.datetime(start_year, start_month, start_day), catchup=False, tags=["ods", "userdata", "union_member", "backend"],