Spaces:
Sleeping
Sleeping
| import os | |
| import psycopg2 | |
| from psycopg2.extras import RealDictCursor | |
| import logging | |
| from dotenv import load_dotenv | |
| # Local development ke liye load_dotenv() rahega | |
| # Hugging Face par ye automatically Secrets se environment variable utha lega | |
| load_dotenv() | |
| logger = logging.getLogger(__name__) | |
| # Hugging Face Settings -> Secrets mein 'POSTGRES_URI' naam se link save karein | |
| POSTGRES_URI = os.getenv("POSTGRES_URI", "") | |
| def get_db_connection(): | |
| """ | |
| Establishes a connection to the PostgreSQL database using the URI. | |
| """ | |
| if not POSTGRES_URI: | |
| logger.error("POSTGRES_URI is not set. Please add it to Hugging Face Secrets.") | |
| return None | |
| try: | |
| # Hugging Face se connection ke liye kabhi kabhi sslmode='require' zaroori hota hai | |
| conn = psycopg2.connect(POSTGRES_URI, cursor_factory=RealDictCursor) | |
| return conn | |
| except Exception as e: | |
| logger.error(f"Failed to connect to DB: {e}") | |
| return None | |
| def execute_query(query, params=None): | |
| """ | |
| Helper to execute a query and return results. | |
| """ | |
| conn = get_db_connection() | |
| if not conn: | |
| return [] | |
| try: | |
| with conn.cursor() as cur: | |
| cur.execute(query, params) | |
| if cur.description: # SELECT query ke liye | |
| return cur.fetchall() | |
| conn.commit() # INSERT/UPDATE ke liye | |
| return [] | |
| except Exception as e: | |
| logger.error(f"Query execution failed: {e}") | |
| return [] | |
| finally: | |
| conn.close() |