Pravki

Author: pushistickkk

students_folder/Zhukova_Mariya/lab5/airflow_folder/dags/collect_data.py

@@ -16,6 +16,7 @@
 
 
 def calculate_and_save_metrics(run_date: str):
+    """Рассчитывает метрики за конкретную дату из контекста Airflow"""
     print(f"Обработка метрик за дату: {run_date}")
 
     engine = create_engine("postgresql+psycopg2://airflow:airflow@postgres/airflow")
@@ -44,35 +45,17 @@ def calculate_and_save_metrics(run_date: str):
     print(f"Метрики за {run_date} успешно обновлены. Общая сумма продаж: {total_amount}")
 
 
-def collect_sales_per_day():
-    """Обработка всех дат из таблицы sales"""
-    engine = create_engine("postgresql+psycopg2://airflow:airflow@postgres/airflow")
-    
-    # Получаем все уникальные даты из sales
-    with engine.connect() as conn:
-        result = conn.execute(text("""
-            SELECT DISTINCT sale_date 
-            FROM market.sales 
-            ORDER BY sale_date
-        """))
-        dates = [row[0] for row in result]
-    
-    print(f"Найдены даты для обработки: {dates}")
-    
-    # Обрабатываем каждую дату
-    for date in dates:
-        calculate_and_save_metrics(str(date))
-
 with DAG(dag_id='collect_data',
          default_args=ARGS,
-         schedule_interval='@once', # или daily - если ежедневно
+         schedule_interval='@daily',
          max_active_runs=1,
          start_date=datetime(2025, 3, 20),
          catchup=True,
          tags=['lab5']) as dag:
 
-    t_process_all = PythonOperator(
-        task_id='collect_sales_per_day',
+    t_process_daily = PythonOperator(
+        task_id='calculate_daily_metrics',
         dag=dag,
-        python_callable=collect_sales_per_day
+        python_callable=calculate_and_save_metrics,
+        op_kwargs={"run_date": "{{ ds }}"}  # Передаем дату из контекста Airflow
     )

students_folder/Zhukova_Mariya/lab5/text_for_lab7.txt

@@ -0,0 +1,2 @@
+2-3 объекта и 3-4 эндпоинта
+