Add func to load to db as sqlite

Author: CoolCoderCarl

app/truth_seeker.py

@@ -3,14 +3,16 @@
 
 from newsapi import NewsApiClient
 
+import databaseconnection
 import dynaconfig
 
 API_KEY = dynaconfig.settings["API_KEY"]
+QUERY = dynaconfig.settings["QUERY"]
 
-newsapi = NewsApiClient(api_key=API_KEY)
-
+# Use to get news from yesterday to current time
 YESTERDAY = date.today() - timedelta(days=1)
-QUERY = dynaconfig.settings["QUERY"]
+
+newsapi = NewsApiClient(api_key=API_KEY)
 
 # Logging
 logging.basicConfig(
@@ -42,20 +44,39 @@ def fetch_info() -> dict:
 def list_info(fetch_info: dict):
     if fetch_info:
         for article in fetch_info.get("articles"):
-            for article_point, article_data in article.items():
-                if article_point == "urlToImage":
+            for article_key, article_data in article.items():
+                if article_key == "urlToImage":
                     continue
-                elif article_point == "source":
-                    print(article_point.capitalize(), article_data.get("name"))
+                elif article_key == "source":
+                    print(article_key.capitalize(), article_data.get("name"))
                 else:
-                    print(article_point.capitalize(), article_data)
+                    print(article_key.capitalize(), article_data)
     else:
         logging.warning("Empty response from API")
         raise IndexError
 
 
+def load_to_db():
+    for article in fetch_info().get("articles"):
+        article_list = []
+        for article_key, article_data in article.items():
+            if article_key == "source":
+                pass
+            elif article_key == "content":
+                pass
+            elif article_key == "urlToImage":
+                pass
+            else:
+                article_list.append(article_data)
+        article_list.append("True")
+        databaseconnection.insert_into(
+            databaseconnection.create_connection(databaseconnection.DB_FILE),
+            tuple(article_list),
+        )
+
+
 if __name__ == "__main__":
     if dynaconfig.settings["LOAD_TO_DB"]:
-        pass
+        load_to_db()
     else:
         list_info(fetch_info())

databaseconnection.py

@@ -0,0 +1,73 @@
+import logging
+import sqlite3
+from pathlib import Path
+from sqlite3 import Error
+
+DB_FILE = Path("test.db")
+
+CREATE_TABLE_SQL = """
+CREATE TABLE IF NOT EXISTS news (
+news_id INTEGER PRIMARY KEY,
+author TEXT,
+title TEXT,
+description TEXT,
+url TEXT,
+pub_date TEXT,
+used_bool TEXT
+);
+"""
+
+INSERT_INTO_SQL = """
+INSERT INTO news
+(author,title,description,url,pub_date,used_bool)
+VALUES(?,?,?,?,?,?)
+"""
+
+
+# Logging
+logging.basicConfig(
+    format="%(asctime)s - %(levelname)s - %(message)s", level=logging.INFO
+)
+logging.basicConfig(
+    format="%(asctime)s - %(levelname)s - %(message)s", level=logging.ERROR
+)
+
+
+def create_connection(db_file: Path):
+    conn = None
+    try:
+        conn = sqlite3.connect(db_file)
+        logging.info("Database created successfully !")
+        return conn
+    except Error as create_conn_err:
+        logging.error(create_conn_err)
+    return conn
+
+
+def create_table(conn, create_table_query):
+    try:
+        c = conn.cursor()
+        c.execute(create_table_query)
+        logging.info("Table created successfully !")
+    except Error as create_table_err:
+        logging.error(create_table_err)
+
+
+def insert_into(conn, data: tuple):
+    try:
+        cur = conn.cursor()
+        cur.execute(INSERT_INTO_SQL, data)
+        conn.commit()
+
+        logging.info("Data inserted successfully !")
+        return cur.lastrowid
+    except Error as insert_err:
+        logging.error(insert_err)
+
+
+if __name__ == "__main__":
+    conn = create_connection(DB_FILE)
+    if conn is not None:
+        create_table(conn, CREATE_TABLE_SQL)
+    else:
+        logging.error("Error! Cannot create the database connection.")