Spaces:
Sleeping
Sleeping
File size: 1,698 Bytes
601d457 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 |
import os
import psycopg2
from psycopg2.extras import RealDictCursor
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Database Configuration
DB_HOST = os.getenv("DB_HOST")
DB_NAME = os.getenv("DB_NAME")
DB_USER = os.getenv("DB_USER")
DB_PASSWORD = os.getenv("DB_PASSWORD")
DB_PORT = os.getenv("DB_PORT")
def connect_db():
"""Connects to PostgreSQL database."""
try:
conn = psycopg2.connect(
host=DB_HOST,
database=DB_NAME,
user=DB_USER,
password=DB_PASSWORD,
port=DB_PORT,
cursor_factory=RealDictCursor
)
return conn
except Exception as e:
print("DB conn error")
return f"⚠️ Database connection error: {str(e)}"
def execute_sql_query(sql_query: str):
"""
Executes a SQL query on the Supabase PostgreSQL database.
Args:
sql_query (str): The SQL query to execute.
Returns:
list or str: Query results or an error message.
"""
conn = connect_db()
print(conn)
if isinstance(conn, str): # Error message from connect_db
return conn
try:
with conn.cursor() as cur:
print(f"Executing SQL Query: {sql_query}") # Debugging
cur.execute(sql_query)
result = cur.fetchall()
print(f"Query Result: {result}") # Debugging
conn.close()
return result if result else "🔍 No data found."
except Exception as e:
conn.close()
return f"⚠️ Database error for query '{sql_query}': {str(e)}"
def download_spotify_songs(tracks: list[str]):
print(tracks)
return "Songs Downloading..."
|