Spaces:
Runtime error
Runtime error
import psycopg2 | |
from decouple import config | |
from urllib.parse import urlparse | |
import re | |
import sys | |
import shutil | |
import threading | |
import time | |
import os | |
import time | |
from datetime import datetime, timedelta, datetime | |
from colorama import Fore, Style, Back, init | |
init(autoreset=True) | |
def db_connection(): | |
"""Establishes a connection to the database with SSL and keepalives.""" | |
# Parse DATABASE_URL | |
url = urlparse(config("DATABASE_URL")) | |
# Extract connection details from the DATABASE_URL | |
host = url.hostname | |
port = url.port | |
user = url.username | |
password = url.password | |
database = url.path[1:] | |
try: | |
# Connect using individual parameters | |
conn = psycopg2.connect( | |
host=host, | |
port=port, | |
user=user, | |
password=password, | |
dbname=database, | |
sslmode="require", # Use SSL | |
keepalives=1, # Enable keepalives | |
keepalives_idle=300, # Set idle time before keepalives (in seconds) | |
keepalives_interval=60, # Interval for keepalives (in seconds) | |
keepalives_count=5 # Number of keepalive retries before disconnecting | |
) | |
return conn | |
except psycopg2.OperationalError as db_error: | |
return | |
except Exception as e: | |
print(f"Database connection error: {e}"), 500 | |
return conn |