Fix date validation and loop csvs

api/app/validation.py

@@ -44,13 +44,13 @@ def decorated_function(*args, **kwargs):
             if invalid_indicators:
                 return jsonify({"error": f"Invalid indicators: {', '.join(invalid_indicators)}"}), 400
         if date:
-            if not date_in_db(date, Level.NATIONAL):
+            if not date_in_db(date, Level.NATIONAL) and not date_in_db(date, Level.SUBNATIONAL):
                 return jsonify({"error": f"Invalid date {date}"}), 400
         if start_date:
-            if not date_in_db(start_date, Level.NATIONAL):
+            if not date_in_db(start_date, Level.NATIONAL) and not date_in_db(start_date, Level.SUBNATIONAL):
                 return jsonify({"error": f"Invalid start date {start_date}"}), 400
         if end_date:
-            if not date_in_db(end_date, Level.NATIONAL):
+            if not date_in_db(end_date, Level.NATIONAL) and not date_in_db(end_date, Level.SUBNATIONAL):
                 return jsonify({"error": "Invalid end date {end_date}"}), 400
         return f(*args, **kwargs)
 

scripts/helpers.py

@@ -4,7 +4,7 @@
 from io import BytesIO
 from typing import Union
 
-ROOT_URL = "http://localhost:8080/api/v2"
+ROOT_URL = "http://3.11.85.207/api/v2"
 
 
 class Level(Enum):

scripts/post_national.py

@@ -21,6 +21,7 @@
 import dotenv
 import helpers
 from typing import Union
+import pycountry
 
 BASE = Path(__file__).resolve().parent
 
@@ -38,29 +39,36 @@ def post_data(token: str, data: list[dict], level: helpers.Level):
     """Post data using a list of dictionaries representing the data to be posted."""
     url = f"{ROOT_URL}/post_{level.value}_data"
     response = requests.post(url, json=data, headers={"Authorization": f"Bearer {token}"})
-    response.status_code == 200, response.json()
+    assert response.status_code == 200, response.json()
 
 
-def post_data_from_csv(path_to_csv: Union[Path, str], level: helpers.Level):
+def post_data_from_csv(path_to_csv_dir: Union[Path, str], level: helpers.Level):
     """Posts data from a CSV file."""
-    assert Path(path_to_csv).exists(), "CSV File does not exist."
-    df = pd.read_csv(path_to_csv)
-    chunk_size = 1000
-    token = helpers.get_token(username=POST_USERNAME, password=POST_PASSWORD)
-    print("POSTING DATA...")
-    for start in range(0, len(df), chunk_size):
-        end = start + chunk_size
-        chunk = df.iloc[start:end]
-        data = chunk.to_dict(orient="records")
-        post_data(token, data, level)
-        print(f"Posted {start} to {end} records.")
+    assert Path(path_to_csv_dir).exists(), "CSV File does not exist."
+    csv_files = list(Path(path_to_csv_dir).rglob("*.csv"))
+    for path_to_csv in csv_files:
+        df = pd.read_csv(path_to_csv)
+        if "country" not in df.columns:
+            df["country"] = df.apply(lambda x: pycountry.countries.get(alpha_3=x["gid_0"]).name, axis=1)
+        chunk_size = 1000
+        token = helpers.get_token(username=POST_USERNAME, password=POST_PASSWORD)
+        print("POSTING DATA...")
+        for start in range(0, len(df), chunk_size):
+            end = start + chunk_size
+            chunk = df.iloc[start:end]
+            data = chunk.to_dict(orient="records")
+            post_data(token, data, level)
+            print(f"Posted {start} to {end} records.")
+        print(f"POSTED DATA FROM {path_to_csv.name}.")
     print("POSTED DATA.")
 
 
 if __name__ == "__main__":
-    CSV = BASE.joinpath("test_post_delete_national.csv")
+    #CSV = BASE.joinpath("test_post_delete_national.csv")
+    #CSV_DIR = BASE.joinpath("dgg_data_national/dgg_data_national")
+    CSV_DIR = BASE.joinpath("dgg_test_upload")
     from datetime import datetime
     start = datetime.now()
-    post_data_from_csv(CSV, helpers.Level.NATIONAL)
+    post_data_from_csv(CSV_DIR, helpers.Level.NATIONAL)
     end = datetime.now()
     print(f"Time taken to post data: {end-start}")

scripts/requirements.txt

@@ -0,0 +1 @@
+pycountry==24.6.1