Multy user DB release

Author: EEnota

main.py

@@ -60,7 +60,7 @@ def parse_args():
         default=1
         )
     parser.add_argument(
-        "--serch_request_id",
+        "--search_request_id",
         type=int,
         default=1
     )
@@ -69,10 +69,10 @@ def parse_args():
 def init_all_tables(debug: bool) -> None:
     init_database()
     create_app_users_table()
-    create_articles_table()
-    create_request_stats_table()
     create_search_requests_table()
+    create_articles_table()
     create_user_news_table()
+    create_request_stats_table()
     if debug:
         create_news_tables()
 
@@ -85,7 +85,7 @@ def main()-> None:
         if args.debug:
             loaded = run_debug_pipeline(args.keyword, args.limit, args.page_size)
         else:
-            loaded = run_pipeline_for_web_user(args.user_id, args.search_request_id, args.ketword, args.limit, args.page_size)
+            loaded = run_pipeline_for_web_user(args.user_id, args.search_request_id, args.keyword, args.limit, args.page_size)
         logger.info("Pipline finished. loaded rows: %s", loaded)
     except Exception as e:
         logger.exception("pipeline failed: %s", e)

src/__init__.py

@@ -1,5 +1,5 @@
 from .extract import make_extract_debug, make_extract_web
 from .transform import transform_article_web, transform_article_debug
-from .load import load_news, load_web_pipeline
+from .load import load_news, load_web_pipeline, load_request_stats
 from .db import init_database, create_news_tables, create_app_users_table, create_search_requests_table, create_articles_table, create_user_news_table, create_request_stats_table
 from .pipeline import run_pipeline_for_web_user, run_debug_pipeline

src/load.py

@@ -5,6 +5,7 @@
 from typing import Optional
 BASE_DIR = (Path(__file__).resolve().parent.parent)/"data"/"clean"
 
+    
 def load_news(clean_news:str, max_rows: Optional[int] = None)-> int:
     LOAD_DIR = BASE_DIR/clean_news
     num_of_news = 0
@@ -95,29 +96,39 @@ def load_user_news(cur, user_id: int, search_request_id: int, article_id: int, k
     cur.execute(query,(user_id, search_request_id, article_id, keyword, fetched_at))
     return cur.rowcount
 
-def load_request_stats(cur, search_request_id: int, stats: dict) -> None:
+def load_request_stats(search_request_id: int, stats: dict) -> None:
     query = """
             INSERT INTO request_stats (
             search_request_id,
             income_articles,
             accepted_articles,
             rejected_articles,
             reasons_counts,
-            prime_reason
+            prime_reasons
             )
             VALUES (%s,%s,%s,%s,%s,%s)
+            ON CONFLICT (search_request_id) DO UPDATE
+            SET
+                income_articles = EXCLUDED.income_articles,
+                accepted_articles = EXCLUDED.accepted_articles,
+                rejected_articles = EXCLUDED.rejected_articles,
+                reasons_counts = EXCLUDED.reasons_counts,
+                prime_reasons = EXCLUDED.prime_reasons
             """
-    cur.execute(query, (
+    
+    with get_cursor(settings.db_news) as (conn, cur):
+        cur.execute(query, (
             search_request_id,
             stats["income_articles"],
             stats["accepted_articles"],
             stats["rejected_articles"],
             json.dumps(stats["reasons_counts"]),
             json.dumps(stats["prime_reason"])
         ))
-    return None
+        conn.commit()
+
 
-def load_web_pipeline(user_id: int, search_request_id: int, clean_data: list[dict], stats: dict) -> int:
+def load_web_pipeline(user_id: int, search_request_id: int, clean_data: list[dict]) -> int:
     loaded_count = 0
     with get_cursor(settings.db_news) as (conn, cur):
         for article in clean_data:
@@ -126,7 +137,6 @@ def load_web_pipeline(user_id: int, search_request_id: int, clean_data: list[dic
             article_id = upsert_article(cur,article)
             inserted = load_user_news(cur, user_id, search_request_id, article_id, keyword, fetched_at)
             loaded_count += inserted   
-        load_request_stats(cur, search_request_id, stats)
         conn.commit()
     return loaded_count
 

src/pipeline.py

@@ -5,12 +5,44 @@
     transform_article_web,
     transform_article_debug,
     load_web_pipeline,
-    load_news,
+    load_news, 
+    load_request_stats
 )
 
 logger = logging.getLogger(__name__)
+def make_empty_stats()->dict:
+    return {
+        "income_articles" : 0,
+        "accepted_articles" : 0,
+        "rejected_articles" : 0,
+        "reasons_counts" :{
+            "no_author":0,
+            "no_title":0,
+            "no_description":0,
+            "short_description":0,
+            "no_url":0
+        },
+        "prime_reason":{
+            "no_author":0,
+            "no_title":0,
+            "no_description":0,
+            "short_description":0,
+            "no_url":0
+        }
+    }
+
+def merge_stats(stats: dict, page_stats: dict)-> None:
+    stats["income_articles"] += page_stats["income_articles"]
+    stats["accepted_articles"] += page_stats["accepted_articles"]
+    stats["rejected_articles"] += page_stats["rejected_articles"]
+    for k,v in page_stats["reasons_counts"].items():
+        stats["reasons_counts"][k] += v
+    
+    for k,v in page_stats["prime_reason"].items():
+        stats["prime_reason"][k] += v
 
 def run_pipeline_for_web_user(user_id: int, search_request_id: int, key_word: str, limit: int, page_size: int) -> int:
+    statistic = make_empty_stats()
     num_of_news = 0
     page = 1
     while num_of_news < limit:
@@ -19,9 +51,11 @@ def run_pipeline_for_web_user(user_id: int, search_request_id: int, key_word: st
             logger.warning("there is no more artical")
             break
         clean_data, stats = transform_article_web(payload)
-        result_num_of_news = load_web_pipeline(user_id, search_request_id, clean_data, stats)
+        merge_stats(statistic, stats)
+        result_num_of_news = load_web_pipeline(user_id, search_request_id, clean_data)
         num_of_news += result_num_of_news
         page += 1
+    load_request_stats(search_request_id, statistic)
     logger.info(f"{num_of_news} news on key word {key_word} already aploaded")
     return num_of_news
 

src/worker.py

@@ -46,7 +46,7 @@ def mark_as_success(search_request_id: int) -> None:
 def mark_as_error(search_request_id: int, error_text: str) -> None:
     query = """UPDATE search_requests
                 SET
-                status = 'ERROR',
+                status = 'failed',
                 finished_at = NOW(),
                 error_text = %s
                 WHERE id = %s"""
@@ -74,7 +74,7 @@ def one_request() -> bool:
         logger.info(f"pushed {amount_of_articles} articles")
     except Exception as e:
         logger.exception(f"Pipeline for {user_id} on {search_request_id} by {keyword} failed: {e}")
-        mark_as_error(search_request_id, e)
+        mark_as_error(search_request_id, str(e))
     return True
 
 def run_worker_loop(pull_interval: int = 3) -> None: