Continued Groundwork for GoEx APIs

.gitignore

@@ -10,12 +10,14 @@ dist
 .editorconfig
 .DS_Store
 **/*.pyc
+goex/goex-env/
 goex/exec_engine/checkpoints/
 goex/exec_engine/credentials/*
 !goex/exec_engine/credentials/credentials_utils.py
 !goex/exec_engine/credentials/supported.txt
 goex/docker/*/requirements.txt
 goex/docker/misc/images.json
+goex-env/*
 
 ################## Berkley Function Call Leaderboard ##########################
 

goex/authorizations/scripts/spotify_authorization.py

@@ -10,8 +10,8 @@
 import wsgiref.simple_server
 import wsgiref.util
 from urllib.parse import quote, urlparse, parse_qs
-from .consts import AUTH_URL, CERT_FILE_PATH, KEY_FILE_PATH
-
+from consts import AUTH_URL, CERT_FILE_PATH, KEY_FILE_PATH
+print(CERT_FILE_PATH)
 SPOTIFY_FOLDER_PATH = os.path.join(os.path.dirname(Path(os.path.realpath(__file__)).parent.parent), 'exec_engine/credentials/spotify')
 TOKEN_FILE = os.path.join(SPOTIFY_FOLDER_PATH, 'token.pickle')
 
@@ -23,6 +23,7 @@ def main():
             creds = pickle.load(token)
     # If there are no (valid) credentials available, let the user log in
     creds = run_spotify_flow()
+    print(creds)
     # Save the credentials for the next run
     if not os.path.isdir(SPOTIFY_FOLDER_PATH):
         os.mkdir(SPOTIFY_FOLDER_PATH)

goex/cli.py

@@ -10,7 +10,7 @@
 from main import ExecutionEngine, PythonAPIExecutor
 
 from exec_engine.utils import SQL_Type, Filesystem_Type
-from exec_engine.db_manager import MySQLManager, SQLiteManager
+from exec_engine.db_manager import MySQLManager, SQLiteManager, PostgreSQLManager
 from dotenv import load_dotenv
 import questionary
 
@@ -213,6 +213,8 @@ def db_callback(prompt, generate_mode):
             db_manager = MySQLManager(config, docker_sandbox=engine.docker_sandbox)
         elif db_type == 'sqlite':
             db_manager = SQLiteManager(config, docker_sandbox=engine.docker_sandbox)
+        elif db_type == 'postgresql':
+            db_manager = PostgreSQLManager(config, docker_sandbox=engine.docker_sandbox)
     except Exception as e:
         print(f"Error during {db_type} Manager Init: {e}")
         return

goex/exec_engine/db_manager.py

@@ -300,3 +300,120 @@ def close(self):
         if self.conn:
             self.cursor.close()
             self.conn.close()
+
+class PostgreSQLManager(DBManager):
+    """PostgreSQL database manager.
+
+    Attributes:
+        _postgresql_imported (bool): flag to check if postgresql is imported.
+
+    Methods:
+        connect: Establish connections to the DB
+        execute_db_call: Execute SQL call
+        commit_db_calls: Commit SQL calls
+        rollback_db_calls: Rollback SQL calls
+        close: Close the connection to the database
+    """
+    _postgresql_imported = False
+    db_type = "postgresql"
+    TEST_CONFIG = "{'host': '127.0.0.1', 'user': 'root', 'password': ''}\n Use psycopg2 and make sure to create the database using subprocess before connection."
+    def __init__(self, connection_config, docker_sandbox: DockerSandbox = None):
+        """Initialize the PostgreSQLManager.
+
+        Args:
+            connection_config (dict): configuration for the database connection, including keys for 'user', 'password', 'host', and 'database'.
+        """
+        if not PostgreSQLManager._postgresql_imported:
+            global psycopg2
+            import psycopg2
+            PostgreSQLManager._postgresql_imported = True
+
+        keys = connection_config.keys()
+
+        if any(key not in keys for key in ['host', 'user', 'password', 'database']):
+            raise ValueError("Failed to initialize PostgreSQL Manager due to bad configs")
+        elif any([not connection_config['host'], not connection_config['user'], not connection_config['password'], not connection_config['database']]):
+            raise ValueError("Failed to initialize PostgreSQL Manager due to missing configs")
+
+        self.connection_config = {
+            'dbname': connection_config['database'] if 'database' in connection_config else 'postgres',
+            'user': connection_config['user'] if 'user' in connection_config else 'postgres',
+            'password': connection_config['password'] if 'password' in connection_config else '',
+            'host': connection_config['host'] if 'host' in connection_config else '127.0.0.1'
+        }
+
+    def connect(self):
+        """Establish connection to the MySQL database and create a cursor."""
+        connection = None
+        try:
+            connection = psycopg2.connect(**self.connection_config)
+            self.conn = connection
+            self.cursor = connection.cursor()
+            self.update_schema_info()
+        except Exception as e:
+            if connection:
+                connection.close()
+            print("Failed to connect to the database. Error:", e)
+
+    def update_schema_info(self):
+        schema_info = {}
+        get_all_tables_query = """
+        SELECT table_name
+        FROM information_schema.tables
+        WHERE table_schema = 'public'
+        """
+        self.cursor.execute(get_all_tables_query)
+        tables = self.cursor.fetchall()
+        for (table_name,) in tables:
+            self.cursor.execute(f"SELECT column_name, data_type, is_nullable, column_default FROM information_schema.columns WHERE table_name = '{table_name}';")
+            schema_info[table_name] = self.cursor.fetchall()
+
+        self.schema = schema_info
+
+    def execute_db_call(self, call):
+        """Execute a SQL call using the cursor."""
+        if not self.conn:
+            self.connect()
+        try:
+            self.cursor.execute(call)
+            self.update_schema_info()
+            return 0
+        except Exception as e:
+            return 1
+
+    def fetch_db_call(self, call: str) -> list[dict]:
+        """Execute a SQL call and return the results.
+
+        Args:
+            call (str): SQL query to execute.
+
+        Returns:
+            list[dict]: A list of dictionaries representing each row in the query result.
+        """
+        if not self.conn:
+            self.connect()
+        try:
+            self.cursor.execute(call)
+            ret_val = self.cursor.fetchall()
+            self.update_schema_info()
+            return ret_val
+        except Exception as e:
+            return []
+
+    def commit_db_calls(self):
+        """Commit SQL calls."""
+        if not self.conn:
+            self.connect()
+        self.conn.commit()
+
+    def rollback_db_calls(self):
+        """Rollback SQL calls not committed."""
+        if not self.conn:
+            self.connect()
+        self.conn.rollback()
+
+    def close(self):
+        """Close the cursor and the connection to the database."""
+        if self.conn:
+            self.cursor.close()
+            self.conn.close()

goex/function/spotify_api_details.py

@@ -0,0 +1,14 @@
+import spotipy
+from spotipy.oauth2 import SpotifyOAuth
+
+def create_spotify_api_object():
+    scope = "user-library-read"
+
+    client_id = 'd711dfc0d97440cb898b08fbdc2083c1' 
+    client_secret = '1d4854768be047438a72d48a8c4622ab'
+    redirect_uri = 'http://localhost:8888/callback'
+
+    sp = spotipy.Spotify(auth_manager=SpotifyOAuth(client_id = client_id,
+                                                client_secret = client_secret,
+                                                redirect_uri = redirect_uri,
+                                                scope=scope))

goex/function/spotify_create_playlist.py

@@ -0,0 +1,25 @@
+import os
+import pickle
+import spotipy
+from spotipy.oauth2 import SpotifyOAuth
+from spotipy.oauth2 import SpotifyClientCredentials
+
+def spotify_play_song(user,name):
+    # Load spotify credentials
+    credentials_path = './credentials/spotify/token.pickle'
+    if os.path.exists(credentials_path):
+        with open(credentials_path, 'rb') as token_file:
+            spotify_token = pickle.load(token_file)
+    else:
+        raise FileNotFoundError("Spotify token file not found.")
+    token_info = SpotifyOAuth(token=spotify_token)
+    # Initialization 
+    if not token_info:
+        print("No account found")
+        return None
+    else:
+        sp = spotipy.Spotify(client_credentials_manager=SpotifyClientCredentials())
+    if sp is None:
+        return None
+    # create playlist with inputted name and user ID
+    sp.user_playlist_create(user, name, public=True, collaborative=False, description='')

goex/function/spotify_list_artist_albums.py

@@ -0,0 +1,43 @@
+import os
+import pickle
+import spotipy
+from spotipy.oauth2 import SpotifyOAuth
+from spotipy.oauth2 import SpotifyClientCredentials
+
+def spotify_list_artist_albums(artist):
+    # Load spotify credentials
+    credentials_path = './credentials/spotify/token.pickle'
+    if os.path.exists(credentials_path):
+        with open(credentials_path, 'rb') as token_file:
+            spotify_token = pickle.load(token_file)
+    else:
+        raise FileNotFoundError("Spotify token file not found.")
+    token_info = SpotifyOAuth(token=spotify_token)
+    # Initialization 
+    if not token_info:
+        print("No account found")
+        return None
+    else:
+        sp = spotipy.Spotify(client_credentials_manager=SpotifyClientCredentials())
+    if sp is None:
+        return None
+    # Returns the most popular artist's URI with the same name 
+    results = sp.search(q=artist, type='artist', limit=1)
+    artista = results['artist']['items']
+    if artista:
+         artistaa = artista[0]['uri'] 
+    else:
+        print("No artist found for:", artist)
+        return None
+    # Uses the artist URI to search through their spotify discography and returns the names of all their albums
+
+    spotify = spotipy.Spotify(client_credentials_manager=SpotifyClientCredentials())
+
+    albums_results = spotify.artist_albums(artistaa, album_type='album')
+    albums = albums_results['items']
+    while albums_results['next']:
+        albums_results = spotify.next(albums_results)
+        albums.extend(albums_results['items'])
+
+    for album in albums:
+        print(album['name'])

goex/function/spotify_pause.py

@@ -0,0 +1,29 @@
+import pyautogui
+import os
+import pickle
+import spotipy
+from spotipy.oauth2 import SpotifyOAuth
+from spotipy.oauth2 import SpotifyClientCredentials
+
+def spotify_pause():
+    # Load spotify credentials
+    credentials_path = './credentials/spotify/token.pickle'
+    if os.path.exists(credentials_path):
+        with open(credentials_path, 'rb') as token_file:
+            spotify_token = pickle.load(token_file)
+    else:
+        raise FileNotFoundError("Spotify token file not found.")
+    token_info = SpotifyOAuth(token=spotify_token)
+    # Initialization 
+    if not token_info:
+        print("No account found")
+        return None
+    else:
+        sp = spotipy.Spotify(client_credentials_manager=SpotifyClientCredentials())
+    if sp is None:
+        return None
+    # Check to see if a song is currently playing
+    if sp.currently_playing:
+        pyautogui.press('stop')
+    else:
+        print("No song currently playing")

goex/function/spotify_play_album.py

@@ -0,0 +1,38 @@
+import os
+import pickle
+import spotipy
+from spotipy.oauth2 import SpotifyOAuth
+from spotipy.oauth2 import SpotifyClientCredentials
+
+def spotify_play_album(album_name):
+    # Load spotify credentials
+    credentials_path = './credentials/spotify/token.pickle'
+    if os.path.exists(credentials_path):
+        with open(credentials_path, 'rb') as token_file:
+            spotify_token = pickle.load(token_file)
+    else:
+        raise FileNotFoundError("Spotify token file not found.")
+    token_info = SpotifyOAuth(token=spotify_token)
+    # Initialization 
+    if not token_info:
+        print("No account found")
+        return None
+    else:
+        sp = spotipy.Spotify(client_credentials_manager=SpotifyClientCredentials())
+    if sp is None:
+        return None
+    # Searches the album name and finds the most popular match and then provides the URI from that
+    results = sp.search(q=album_name, type='album', limit=1)
+    albums = results['album']['items']
+    if albums:
+         album = albums[0]['uri'] 
+    else:
+        print("No tracks found for:", album_name)
+        return None
+    # Uses the URI to play the album on Spotify
+    try:
+        for track in album:
+            sp.start_playback(uris=[track])
+            print("Playing")
+    except spotipy.SpotifyException as e:
+        print("Error", e)

goex/function/spotify_play_song.py

@@ -0,0 +1,38 @@
+import os
+import pickle
+import spotipy
+from spotipy.oauth2 import SpotifyOAuth
+from spotipy.oauth2 import SpotifyClientCredentials
+
+def spotify_play_song(song_name):
+    # Load spotify credentials
+    credentials_path = './credentials/spotify/token.pickle'
+    if os.path.exists(credentials_path):
+        with open(credentials_path, 'rb') as token_file:
+            spotify_token = pickle.load(token_file)
+    else:
+        raise FileNotFoundError("Spotify token file not found.")
+    token_info = SpotifyOAuth(token=spotify_token)
+    # Initialization 
+    if not token_info:
+        print("No account found")
+        return None
+    else:
+        sp = spotipy.Spotify(client_credentials_manager=SpotifyClientCredentials())
+    if sp is None:
+        return None
+    # Get track uri from the song name
+    # '''searches song name and finds the most popular match and then provides the URI from that'''
+    results = sp.search(q=song_name, type='track', limit=1)
+    tracks = results['tracks']['items']
+    if tracks:
+         song = tracks[0]['uri'] 
+    else:
+        print("No tracks found for:", song_name)
+        return None
+    # Uses the URI to play the song on Spotify
+    try:
+        sp.start_playback(uris=[song])
+        print("Playing")
+    except spotipy.SpotifyException as e:
+        print("Error", e)