Merge pull request #34 from acmcsufoss/supabase-migration

Migrate off of ElephantSQL to Supabase

.env.example

@@ -1,6 +1,7 @@
+SUPABASE_URL=""
+SUPABASE_KEY=""
 LINKEDIN_URL=""
 DISCORD_WEBHOOK=""
-DB_URI=""
 DB_TABLE="opportunities_table"
 PALM_API_KEY=""
 GH_INTERN24_URL="https://github.com/pittcsc/Summer2024-Internships"

.github/workflows/automate.yaml

@@ -42,10 +42,11 @@ jobs:
 # fmt: off    
 #LINT.IF .env.example
         env:
+          SUPABASE_URL: ${{ secrets.SUPABASE_URL }}
+          SUPABASE_KEY: ${{ secrets.SUPABASE_KEY }}
+          DB_TABLE_NAME: ${{ secrets.DB_TABLE_NAME}}
           LINKEDIN_URL: ${{ secrets.LINKEDIN_URL}}
           DISCORD_WEBHOOK: ${{ secrets.DISCORD_WEBHOOK}}
-          DB_URI: ${{ secrets.DB_URI}}
-          DB_TABLE: ${{ secrets.DB_TABLE}}
           PALM_API_KEY: ${{ secrets.PALM_API_KEY}}
           LINKEDIN_INTERN_URL: ${{ secrets.LINKEDIN_INTERN_URL}}
           GH_INTERN24_URL: ${{ secrets.GH_INTERN24_URL}}

main.py

@@ -13,6 +13,7 @@
     request_linkedin_internship24_data,
 )
 from utility.palm import gpt_job_analyze
+from utility.error import ErrorMsg
 
 
 # Load and determine if all env variables are set
@@ -70,7 +71,9 @@ async def execute_opportunities_webhook(
     if response.status_code == 204:
         print("Webhook message was sent sucessfully!")
     else:
-        print(f"Failed to send webhook message. Status Code: {response.status_code}")
+        print(
+            f"Failed to send webhook message. {ErrorMsg().status_code_failure(response.status_code)}"
+        )
 
 
 async def main():
@@ -102,6 +105,7 @@ async def main():
         job_opps,
         prompt_object["full_time"],
     )
+
     opps.ingest_opportunities(filtered_job_opps)
 
     # Consolidates all job-related opportunities into a comprehensive List[Opportunity], eliminating repetitive calls to the LLM SERVER.
@@ -123,7 +127,9 @@ async def main():
 
     # db.reset_processed_status()
 
-    internship_data_results = opps.list_opportunities(True, "internship", filtered=True)
+    internship_data_results = opps.list_opportunities(
+        False, "internship", filtered=True
+    )
     job_data_results = opps.list_opportunities(True, "full_time", filtered=True)
 
     internship_formatted_message = opps.format_opportunities(

utility/db.py

@@ -1,66 +1,98 @@
-import psycopg2
 import os
+from supabase import create_client
+from dataclasses import dataclass
+from dotenv import load_dotenv
+import requests
+from utility.error import ErrorMsg
 
+load_dotenv()
+ERROR_MSG = ErrorMsg()
 
-def instantiate_db_connection():
-    """Returns the connection from the DB"""
 
-    db_uri = os.getenv("DB_URI")
-    return psycopg2.connect(db_uri)
+@dataclass
+class SupabaseConnection:
+    """Constructs Supabase connection."""
 
+    CLIENT: any = None
+    SUPABASE_URL: str = os.getenv("SUPABASE_URL")
+    SUPABASE_KEY: str = os.getenv("SUPABASE_KEY")
+    SUPABASE_API_URL: str = f"{SUPABASE_URL}/rest/v1/rpc"
 
-def create(TABLE_NAME: str) -> None:
-    """Creates the DB. Only needs to be called once."""
+    def __post_init__(self):
+        self.CLIENT = self.create_supabase_client()
 
-    with instantiate_db_connection() as connection:
-        cursor = connection.cursor()
+    def create_supabase_client(self):
+        """Creates the Supabase client with the URL and key variables."""
 
-        cursor.execute(
-            f"""CREATE TABLE IF NOT EXISTS {TABLE_NAME}(company TEXT, title TEXT, location TEXT, link TEXT, processed INTEGER DEFAULT 0)"""
-        )
+        return create_client(self.SUPABASE_URL, self.SUPABASE_KEY)
 
-        connection.commit()
 
+SUPABASE_INSTANCE = SupabaseConnection().CLIENT
 
-def add_column(column_name: str, data_type: str) -> None:
-    """Adds a column for adjustment to the table after the table has been created"""
 
-    with instantiate_db_connection() as connection:
-        cursor = connection.cursor()
+def delete_all_opportunity_type(TABLE_NAME: str, opp_type: str) -> None:
+    """Deletes all opportunities of a specific type for testing purposes only."""
+
+    SUPABASE_INSTANCE.table(TABLE_NAME).delete().eq("type", opp_type).execute()
+
 
-        cursor.execute(f"ALTER TABLE jobs_table ADD COLUMN {column_name} {data_type}")
+def reset_processed_status(TABLE_NAME: str) -> None:
+    """Jobs status will be set to _processed = 0 for testing a debugging purposes"""
 
-        connection.commit()
+    SUPABASE_INSTANCE.table(TABLE_NAME).update({"processed": 0}).eq(
+        "processed", 1
+    ).limit(5).execute()
 
 
-def delete_all_opportunity_type(opp_type: str) -> None:
-    """Deletes all opportunities of a specific type for testing purposes only"""
+def execute_sql(sql: str):
+    """Executes a raw SQL query using the Supabase HTTP API."""
+    connection = SupabaseConnection()
+    headers = {
+        "apikey": connection.SUPABASE_KEY,
+        "Authorization": f"Bearer {connection.SUPABASE_KEY}",
+        "Content-Type": "application/json",
+    }
 
-    with instantiate_db_connection() as connection:
-        cursor = connection.cursor()
+    data = {"query": sql}
+    response = requests.post(connection.SUPABASE_API_URL, headers=headers, json=data)
 
-        cursor.execute("DELETE FROM jobs_table WHERE type = %s", (opp_type,))
-        connection.commit()
+    response.raise_for_status()
+    return response
 
 
-def reset_processed_status(TABLE_NAME: str) -> None:
-    """Jobs status will be set to _processed = 0 for testing a debugging purposes"""
+def create_table(TABLE_NAME: str) -> None:
+    """Creates a new table in Supabase database. Only needs to be called once."""
 
-    with instantiate_db_connection() as connection:
-        cursor = connection.cursor()
+    request = f"""
+    CREATE TABLE IF NOT EXISTS {TABLE_NAME} (
+        company TEXT,
+        title TEXT,
+        link TEXT,
+        processed INTEGER DEFAULT 0,
+        type TEXT
+    );
+    """
+
+    response = execute_sql(request)
+
+    if response.status_code != 200:
+        return ERROR_MSG.status_code_failure(response.status_code)
+
+    return "Request executed successfully."
+
+
+def add_column(column_name: str, data_type: str) -> None:
+    """Adds a column for adjustment to the table after the table has been created"""
 
-        cursor.execute(
-            f"SELECT company, title, location FROM {TABLE_NAME} WHERE processed = 1 LIMIT 5"
-        )
+    TABLE_NAME = os.getenv("DB_TABLE")
 
-        rows = cursor.fetchall()
+    request = f"""
+        ALTER TABLE {TABLE_NAME} ADD COLUMN {column_name} {data_type};
+    """
 
-        for row in rows:
-            company, title, location = row[:3]
+    response = execute_sql(request)
 
-            cursor.execute(
-                f"UPDATE {TABLE_NAME} SET processed = 0 WHERE company = %s AND title = %s AND location = %s",
-                (company, title, location),
-            )
+    if response.status_code != 200:
+        return ERROR_MSG.status_code_failure(response.status_code)
 
-        connection.commit()
+    return "Request executed successfully."

utility/error.py

@@ -0,0 +1,24 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class ErrorMsg:
+    """Custom error message data class."""
+
+    @staticmethod
+    def status_code_failure(status_code: str) -> str:
+        """A function returns an status code that is not 200."""
+
+        return f"Execution failure. Status code returned: {status_code}."
+
+    @staticmethod
+    def date_difference_failure(error: str) -> str:
+        """Calculating the date difference is not possible."""
+
+        return f"Error calculating date difference: {error}."
+
+    @staticmethod
+    def file_open_failure(file_path: str) -> str:
+        """Unable to open file path."""
+
+        return f"Unable to open/read file path: '{file_path}'."